TARIFF-01: measure the month's billed peak, and make it survive a restart

The 15-minute block average was already built and unwired. p1.QuarterAverager
computes clock-aligned 900 s blocks, time-weighted over irregular telegram
spacing, offtake-only and per-phase; P1Ingest.blocks was a 96-entry list in
memory that a restart erased. So this is the persistence and the monthly
reduction, not a second accumulator - a second one would be a second source of
truth for the exact quantity the money depends on.

BLOCKS, NOT A SLIDING WINDOW. The card says "rolling"; 9.1 bills "the highest
15-minute average offtake" and states its own worked case per QUARTER - one
4998 W quarter moving a monthly peak from 4832 to 5013 W. A trailing window
reports a peak the meter never bills, and reports it HIGH: a burst straddling
:15 fills the window completely while contributing to two block averages at
partial weight.

DELIVERED, NOT INTENDED. tariff.py only ever sees P1Sample - the meter's own
reading - and never a setpoint. SAFETY-20's open question does not arise for a
capacity tariff: the utility integrates the current that flowed.

AN UNREADABLE STORE IS NOT A ZERO PEAK. peak() answers OBSERVED / NO_RECORD /
NOT_OBSERVED with peak_w None unless observed, which is riglock.holder()'s fix
applied to something that spends money. A peak read as 0 W hands SAFETY-07
maximum headroom and disarms the guard at the moment its own storage breaks.

Restart mid-block resumes the accumulator through a new snapshot()/restore()
pair, checkpointed every 15 s and cleared in the same breath as a block close.
The held sample keeps its ORIGINAL timestamp, so an outage is integrated under
the same max_hold_s rule a running process would apply. quarter_start is the
primary key and a re-bank may only ever RAISE a figure.

Calendar months are LOCAL: a quarter starting 22:15 UTC on 31 August bills in
September.

No new option (the limit is the existing peak_cap_w), so config.yaml and
validate.py are untouched and CFG-03 is not widened. No second peak entity:
peak_forecast_entity stays the input to maintenance_charge_floor(), because
repointing that consumer changes what moves energy and belongs to SAFETY-07.

AC 3 is NOT met. It asks for 30 days of replay; two days of this house's real
meter exist and are committed here. test_tariff.py part 9 replays them, reports
the percentage next to the sample size in days, and asserts nothing about 95 %.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,59 @@
 
 ## Unreleased
 
+**The capacity-tariff peak is now measured, banked and survives a restart
+(TARIFF-01).** New `app/tariff.py`. The 15-minute block average itself was
+already built and unwired - `p1.QuarterAverager` computes clock-aligned 900 s
+blocks, time-weighted over irregular telegram spacing, offtake-only and
+per-phase - and `P1Ingest.blocks` was a 96-entry list a restart erased. This
+adds the SQLite behind it (`/data/tariff.db`): one row per completed quarter in
+the indefinite 15-min tier, one row per calendar month in the permanent peak
+tier, and no `DELETE` against either anywhere in the module.
+
+**Blocks, not a sliding window.** The card says "rolling"; 9.1 bills "the
+highest 15-minute average offtake", stated per QUARTER. "Rolling" here means
+continuously updated, not a trailing-15-minutes window - a trailing window
+reports a peak the meter never bills, and reports it HIGH, because a burst
+straddling :15 fills it completely while contributing to two block averages at
+partial weight.
+
+**A restart mid-block resumes it, and a restart that outlives the block banks
+it.** `QuarterAverager` grew `snapshot()`/`restore()`; the tracker checkpoints
+the open block every 15 s (worst case 1.7 % of one quarter) and clears the
+checkpoint in the same breath as a block close. The held sample keeps its
+ORIGINAL timestamp across the restart, so the outage is integrated under the
+same `max_hold_s` rule a running process would apply and uncovered seconds drag
+the closed block DOWN rather than being held forward. `quarter_start` is the
+primary key and a re-bank may only ever RAISE a figure, so a replay cannot
+double-count and a partial re-run cannot lower a quarter already measured.
+
+**An unreadable store is not a zero peak.** `peak()` answers `OBSERVED` /
+`NO_RECORD` / `NOT_OBSERVED` and carries `peak_w = None` unless observed - the
+`riglock.holder()` fix applied to something that spends money. A peak read as
+0 W hands SAFETY-07 maximum headroom and silently disarms the guard at the
+moment its own storage breaks, so there is no accessor that defaults it: a
+caller who forgets gets a `TypeError`, and the status page prints a different
+row for "no quarter yet" than for "the database did not answer".
+
+**Calendar months are LOCAL, blocks are UTC.** A quarter starting 22:15 UTC on
+31 August bills in September. Block alignment still needs no timezone.
+
+**No new option, and no second peak entity.** The soft limit is the existing
+`peak_cap_w`; `config.yaml` and `validate.py` are untouched, so CFG-03 is not
+widened. `peak_forecast_entity` stays the input to `maintenance_charge_floor()`
+- repointing that consumer changes what moves energy and belongs to SAFETY-07.
+
+**It measures and nothing else.** No claim, no clamp, no command; graded on the
+parsed syntax in `test_tariff.py`, which also proves the same grader lights up
+on `main.py`. Per SAFETY-20, the quantity is DELIVERED power - `tariff.py` only
+ever sees `P1Sample`, the meter's own reading, never a setpoint.
+
+**AC 3 is not met and the check says so.** It asks for 30 days of replay; two
+days of this house's real meter exist (2026-08-20, 2026-08-23), now committed as
+`scenarios/p1-net-*.jsonl.gz`. `test_tariff.py` part 9 replays them, reports
+the percentage under the shipped limit next to the sample size in days, and
+asserts nothing about the 95 % figure.
+
 **A computed setpoint over the rating now alarms instead of being clipped in
 silence (SAFETY-03).** The clamp was never the missing half - specs.md 10.3
 records an unclamped integrator climbing ~130 W per 4 s past 10 kW and reporting

goodwe_controller/app/main.py

@@ -45,6 +45,7 @@ from .hass import HomeAssistant
 from .maintenance import IDLE, MaintConfig, Maintenance
 from .mqtt import MqttPublisher
 from .p1 import P1Ingest, build_source, is_enabled
+from .tariff import PeakStore, PeakTracker
 from .validate import ConfigRefused, refuse, validate_startup_config
 from .watchdog import SignWatchdog, SignWatchdogConfig, loop_is_following
 from . import web
@@ -63,7 +64,8 @@ def load_options() -> dict:
 
 
 class Controller:
-    def __init__(self, opts: dict, hass: HomeAssistant, store, mqtt_pub, *, stop):
+    def __init__(self, opts: dict, hass: HomeAssistant, store, mqtt_pub, *, stop,
+                 peak_db_path: str = "/data/tariff.db"):
         self.o = opts
         self.hass = hass
         self.store = store
@@ -163,6 +165,29 @@ class Controller:
         self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
                            max_age_s=float(opts.get("meter_max_age_s", 30)))
         self.p1_enabled = is_enabled(opts)
+
+        # TARIFF-01. The capacity-tariff peak: measured from the meter, banked
+        # in SQLite, restored across a restart. It is an INSTRUMENT - it files no
+        # claim and clamps nothing. SAFETY-07 is what will read it.
+        #
+        # ⚠️ NO NEW OPTION. The soft limit is `peak_cap_w`, already shipped,
+        # already validated, already read by control.peak_at_risk(). CFG-03
+        # tracks three places option defaults live; a fourth, for a limit that
+        # already exists, is the mistake the clamp watchdog declined to make ~30
+        # lines up. config.yaml and validate.py are untouched by this ticket.
+        #
+        # ⚠️ AND NO SECOND PEAK ENTITY. `peak_forecast_entity` is the existing
+        # HA-side delegation and it stays the input to maintenance_charge_floor():
+        # repointing that consumer at this measurement CHANGES WHAT MOVES ENERGY
+        # during a maintenance charge, which belongs to SAFETY-07 with its own
+        # acceptance, not to a measurement ticket. What this adds is the MEASURED
+        # MONTH PEAK - a different quantity from a forecast entity, published
+        # nowhere as a duplicate of it.
+        self.peak_store = PeakStore(peak_db_path)
+        self.peak = PeakTracker(self.peak_store,
+                                cap_w=float(opts.get("peak_cap_w", 3500)))
+        self.peak.restore(self.p1.averager)
+        _LOG.info("capacity-tariff peak store: %s", self.peak.restore_state)
         # Set by amain() once the transport is built, so the status page can show
         # what only the transport knows (homewizard_local's unchanged_s). Stays
         # None when P1 is off, or under a transport that has no such counter.
@@ -472,15 +497,42 @@ class Controller:
         self.reason = resolution.explain()
         await self.write_setpoint(self.target)
 
-        # --- SAFETY-18, and it runs LAST on purpose --------------------------
-        # Both terms of the conjunction describe the state that now exists: the
-        # command that has just gone on the wire, and how long the meter that
-        # chose it has been standing still. Evaluated before write_setpoint()
-        # it would be reasoning about the PREVIOUS command against THIS cycle's
-        # meter - the one combination that is true of neither moment.
+        # --- SAFETY-03, and its boundary is the WRITE ------------------------
+        # ⚠️ NOT THE SAME BOUNDARY AS SAFETY-18's BELOW, THOUGH THE CALLS ABUT.
+        # This one reads `self.clamp_last`, which is None until write_setpoint()
+        # has actually run, so it has to follow the write. The one below grades
+        # `self.target`, which is assigned further up - see check_output_clamp()
+        # and check_meter_frozen() for each. Two guards, two orderings, adjacent
+        # lines: do not "tidy" them into one rule.
         await self.check_output_clamp()
+
+        # --- SAFETY-18, and its boundary is the ASSIGNMENT --------------------
+        # ⚠️ THE BOUNDARY IS `self.target = resolution.target_w` ABOVE, NOT
+        # write_setpoint(). This comment used to say "the command that has just
+        # gone on the wire", which is wrong twice over and was corrected by
+        # TEST-06's reviewer: `check_meter_frozen()` grades `self.target` - the
+        # arbiter's resolved intent - and `write_setpoint()` clamps only a local
+        # `value` and can skip the write entirely, so that number may never have
+        # gone anywhere.
+        #
+        # What both terms of the conjunction must describe is the SAME moment:
+        # the command this cycle resolved, against how long the meter that chose
+        # it has been standing still. Evaluated before
+        # `self.target = resolution.target_w` it would grade the PREVIOUS
+        # cycle's command against THIS cycle's meter - the one combination that
+        # was true of neither moment. Moving it between the assignment and
+        # write_setpoint() is therefore harmless; moving it above the assignment
+        # is not, and test_freeze.py part 5 fails on exactly that.
         await self.check_meter_frozen()
 
+        # --- TARIFF-01, and it is LAST on purpose -----------------------------
+        # Banking a quarter is bookkeeping about a moment that has already
+        # happened; a slow SD card must never sit between the meter and the
+        # setpoint. Its position is unconstrained in the way SAFETY-03's and
+        # SAFETY-18's are not - it grades neither `target` nor `standing` and
+        # reads only `self.p1`, which no line in this method writes.
+        self.peak.sync(self.p1)
+
     async def check_output_clamp(self) -> None:
         """SAFETY-03. Alarm whenever a clamp actually engaged this cycle.
 
@@ -818,6 +870,27 @@ class Controller:
                                     f"{self.freeze_wd.cfg.hold_s:.0f} s unchanged "
                                     f"while the command is over "
                                     f"{rail_w(self.o.get('max_w', 2000)):.0f} W)"})
+
+            # TARIFF-01. The month's billed peak, and - the load-bearing half -
+            # a DIFFERENT row when the store could not answer. "no peak yet" is
+            # a normal state on the 1st; "unreadable" is a broken instrument, and
+            # a status page that spelled them the same way would let a 0 W read
+            # look like a quiet month right up until SAFETY-07 shaved nothing.
+            _pk = self.peak.peak()
+            if _pk.is_unreadable:
+                out.append({"ok": False, "warn": True,
+                            "text": "capacity-tariff peak: " + _pk.explain()})
+            else:
+                # §9.1's "warning entity at 80 %", as a status row rather than
+                # a second entity - see the constructor on not doubling up.
+                _frac = self.peak.warning_frac() or 0.0
+                _ok = _frac < 0.8
+                out.append({"ok": _ok, "warn": not _ok,
+                            "text": _pk.explain()
+                                    + f" - {_frac * 100:.0f} % of the "
+                                      f"{self.peak.cap_w:.0f} W soft limit"
+                                    + f"; {self.peak.recorded} quarter(s) banked"
+                                    + f"; on start: {self.peak.restore_state}"})
         rows = [("battery SoC", self.soc, o.get("soc_entity")),
                 ("battery power", self.batt, o.get("batt_entity"))]
         if not self.p1_enabled:

goodwe_controller/app/p1.py

@@ -353,6 +353,54 @@ class QuarterAverager:
         self._last_pp = sample.per_phase_w
         return closed
 
+    # -- surviving a restart (TARIFF-01) ------------------------------------
+    def snapshot(self) -> dict:
+        """The open block's whole state, as JSON-safe primitives.
+
+        ⚠️ `last_t` is the HELD SAMPLE'S OWN timestamp and is stored as it is,
+        not refreshed to "now". That is what makes a restart integrate the gap
+        under exactly the `max_hold_s` rule a running process would have used:
+        a short restart is a plateau, a long one leaves uncovered seconds that
+        drag the closed block's average down. Rewriting it to the restore time
+        would silently credit the whole outage at the last value seen before it.
+        """
+        return {
+            "block": self._block,
+            "acc": self._acc,
+            "pp_acc": list(self._pp_acc),
+            "elapsed": self._elapsed,
+            "last_t": self._last_t,
+            "last_net": self._last_net,
+            "last_pp": list(self._last_pp) if self._last_pp is not None else None,
+            "phases": self.phases,
+        }
+
+    def restore(self, state: dict) -> bool:
+        """Put a `snapshot()` back. False if it does not fit this averager.
+
+        A phase-count change is a reconfiguration, not a resumption: the
+        per-phase accumulator would be reshaped mid-block and the two halves
+        would not be the same measurement. Refusing is the honest answer, and
+        the caller reports starting fresh rather than pretending it resumed.
+        """
+        try:
+            if int(state.get("phases", self.phases)) != self.phases:
+                return False
+            pp_acc = list(state.get("pp_acc") or [])
+            if len(pp_acc) != self.phases:
+                return False
+            self._block = None if state["block"] is None else int(state["block"])
+            self._acc = float(state["acc"])
+            self._pp_acc = [float(v) for v in pp_acc]
+            self._elapsed = float(state["elapsed"])
+            self._last_t = None if state["last_t"] is None else float(state["last_t"])
+            self._last_net = float(state["last_net"])
+            lp = state.get("last_pp")
+            self._last_pp = tuple(float(v) for v in lp) if lp is not None else None
+            return True
+        except (KeyError, TypeError, ValueError):
+            return False
+
 
 # --------------------------------------------------------------------------- #
 # what the rest of the add-on talks to

goodwe_controller/app/tariff.py

@@ -0,0 +1,682 @@
+"""TARIFF-01 - the monthly capacity-tariff peak, and the record that outlives it.
+
+WHAT THIS IS, AND WHAT IT DELIBERATELY IS NOT. This module is an INSTRUMENT. It
+measures the number the Belgian capaciteitstarief bills on - the highest
+15-minute average offtake of the calendar month (Fluvius / Resa / Ores) - and it
+writes that number down somewhere a restart cannot reach. It makes no claim,
+applies no clamp, and commands nothing. The guard that will read it is SAFETY-07,
+and it belongs in `arbiter.py` as a `P_PEAK` claim when it is built. Keeping the
+measurement and the actuation apart is the whole reason this file is small.
+
+WHAT WAS ALREADY BUILT, AND WHY THIS CARD IS NOT 8 SP OF NEW ARITHMETIC.
+`p1.QuarterAverager` already computes AC 1 in full: clock-aligned 900 s blocks,
+time-weighted over irregular telegram spacing, offtake-only, per-phase, with a
+`max_hold_s` gap rule. It was unwired and unpersisted - `P1Ingest.blocks` is a
+96-entry list in memory that a restart erases. So this file is the persistence
+and the monthly reduction, not a second accumulator. Adding one would be adding
+a second source of truth for the exact quantity the money depends on.
+
+  BLOCKS, NOT A SLIDING WINDOW.  The Description says "rolling"; AC 1 says
+  "15-min block average". They are different numbers and the utility bills the
+  second one. §9.1 calls the billed unit the "highest 15-minute average offtake",
+  and §9.1's own worked case - "one 4998 W quarter moved a monthly peak from
+  4832 -> 5013 W" - is stated per QUARTER, an object a sliding window does not
+  have. "Rolling" here means CONTINUOUSLY UPDATED (the open block's partial
+  average is readable at every sample), not a trailing-15-minutes window. A
+  trailing window would report a peak the meter never bills, and it reports it
+  HIGHER than the real one: a load burst straddling :15 fills a trailing window
+  completely while contributing to two block averages at partial weight. Guarding
+  that would spend battery defending an interval nobody charges for.
+
+  DELIVERED POWER, NOT INTENDED POWER.  SAFETY-20 has an open question about
+  whether guards should grade `Decision.target` (pre-clamp intent) or `standing`
+  (what actually reached the wire). For a capacity tariff there is no question to
+  ask: the utility's meter integrates the current that flowed. This module never
+  sees a setpoint at all - it consumes `P1Sample`, which is the meter's own
+  reading - and that is on purpose. A peak computed from intent would be wrong in
+  the expensive direction (a clamped 9 kW request that delivered 5 kW would
+  record 9 kW, and every consumer would shave against a peak that never existed).
+
+  AN UNREADABLE STORE IS NOT A ZERO PEAK.  This is the defect this repo spent a
+  day removing from `sim/riglock.py`, where `holder()` was rewritten to answer
+  `observed / not_observed / gone` instead of a partial dict every caller
+  `.get()`s a default out of. The same trap is here and it is worse, because the
+  degenerate value is not merely wrong, it is DISARMING: a peak read as 0 W gives
+  SAFETY-07 the maximum possible headroom and silently switches the guard off at
+  the moment its own storage is broken. So `peak()` returns a `PeakReading` whose
+  `state` is one of OBSERVED / NO_RECORD / NOT_OBSERVED, `peak_w` is `None`
+  unless the state is OBSERVED, and there is no `.get(..., 0)` shaped accessor
+  anywhere on it. NO_RECORD ("this month has no completed block yet", a normal
+  state on the 1st) and NOT_OBSERVED ("the database did not answer") are
+  different types of thing and are never spelled the same way.
+
+CONFIG: NONE ADDED, ON PURPOSE.  CFG-03 is open because the option defaults
+already live in three places held apart by a tripwire in `test_validate.py`. The
+soft limit this measurement is graded against already exists as `peak_cap_w`
+(config.yaml, default 3500), already validated, already read by
+`control.peak_at_risk()`. §14 names a `tariffs.capacity_peak_limit_kw: 2.5` - in
+KILOwatts, against watts everywhere else - and adding it would have made a fourth
+defaults site AND imported the unit mismatch §14's own warning box says must be
+caught at startup. So this module takes its limit as a plain constructor
+argument in WATTS and config.yaml is untouched by this ticket.
+
+THE DATABASE. SQLite in /data, the add-on's only durable volume, alongside
+`store.py`'s state.json. Two tiers from §15, and their retention is the point:
+
+  quarter_blocks   one row per completed 900 s block   INDEFINITE
+  monthly_peak     one row per calendar month          PERMANENT
+
+Neither is ever pruned here, and no sweep in this file can prune them - there is
+no DELETE statement against either table anywhere in this module. §15's asymmetry
+is why: raw samples vanish after 7 days, so a 15-minute aggregate that was not
+written AS IT HAPPENED can never be recomputed. `monthly_peak` is maintained as a
+derived record - upserted on every block close, and `recompute_month()` re-derives
+it from `quarter_blocks` so the two can be PROVEN not to have diverged rather
+than assumed not to have.
+
+RESTART MID-BLOCK. The open block's accumulator is memory. A restart at minute 12
+that dropped it would under-read the rest of the quarter, and SAFETY-07's clamp
+would release exactly when it was most needed. So the accumulator is checkpointed
+(`QuarterAverager.snapshot()`) into a single row, at most `CHECKPOINT_S` stale,
+and restored on startup. Three cases, all of them exercised in `test_tariff.py`:
+
+  * the checkpoint's block is the block containing `now`  -> restore it verbatim,
+    including the held sample's ORIGINAL timestamp, so the first sample after the
+    restart integrates the gap under the same `max_hold_s` rule a running process
+    would have applied. A restart longer than `max_hold_s` therefore leaves
+    uncovered seconds, and a closed block is always divided by the full 900 s, so
+    the gap DRAGS THE BILLED AVERAGE DOWN instead of being averaged away;
+  * the checkpoint's block ended while we were down -> that block really happened
+    and its measured part is real, so it is closed at `acc / 900` and recorded,
+    then a fresh block is opened. Discarding it would lose measured energy;
+  * no checkpoint, or one that cannot be read -> start clean and SAY SO. The
+    return value of `restore()` names which of the three happened; it is never
+    silently the first one.
+
+Double counting is answered structurally rather than by care: `quarter_start` is
+the PRIMARY KEY, and the recorded average may only ever be RAISED by a later
+write (`ON CONFLICT ... WHERE excluded.offtake_avg_w > ...`). Replaying the same
+block twice is therefore a no-op, and a restart that reprocesses a block with
+LESS coverage than the first pass cannot lower a figure already banked.
+
+MONTH BOUNDARY. Calendar months are LOCAL (Europe/Brussels); blocks are stored in
+UTC. In August those differ by two hours, so a block starting 2026-08-31 22:15
+UTC belongs to SEPTEMBER and putting it in August would move a month's bill.
+Block alignment itself needs no timezone - every Belgian offset is a whole number
+of hours, so the 900 s UTC grid lands on :00/:15/:30/:45 local on both sides of a
+DST transition, which is the argument `QuarterAverager` already carries.
+"""
+
+import json
+import logging
+import os
+import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone, tzinfo
+
+_LOG = logging.getLogger("goodwe.tariff")
+
+QUARTER_S = 900
+
+# How stale the partial-block checkpoint may be. This is the ONLY thing a crash
+# can lose from an open block, and the loss is bounded and one-directional: the
+# lost seconds were already integrated into `_acc`, so dropping them makes the
+# restored block read LOW, never high. At 15 s against a 900 s block the
+# worst-case under-read of one quarter's average is 15/900 = 1.7 % of that
+# quarter's own offtake - see test_tariff.py, which computes that bound from
+# these two constants rather than quoting it.
+CHECKPOINT_S = 15.0
+
+# -- reading states. Three, and never fewer. -------------------------------- #
+OBSERVED = "observed"          # the store answered and there is a peak
+NO_RECORD = "no_record"        # the store answered and this month has none yet
+NOT_OBSERVED = "not_observed"  # the store did not answer. NOT a zero peak.
+
+
+# --------------------------------------------------------------------------- #
+# local time, without a hard dependency on the tz database
+# --------------------------------------------------------------------------- #
+class _EUTimezone(tzinfo):
+    """Europe/Brussels by the EU harmonised rule, for hosts with no tzdata.
+
+    ⚠️ THIS IS A FALLBACK AND IT IS NOT SILENT. `zoneinfo` is tried first and is
+    what runs on the add-on's container. The fallback exists because the
+    alternative - quietly using UTC when tzdata is missing - misfiles every block
+    in the last two hours of a month into the wrong month's bill, and does it
+    without a symptom. Encoding the rule (CET, CEST from 01:00 UTC on the last
+    Sunday of March to 01:00 UTC on the last Sunday of October) is 20 lines and
+    is checked against zoneinfo, date by date, wherever zoneinfo is available.
+    """
+
+    # The rule, in both frames. `fromutc` is overridden rather than inherited
+    # because the base implementation calls back into `utcoffset` with a
+    # datetime whose tzinfo is self - and the first version of this class
+    # answered that by calling `astimezone`, which calls `utcoffset`, which
+    # recursed until the stack ran out. A tzinfo must answer from the value it
+    # is handed, never by converting it.
+    @staticmethod
+    def _last_sunday(year: int, month: int, hour: int) -> datetime:
+        d = datetime(year, month + 1, 1) - timedelta(days=1)
+        return (d - timedelta(days=(d.weekday() + 1) % 7)).replace(hour=hour)
+
+    def _dst_local(self, naive: datetime) -> bool:
+        """From a LOCAL wall clock: 02:00 CET last Sunday March -> 03:00 CEST
+        last Sunday October. The one ambiguous hour each autumn resolves to
+        standard time, which cannot move a month boundary."""
+        y = naive.year
+        return self._last_sunday(y, 3, 2) <= naive < self._last_sunday(y, 10, 3)
+
+    def _dst_utc(self, naive_utc: datetime) -> bool:
+        """From UTC: both EU transitions happen at 01:00 UTC, everywhere."""
+        y = naive_utc.year
+        return self._last_sunday(y, 3, 1) <= naive_utc < self._last_sunday(y, 10, 1)
+
+    def fromutc(self, dt: datetime) -> datetime:
+        return dt + timedelta(hours=2 if self._dst_utc(dt.replace(tzinfo=None)) else 1)
+
+    def utcoffset(self, dt):
+        if dt is None:
+            return timedelta(hours=1)
+        return timedelta(hours=2 if self._dst_local(dt.replace(tzinfo=None)) else 1)
+
+    def dst(self, dt):
+        if dt is None:
+            return timedelta(0)
+        return timedelta(hours=1 if self._dst_local(dt.replace(tzinfo=None)) else 0)
+
+    def tzname(self, dt):
+        return "CEST" if dt is not None and self._dst_local(dt.replace(tzinfo=None)) else "CET"
+
+
+def _billing_tz():
+    try:
+        from zoneinfo import ZoneInfo
+        tz = ZoneInfo("Europe/Brussels")
+        # Prove it resolves before trusting it: a ZoneInfo built from an empty
+        # tzdata answers, and answering wrongly is the failure being avoided.
+        if datetime(2026, 7, 1, tzinfo=timezone.utc).astimezone(tz).hour == 2:
+            return tz
+        _LOG.warning("zoneinfo Europe/Brussels resolved but gave the wrong "
+                     "summer offset - falling back to the encoded EU rule")
+    except Exception as err:                      # noqa: BLE001 - any failure
+        _LOG.info("no usable tz database (%s) - using the encoded EU rule", err)
+    return _EUTimezone()
+
+
+BILLING_TZ = _billing_tz()
+
+
+def month_key(when: datetime) -> str:
+    """The calendar month a UTC instant is billed in, as 'YYYY-MM' LOCAL time."""
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=timezone.utc)
+    local = when.astimezone(BILLING_TZ)
+    return f"{local.year:04d}-{local.month:02d}"
+
+
+# --------------------------------------------------------------------------- #
+# the per-phase reduction policy - one function, deliberately
+# --------------------------------------------------------------------------- #
+# §9.1 requires the average be computed per-phase-aware, but does NOT say whether
+# the BILLED figure on a three-phase connection is the sum of the phases or the
+# worst single phase. §5.2 requires the per-phase data; nothing states the
+# reduction. So it is one function with one argument, and correcting it later is
+# a one-line change plus a test parameter rather than a change to the
+# accumulator. §17's "Three-phase unbalanced load" scenario is the case that
+# tells the three apart - on a balanced load all three agree, which is exactly
+# why a balanced fixture must not be what this is tested on.
+POLICY_NET = "net"   # default: the connection's net offtake, what the meter bills
+POLICY_SUM = "sum"   # sum of the per-phase offtakes (differs when a phase exports)
+POLICY_MAX = "max"   # the worst single phase
+
+
+def billed_w(block, policy: str = POLICY_NET) -> float:
+    """Reduce one closed block to the single figure the month is billed on."""
+    pp = getattr(block, "per_phase_offtake_avg_w", None)
+    if policy == POLICY_NET or not pp:
+        return float(block.offtake_avg_w)
+    if policy == POLICY_SUM:
+        return float(sum(pp))
+    if policy == POLICY_MAX:
+        return float(max(pp))
+    raise ValueError(f"unknown per-phase reduction policy: {policy!r}")
+
+
+# --------------------------------------------------------------------------- #
+# what a read of the store answers
+# --------------------------------------------------------------------------- #
+@dataclass(frozen=True)
+class PeakReading:
+    """An answer about a month's peak, or an honest refusal to give one.
+
+    ⚠️ There is no `peak_w_or(default)` and there must never be one. Every
+    accessor that lets a caller turn "the database did not answer" into a number
+    re-creates the defect this type exists to remove. Callers ask `is_observed`
+    first, and a caller that forgets gets `None` and a TypeError, which is loud.
+    """
+
+    state: str
+    month: str
+    peak_w: float | None = None
+    quarter_start: datetime | None = None
+    per_phase_w: tuple | None = None
+    detail: str = ""
+
+    @property
+    def is_observed(self) -> bool:
+        return self.state == OBSERVED
+
+    @property
+    def is_unreadable(self) -> bool:
+        return self.state == NOT_OBSERVED
+
+    def explain(self) -> str:
+        if self.state == OBSERVED:
+            at = self.quarter_start.isoformat() if self.quarter_start else "?"
+            return f"{self.month}: peak {self.peak_w:.0f} W, set by the quarter at {at}"
+        if self.state == NO_RECORD:
+            return f"{self.month}: no completed quarter recorded yet"
+        return (f"{self.month}: PEAK UNREADABLE ({self.detail}) - this is NOT a "
+                f"zero peak and must not be used as headroom")
+
+
+# --------------------------------------------------------------------------- #
+# the store
+# --------------------------------------------------------------------------- #
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS quarter_blocks (
+    quarter_start   TEXT PRIMARY KEY,   -- ISO-8601 UTC, aligned to 900 s
+    month           TEXT NOT NULL,      -- 'YYYY-MM', LOCAL calendar month
+    offtake_avg_w   REAL NOT NULL,
+    per_phase_json  TEXT
+);
+CREATE INDEX IF NOT EXISTS ix_quarter_month ON quarter_blocks(month);
+CREATE TABLE IF NOT EXISTS monthly_peak (
+    month           TEXT PRIMARY KEY,
+    peak_w          REAL NOT NULL,
+    quarter_start   TEXT NOT NULL,
+    per_phase_json  TEXT
+);
+CREATE TABLE IF NOT EXISTS partial_block (
+    id              INTEGER PRIMARY KEY CHECK (id = 1),
+    state_json      TEXT NOT NULL
+);
+"""
+
+
+class PeakStore:
+    """SQLite behind /data. Every read answers with a state, never with a default."""
+
+    def __init__(self, path: str = "/data/tariff.db"):
+        self.path = path
+        self.ok = False
+        self.last_error: str | None = None
+        self._db: sqlite3.Connection | None = None
+        self._open()
+
+    def _open(self) -> None:
+        parent = os.path.dirname(self.path)
+        # ⚠️ THE DIRECTORY IS NOT CREATED. /data is mounted by the Supervisor and
+        # always exists on a real install, so `makedirs` here could only ever
+        # succeed where the durable volume is MISSING - manufacturing a database
+        # on the container's ephemeral layer that looks durable, reads fine, and
+        # is thrown away on the next restart. That is a peak store that silently
+        # forgets, which is worse than one that says it cannot answer.
+        if parent and not os.path.isdir(parent):
+            self.last_error = f"no durable volume at {parent}"
+            # INFO, not ERROR: on a dev host or in a test container there is no
+            # /data and nothing is wrong. A store that OPENED and then broke is
+            # the error case, and it is logged as one below.
+            _LOG.info("no peak store: %s - the monthly peak will read "
+                      "NOT_OBSERVED", self.last_error)
+            return
+        try:
+            self._db = sqlite3.connect(self.path, timeout=5.0,
+                                       check_same_thread=False)
+            # WAL: a power cut mid-write must not leave a torn database that
+            # reads as "no peak this month" - the exact shape of the defect.
+            self._db.execute("PRAGMA journal_mode=WAL")
+            self._db.executescript(_SCHEMA)
+            self._db.commit()
+            self.ok = True
+            self.last_error = None
+        except sqlite3.Error as err:
+            self._db = None
+            self.ok = False
+            self.last_error = str(err)
+            # ⚠️ Deliberately NOT raised. §6's fail-open rule: an aggregation
+            # fault must not stop a controller that is regulating a battery. The
+            # cost of continuing is a peak that reads NOT_OBSERVED, and every
+            # reader is built to treat that as "do not use me", not as 0 W.
+            _LOG.error("peak store unusable at %s (%s) - the monthly peak will "
+                       "read NOT_OBSERVED, which is not a zero peak", self.path, err)
+
+    def close(self) -> None:
+        if self._db is not None:
+            try:
+                self._db.close()
+            finally:
+                self._db = None
+                self.ok = False
+
+    # -- writing ------------------------------------------------------------
+    def record_block(self, block, policy: str = POLICY_NET) -> bool:
+        """Persist one completed block and fold it into its month's peak.
+
+        Both tables move in ONE transaction. A crash between them would leave a
+        banked quarter that the monthly peak does not know about, and the peak is
+        the record the money is read off.
+        """
+        if self._db is None:
+            return False
+        w = billed_w(block, policy)
+        start = block.start.astimezone(timezone.utc).isoformat()
+        month = month_key(block.start)
+        pp = getattr(block, "per_phase_offtake_avg_w", None)
+        pp_json = json.dumps(list(pp)) if pp else None
+        try:
+            with self._db:
+                # A block may only ever be RAISED. A restart that reprocesses a
+                # quarter it had already banked will have LESS coverage than the
+                # first pass (its first seconds are gone), so an unconditional
+                # REPLACE would silently lower a recorded figure.
+                self._db.execute(
+                    "INSERT INTO quarter_blocks(quarter_start, month, offtake_avg_w, per_phase_json)"
+                    " VALUES(?,?,?,?)"
+                    " ON CONFLICT(quarter_start) DO UPDATE SET"
+                    "   offtake_avg_w=excluded.offtake_avg_w,"
+                    "   per_phase_json=excluded.per_phase_json"
+                    " WHERE excluded.offtake_avg_w > quarter_blocks.offtake_avg_w",
+                    (start, month, w, pp_json))
+                self._db.execute(
+                    "INSERT INTO monthly_peak(month, peak_w, quarter_start, per_phase_json)"
+                    " VALUES(?,?,?,?)"
+                    " ON CONFLICT(month) DO UPDATE SET"
+                    "   peak_w=excluded.peak_w,"
+                    "   quarter_start=excluded.quarter_start,"
+                    "   per_phase_json=excluded.per_phase_json"
+                    " WHERE excluded.peak_w > monthly_peak.peak_w",
+                    (month, w, start, pp_json))
+            return True
+        except sqlite3.Error as err:
+            self.last_error = str(err)
+            _LOG.error("could not record quarter %s (%s)", start, err)
+            return False
+
+    # -- reading ------------------------------------------------------------
+    def peak(self, when: datetime | None = None, month: str | None = None) -> PeakReading:
+        """The billed peak for a month. Three possible answers, never a default."""
+        if month is None:
+            month = month_key(when or datetime.now(timezone.utc))
+        if self._db is None:
+            return PeakReading(NOT_OBSERVED, month,
+                               detail=self.last_error or "store never opened")
+        try:
+            row = self._db.execute(
+                "SELECT peak_w, quarter_start, per_phase_json FROM monthly_peak"
+                " WHERE month=?", (month,)).fetchone()
+        except sqlite3.Error as err:
+            self.last_error = str(err)
+            return PeakReading(NOT_OBSERVED, month, detail=str(err))
+        if row is None:
+            return PeakReading(NO_RECORD, month)
+        pp = tuple(json.loads(row[2])) if row[2] else None
+        return PeakReading(OBSERVED, month, peak_w=float(row[0]),
+                           quarter_start=datetime.fromisoformat(row[1]),
+                           per_phase_w=pp)
+
+    def recompute_month(self, month: str) -> PeakReading:
+        """Re-derive the peak from `quarter_blocks` - the divergence check.
+
+        `monthly_peak` is a derived record, so it can be PROVEN equal to the
+        tier it derives from instead of trusted to be. Read-only.
+        """
+        if self._db is None:
+            return PeakReading(NOT_OBSERVED, month,
+                               detail=self.last_error or "store never opened")
+        try:
+            row = self._db.execute(
+                "SELECT offtake_avg_w, quarter_start, per_phase_json FROM quarter_blocks"
+                " WHERE month=? ORDER BY offtake_avg_w DESC, quarter_start ASC LIMIT 1",
+                (month,)).fetchone()
+        except sqlite3.Error as err:
+            self.last_error = str(err)
+            return PeakReading(NOT_OBSERVED, month, detail=str(err))
+        if row is None:
+            return PeakReading(NO_RECORD, month)
+        pp = tuple(json.loads(row[2])) if row[2] else None
+        return PeakReading(OBSERVED, month, peak_w=float(row[0]),
+                           quarter_start=datetime.fromisoformat(row[1]),
+                           per_phase_w=pp)
+
+    def blocks_in(self, month: str) -> list:
+        """(quarter_start, offtake_avg_w) for a month, oldest first. [] on error
+        is safe here ONLY because no caller reads emptiness as a peak - AC 3's
+        counter reports the sample size it actually got."""
+        if self._db is None:
+            return []
+        try:
+            return [(datetime.fromisoformat(a), float(b)) for a, b in self._db.execute(
+                "SELECT quarter_start, offtake_avg_w FROM quarter_blocks"
+                " WHERE month=? ORDER BY quarter_start", (month,))]
+        except sqlite3.Error as err:
+            self.last_error = str(err)
+            return []
+
+    # -- the open block's checkpoint ----------------------------------------
+    def save_partial(self, state: dict) -> bool:
+        if self._db is None:
+            return False
+        try:
+            with self._db:
+                self._db.execute(
+                    "INSERT INTO partial_block(id, state_json) VALUES(1,?)"
+                    " ON CONFLICT(id) DO UPDATE SET state_json=excluded.state_json",
+                    (json.dumps(state),))
+            return True
+        except (sqlite3.Error, TypeError, ValueError) as err:
+            self.last_error = str(err)
+            return False
+
+    def load_partial(self) -> dict | None:
+        if self._db is None:
+            return None
+        try:
+            row = self._db.execute(
+                "SELECT state_json FROM partial_block WHERE id=1").fetchone()
+            return json.loads(row[0]) if row else None
+        except (sqlite3.Error, ValueError) as err:
+            self.last_error = str(err)
+            return None
+
+    def clear_partial(self) -> None:
+        # The one DELETE in this module, and it is against the checkpoint - a
+        # scratch row, not a retention tier. Neither quarter_blocks nor
+        # monthly_peak is deleted from anywhere in this file.
+        if self._db is None:
+            return
+        try:
+            with self._db:
+                self._db.execute("DELETE FROM partial_block WHERE id=1")
+        except sqlite3.Error as err:
+            self.last_error = str(err)
+
+
+# --------------------------------------------------------------------------- #
+# the tracker: what main.py holds
+# --------------------------------------------------------------------------- #
+@dataclass(frozen=True)
+class Projection:
+    """The open block's end-of-quarter estimate, with the reasons to distrust it.
+
+    `elapsed_s` travels WITH the number rather than beside it, because early in a
+    quarter a projection is a 10-second extrapolation over 900 s and a consumer
+    that cannot see that will react to noise. `covered_frac` is separately useful:
+    it is below 1.0 when the meter had a gap, which is the "unreliable quarter"
+    the functional note asks be marked rather than reported low.
+    """
+
+    projected_w: float
+    elapsed_w: float
+    elapsed_s: float
+    covered_frac: float
+    complete: bool
+    block_start: datetime | None
+
+    def explain(self) -> str:
+        return (f"quarter projects {self.projected_w:.0f} W "
+                f"({self.elapsed_s:.0f} s elapsed, {self.covered_frac * 100:.0f} % covered"
+                + ("" if self.complete else ", INCOMPLETE") + ")")
+
+
+class PeakTracker:
+    """Drives the store from `P1Ingest`. Measures; never commands."""
+
+    def __init__(self, store: PeakStore, cap_w: float = 3500.0,
+                 policy: str = POLICY_NET, checkpoint_s: float = CHECKPOINT_S):
+        self.store = store
+        self.cap_w = float(cap_w)
+        self.policy = policy
+        self.checkpoint_s = float(checkpoint_s)
+        self.recorded = 0
+        self.restore_state = "not attempted"
+        self._seen: set = set()
+        self._last_ckpt = 0.0
+
+    # -- startup ------------------------------------------------------------
+    def restore(self, averager, now: datetime | None = None) -> str:
+        """Put the open block back. Returns which of the three cases happened."""
+        now = now or datetime.now(timezone.utc)
+        snap = self.store.load_partial()
+        if not snap:
+            self.restore_state = ("fresh: no checkpoint" if self.store.ok
+                                  else "fresh: store unreadable")
+            return self.restore_state
+        block = snap.get("block")
+        if block is None:
+            self.restore_state = "fresh: checkpoint had no open block"
+            return self.restore_state
+
+        current = int(now.timestamp() // QUARTER_S) * QUARTER_S
+        if block == current:
+            if not averager.restore(snap):
+                # A reshaped accumulator (meter_phases changed) is a different
+                # measurement, not a resumable one. Say so; do not half-resume.
+                self.store.clear_partial()
+                self.restore_state = "fresh: checkpoint does not fit this averager"
+                return self.restore_state
+            self.restore_state = (
+                f"resumed the open block at {snap.get('elapsed', 0.0):.0f} s "
+                f"of {QUARTER_S} s ({snap.get('acc', 0.0) / QUARTER_S:.0f} W banked)")
+            return self.restore_state
+
+        # The block ended while the add-on was down. Its measured part really
+        # happened, so bank it at acc/900 rather than discard measured energy.
+        # It reads LOW because the seconds after the shutdown were never covered,
+        # and that is the honest figure: coverage a closed block did not get.
+        closed = _ClosedBlock(
+            start=datetime.fromtimestamp(block, timezone.utc),
+            offtake_avg_w=float(snap.get("acc", 0.0)) / QUARTER_S,
+            per_phase_offtake_avg_w=(tuple(v / QUARTER_S for v in snap["pp_acc"])
+                                     if snap.get("pp_acc") and snap.get("last_pp") else None))
+        banked = self.store.record_block(closed, self.policy)
+        self.store.clear_partial()
+        self.restore_state = (
+            f"closed a stale block from {closed.start.isoformat()} at "
+            f"{closed.offtake_avg_w:.0f} W (partial coverage), "
+            + ("banked" if banked else "NOT banked - store unwritable"))
+        return self.restore_state
+
+    # -- steady state -------------------------------------------------------
+    def sync(self, ingest, now_mono: float | None = None) -> int:
+        """Bank any blocks that closed, and refresh the checkpoint. Returns the
+        number of blocks newly recorded on this call."""
+        import time as _t
+        n = 0
+        for block in getattr(ingest, "blocks", ()):
+            key = block.start.timestamp()
+            if key in self._seen:
+                continue
+            if self.store.record_block(block, self.policy):
+                self._seen.add(key)
+                self.recorded += 1
+                n += 1
+        if n:
+            # A block just closed, so the checkpoint's block is gone: clearing it
+            # in the same breath is what stops a crash here from replaying it.
+            self.store.clear_partial()
+            self._last_ckpt = 0.0
+        self.checkpoint(ingest, now_mono or _t.monotonic())
+        return n
+
+    def checkpoint(self, ingest, now_mono: float, force: bool = False) -> bool:
+        if not force and (now_mono - self._last_ckpt) < self.checkpoint_s:
+            return False
+        avg = getattr(ingest, "averager", None)
+        if avg is None:
+            return False
+        self._last_ckpt = now_mono
+        return self.store.save_partial(avg.snapshot())
+
+    # -- reading ------------------------------------------------------------
+    def peak(self, when: datetime | None = None) -> PeakReading:
+        return self.store.peak(when)
+
+    def headroom_w(self, when: datetime | None = None) -> float | None:
+        """Watts left under the soft limit, or None when it cannot be known.
+
+        ⚠️ `None`, not `self.cap_w`. An unreadable peak returning full headroom
+        is the disarming default this whole module is shaped against.
+        """
+        r = self.peak(when)
+        if r.state == NO_RECORD:
+            return self.cap_w
+        if not r.is_observed:
+            return None
+        return max(0.0, self.cap_w - r.peak_w)
+
+    def warning_frac(self, when: datetime | None = None) -> float | None:
+        """Peak as a fraction of the soft limit - §9.1's 80 % warning entity."""
+        r = self.peak(when)
+        if r.state == NO_RECORD:
+            return 0.0
+        if not r.is_observed or self.cap_w <= 0:
+            return None
+        return r.peak_w / self.cap_w
+
+    @staticmethod
+    def project(averager, now: datetime) -> Projection:
+        """End-of-quarter average if the last sample held for the rest of it.
+
+        ⚠️ The two divisors are DIFFERENT and only coincide at the end of the
+        block. `elapsed_w` divides by the seconds actually covered (what the
+        meter has shown so far); `projected_w` divides by the full 900 s and
+        credits the held value over what remains. At `elapsed_s == 900` and full
+        coverage they are equal - which is precisely why nothing may be tested
+        only at that point.
+        """
+        start = averager.block_start
+        if start is None:
+            return Projection(0.0, 0.0, 0.0, 0.0, False, None)
+        elapsed_wall = max(0.0, (now - start).total_seconds())
+        elapsed = averager.elapsed_s
+        acc = averager.partial_ws
+        remaining = max(0.0, QUARTER_S - elapsed_wall)
+        held = max(getattr(averager, "_last_net", 0.0), 0.0)
+        return Projection(
+            projected_w=(acc + held * remaining) / QUARTER_S,
+            elapsed_w=(acc / elapsed) if elapsed > 0 else 0.0,
+            elapsed_s=elapsed_wall,
+            covered_frac=(elapsed / elapsed_wall) if elapsed_wall > 0 else 0.0,
+            complete=elapsed_wall <= 0 or elapsed >= elapsed_wall - 1e-9,
+            block_start=start)
+
+
+@dataclass(frozen=True)
+class _ClosedBlock:
+    """A block reconstructed from a checkpoint. Shaped like `p1.QuarterBlock` so
+    `record_block` has one input type; kept private so nothing else makes one."""
+
+    start: datetime
+    offtake_avg_w: float
+    per_phase_offtake_avg_w: tuple | None

goodwe_controller/test_freeze.py

@@ -2,7 +2,10 @@
 
     python3 test_freeze.py
 
-Three parts, and the last two are the ones that matter:
+Six parts. Parts 2 and 3 are the acceptance argument; parts 5 and 6 exist
+because TEST-06 found three mutations that survived parts 1-4 green - the
+evaluation POSITION inside cycle(), the NOT COVERED status string, and the
+shipped `meter_freeze_s` itself:
 
   1. the predicate's boundaries and the edge-trigger, table-driven;
   2. SILENT ON REAL CONVERGED DATA. `scenarios/p1-age-2026-08-26.jsonl.gz` is
@@ -18,6 +21,14 @@ Three parts, and the last two are the ones that matter:
      repo is next to this one (or `$GOODWE_SIM`), and an equivalent local stub
      otherwise. Both are exercised; the run says which one it used.
 
+  4. THE WIRING: the real `Controller.cycle()`, asserted on the service call
+     that leaves the process;
+  5. WHERE IN `cycle()` the detector is evaluated - by consequence, not by
+     source order. It runs after `write_setpoint()`, and that is what makes two
+     compositions safe (see the section for both);
+  6. WHAT THE STATUS PAGE SAYS about an uncovered transport, and the shipped
+     threshold read back out of `config.yaml`.
+
 ⚠️ Part 3 runs in WALL CLOCK - `unchanged_s` is `time.monotonic()` and the
 transport really polls - so it drives the detector at `hold_s=10`, not the
 shipped 300. The THRESHOLD is config; the MECHANISM is what is under test, and
@@ -57,6 +68,7 @@ import asyncio
 import gzip
 import json
 import os
+import re
 import subprocess
 import sys
 import time
@@ -79,6 +91,29 @@ HERE = Path(__file__).resolve().parent
 # The live install's numbers, which are the ones the exposure is written at.
 LIVE_MAX_W = 5000.0
 
+# --------------------------------------------------------------------------- #
+# THE SHIPPED THRESHOLD, READ FROM config.yaml (TEST-06 / AC3)
+# --------------------------------------------------------------------------- #
+# ⚠️ `meter_freeze_s: 300` used to appear in this file only as the literal 300,
+# retyped six times, so `300 -> 3600` in config.yaml left every check here
+# green. Everything below is now driven off the SHIPPED value, and the value
+# itself is pinned in test_validate.py the way MAINT-04 pins the other defaults.
+#
+# Regex, not PyYAML: the add-on image installs py3-aiohttp/py3-paho-mqtt only,
+# and a check that skips on ImportError is not a check.
+_CFG_TXT = (HERE / "config.yaml").read_text(encoding="utf-8")
+
+
+def _cfg_default(key):
+    """The single `  key: <number>` default line in config.yaml's options block."""
+    found = re.findall(rf"^  {key}: (-?\d+(?:\.\d+)?)\s*$", _CFG_TXT, re.M)
+    return float(found[0]) if len(found) == 1 else None
+
+
+SHIPPED_HOLD_S = _cfg_default("meter_freeze_s")
+SHIPPED_DEADBAND_W = _cfg_default("deadband_w")
+SHIPPED_MAX_W = _cfg_default("max_w")
+
 
 def check(name, cond):
     print(f"  {'ok  ' if cond else 'FAIL'} {name}")
@@ -135,6 +170,35 @@ check("the reason names both terms",
       "unchanged over 300" in wd.reason and "command over 4750" in wd.reason)
 
 
+# --------------------------------------------------------------------------- #
+# 1b. the SHIPPED threshold, and the two properties it has to have (AC3)
+# --------------------------------------------------------------------------- #
+# ⚠️ This is the number the whole ticket is an argument about, and it was pinned
+# nowhere: `meter_freeze_s: 300 -> 3600` used to leave this file and
+# test_validate.py both green. test_validate.py pins the literal in MAINT-04's
+# manner; here it is pinned by its CONSEQUENCES, so a schema-legal value that
+# gives up the argument goes red even though it is a legal value.
+print("the shipped meter_freeze_s (AC3)")
+check(f"config.yaml's numeric defaults were read (freeze {SHIPPED_HOLD_S}, "
+      f"deadband {SHIPPED_DEADBAND_W}, max_w {SHIPPED_MAX_W})",
+      None not in (SHIPPED_HOLD_S, SHIPPED_DEADBAND_W, SHIPPED_MAX_W))
+check("the code default and the shipped default are the same number - "
+      "FreezeWatchdogConfig() is a default, config.yaml is the ship",
+      FreezeWatchdogConfig().hold_s == SHIPPED_HOLD_S)
+check("this file's `tuning()` really is config.yaml's deadband, so the "
+      "mutual-exclusion arithmetic below is about the shipped law",
+      tuning().deadband_w == SHIPPED_DEADBAND_W)
+
+# The upper bound is money: config.yaml's own comment says 300 s is "still
+# inside the 900 s quarter that bills the capacity tariff". A hold that outlives
+# the quarter notifies after the peak is already billed.
+QUARTER_S = 900.0
+check(f"the alarm still lands inside the {QUARTER_S:.0f} s billing quarter "
+      f"({SHIPPED_HOLD_S:.0f} s hold)", SHIPPED_HOLD_S < QUARTER_S)
+# The lower bound is the conjunction itself - see the derivation below, which
+# now runs off the shipped numbers rather than off two retyped literals.
+
+
 # --------------------------------------------------------------------------- #
 # 2. silent on 40 minutes of real converged behaviour
 # --------------------------------------------------------------------------- #
@@ -183,10 +247,10 @@ def replay(hold_s, max_w, cycle_floor_s=15.0):
 
 
 for mw in (2000.0, LIVE_MAX_W):
-    wd_r, peak_cmd, peak_unchanged, fired = replay(300.0, mw)
+    wd_r, peak_cmd, peak_unchanged, fired = replay(SHIPPED_HOLD_S, mw)
     print(f"    max_w {mw:.0f}: worst |command| {peak_cmd:.0f} W (rail "
           f"{rail_w(mw):.0f} W), worst unchanged {peak_unchanged:.1f} s "
-          f"(hold 300 s), episodes {wd_r.episodes}")
+          f"(hold {SHIPPED_HOLD_S:.0f} s), episodes {wd_r.episodes}")
     check(f"max_w {mw:.0f}: never alarms on the converged capture", fired is None)
     check(f"max_w {mw:.0f}: and the rail term had margin to spare",
           peak_cmd < rail_w(mw))
@@ -194,7 +258,7 @@ for mw in (2000.0, LIVE_MAX_W):
 # The margin on the OTHER term, as numbers rather than as trust. Both figures
 # are quoted in config.yaml and freeze.py, so they are re-measured every run and
 # a capture swapped underneath them cannot leave the prose saying the old thing.
-_, _, worst_unchanged, _ = replay(300.0, LIVE_MAX_W)
+_, _, worst_unchanged, _ = replay(SHIPPED_HOLD_S, LIVE_MAX_W)
 gaps, last, prev = [], rows[0]["t"], None
 for r in rows:
     if prev is None or r["pwr"] != prev:
@@ -208,7 +272,8 @@ check("worst observed unchanged_s is the 8 s the docs quote",
       7.5 < worst_unchanged < 8.5)
 check("longest distinct-reading interval is the 15.4 s the docs quote",
       15.0 < max(gaps) < 16.0)
-check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
+check(f"the shipped {SHIPPED_HOLD_S:.0f} s is at least 15x the worst of the two",
+      SHIPPED_HOLD_S >= 15 * max(gaps))
 
 # --------------------------------------------------------------------------- #
 # 2b. TEL-05's case, reconstructed, because the capture cannot contain it
@@ -226,22 +291,26 @@ check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
 # -14.2 W) rather than invented: one hour of a perfectly healthy converged meter
 # that happens not to move.
 print("TEL-05's low-power case - a converged meter that legitimately repeats")
-tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=300))
+tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=SHIPPED_HOLD_S))
 t5 = tuning(LIVE_MAX_W)
 tgt, iw, sc = 0.0, 0.0, 0
 bare_would_fire = False
 tel05_cmd = 0.0
-for k in range(3600):
+# Long enough that the REJECTED bare threshold has to get its chance even if
+# someone raises meter_freeze_s - the mutation must be caught by the pins above,
+# not by this case going red for an incidental reason.
+TEL05_S = max(3600, int(SHIPPED_HOLD_S) + 300)
+for k in range(TEL05_S):
     if k % 15 == 0:                       # cycle_floor_s, the only trigger left
         d = compute(prev_w=tgt, grid_w=-14.2, actual_w=tgt, tuning=t5,
                     sat_count=sc, i_w=iw)
         tgt, iw, sc = d.target_w, d.i_w, d.sat_count
     tel05_cmd = max(tel05_cmd, abs(tgt))
     # The REJECTED design, written out so the case can be shown to be real.
-    bare_would_fire = bare_would_fire or float(k) >= 300
+    bare_would_fire = bare_would_fire or float(k) >= SHIPPED_HOLD_S
     tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
-print(f"    an hour at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
-      f"unchanged_s reaches 3599 s, episodes {tel05.episodes}")
+print(f"    {TEL05_S / 60:.0f} min at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
+      f"unchanged_s reaches {TEL05_S - 1} s, episodes {tel05.episodes}")
 check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
       bare_would_fire)
 check("the conjunction does not - this is what the rail term buys",
@@ -255,8 +324,11 @@ check("...and the command stayed nowhere near the rail",
 # figure is inside the deadband, so the command there is parked at 0 W. If
 # someone lowers meter_freeze_s or narrows the deadband, this is the check that
 # should stop them.
-DEADBAND_W = 15.0
-slowest_healthy_w = 3600.0 / 300.0
+DEADBAND_W = SHIPPED_DEADBAND_W          # config.yaml's, not a retyped 15
+slowest_healthy_w = 3600.0 / SHIPPED_HOLD_S
+print(f"    at the shipped {SHIPPED_HOLD_S:.0f} s hold, the slowest healthy "
+      f"meter that still looks unchanged is {slowest_healthy_w:.1f} W, against "
+      f"a {DEADBAND_W:.0f} W deadband")
 check("a healthy meter slow enough to look unchanged is inside the deadband",
       slowest_healthy_w < DEADBAND_W)
 _park = compute(prev_w=0.0, grid_w=-slowest_healthy_w, actual_w=0.0,
@@ -277,11 +349,19 @@ check("...and the same replay DOES alarm once both terms are made reachable",
 # 3. a genuinely frozen meter, over real HTTP, through the real transport
 # --------------------------------------------------------------------------- #
 def find_hwsim():
-    """The rig's own fake meter, if the infra repo is reachable."""
+    """The rig's own fake meter, if the infra repo is reachable.
+
+    ⚠️ `HERE.parents[1]` only exists when this package sits at least two levels
+    below the filesystem root - it raised IndexError and took the WHOLE suite
+    down on a checkout at `C:/goodwe_controller`, which is a failure of the
+    lookup, not of the thing being tested. Guarded, so a shallow checkout falls
+    through to the stub like any other missing sibling repo.
+    """
     cands = []
     if os.environ.get("GOODWE_SIM"):
         cands.append(Path(os.environ["GOODWE_SIM"]) / "hwsim.py")
-    cands.append(HERE.parents[1] / "goodwe" / "sim" / "hwsim.py")
+    if len(HERE.parents) > 1:
+        cands.append(HERE.parents[1] / "goodwe" / "sim" / "hwsim.py")
     for c in cands:
         if c.exists():
             return c
@@ -379,7 +459,15 @@ def with_server(argv, port, fn):
 
 
 hw = find_hwsim()
-print(f"frozen meter over HTTP - using {'the rig hwsim at ' + str(hw) if hw else 'the local stub'}")
+# ⚠️ WHICH TRANSPORT BACKED PART 3 IS PART OF THE RESULT, and it is repeated in
+# the summary at the end of the run. The two are not equivalent: pointed at the
+# real `sim/hwsim.py` the rig path failed 1 run in 3 (a 16 s window against a
+# 10 s hold, when hwsim.serve() is slow to start), while the stub is tuned never
+# to fail. A green log that does not say which one ran is a green log that can
+# have quietly substituted the weaker stand-in - the same shape as an assertion
+# that cannot fail, from the other end.
+PART3_TRANSPORT = f"the rig hwsim at {hw}" if hw else "the local stub"
+print(f"frozen meter over HTTP - using {PART3_TRANSPORT}")
 if hw:
     def argv_for(mode):
         return [HWSIM_DRIVER, "8791" if mode == "freeze" else "8792", mode,
@@ -454,11 +542,17 @@ class _Hass:
         # runaway under test never happens - which would have made this a check
         # that passes for the wrong reason.
         self.batt = 0.0
+        # ⚠️ The dead-poll-task case (AC1's second half): every input reads
+        # None, exactly as it does when the transport task has died while the
+        # controller keeps cycling.
+        self.dead = False
 
     async def limits(self, entity):
         return (-5000.0, 5000.0, 1.0)
 
     async def number(self, entity, invert=False):
+        if self.dead:
+            return None
         return {"sensor.grid": self.grid, "sensor.soc": 50.0,
                 "sensor.batt": self.batt}.get(entity)
 
@@ -479,44 +573,113 @@ class _Source:
         self.unchanged_s = unchanged_s
 
 
+# ⚠️ meter_freeze_s is deliberately NOT in here. A controller built with the key
+# absent must land on the shipped number by itself, and that is asserted below -
+# supplying it here would test this dict instead of the shipped default.
+STALE_INPUT_S = 15.0
+
+
+def _opts(**over):
+    o = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+         "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
+         "slew_w": 1000, "deadband_w": SHIPPED_DEADBAND_W,
+         "target_grid_w": -10, "step_w": 10, "heartbeat_s": 10,
+         "stale_input_s": STALE_INPUT_S}
+    o.update(over)
+    return o
+
+
+# ⚠️ TWO SURFACES, AND EVERY ASSERTION BELOW SAYS WHICH ONE IT IS ABOUT.
+# The first cut of this helper selected by SERVICE, which measured "any guard's
+# notification" while every assertion built on it was named for THIS guard's.
+# That was true only because `check_meter_frozen()` was the sole
+# `persistent_notification.create` in `app/`; SAFETY-03 (#12) adds a second one
+# (`goodwe_output_clamp`) raised by scenarios that rail the command on purpose,
+# and eight assertions here went red against it - AC1's core check included.
+#
+# Blanket id-keying is NOT the repair. Three of those eight are claims about the
+# WHOLE notification surface, and narrowing them to this guard's id would have
+# silently shrunk a claim rather than fixed a conflation - which is the defect
+# class this card exists to close. So both surfaces are available by name, and
+# each assertion picks one deliberately.
+FREEZE_NOTE_ID = "goodwe_meter_frozen"
+
+
+def all_notes_of(hass):
+    """EVERY HA persistent notification raised this run, whoever raised it."""
+    return [d for dom, svc, d in hass.calls
+            if (dom, svc) == ("persistent_notification", "create")]
+
+
+def _is_freeze_note(d):
+    """Is this notification the freeze detector's?
+
+    ⚠️ THE ID **OR** THE TITLE, and the `or` is measured rather than tidy.
+    Keyed on the id alone, renaming the id empties every list in this file and
+    SIX assertions collapse together - so the one check whose subject IS the id
+    could no longer fail on its own, in the file about assertions that cannot
+    fail. Recognising the guard by what it says as well leaves the rename
+    exactly one red: the id's own check. Measured both ways:
+
+        selector                      vs #12's app/    id renamed
+        by service (the first cut)      8 red            1 red
+        by id alone                     3 red            6 red
+        by id or title (this)           3 red            1 red
+
+    The union cannot re-admit another guard's alarm unless that guard also
+    calls itself frozen, which would be a confusable alarm on its own account.
+    """
+    return (d.get("notification_id") == FREEZE_NOTE_ID
+            or "frozen" in str(d.get("title", "")).lower())
+
+
+def notes_of(hass):
+    """Only THIS guard's - see _is_freeze_note()."""
+    return [d for d in all_notes_of(hass) if _is_freeze_note(d)]
+
+
+def notes_by_text_of(hass):
+    """This guard's, recognised by what it SAYS and never by its id.
+
+    Exists for the two assertions whose subject IS the id: selected by the id,
+    both would be true by construction.
+    """
+    return [d for d in all_notes_of(hass)
+            if "frozen" in str(d.get("title", "")).lower()]
+
+
 def wired(unchanged_s, grid, cycles=60, **over):
+    """Returns (hass, controller, trace).
+
+    `trace` is one row per cycle - (command after the cycle, the watchdog's
+    level, notifications raised so far) - which is what AC1 needs: the
+    QUESTION IS NOT WHETHER THE ALARM HAPPENS BUT ON WHICH CYCLE.
+    """
     async def go():
         hass = _Hass(grid)
-        c = Controller(dict(
-            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
-             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
-             "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
-             "slew_w": 1000, "deadband_w": 15, "target_grid_w": -10,
-             "step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
-            **over), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c = Controller(_opts(**over), hass, _Store(), _Pub(),
+                       stop=asyncio.Event())
         c.auto = True
         c.dev_min, c.dev_max = -5000.0, 5000.0
         c.p1_source = _Source(unchanged_s)
+        trace = []
         for _ in range(cycles):
             await c.read_inputs()
             await c.cycle()
-        return hass, c
+            trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        return hass, c, trace
     return asyncio.run(go())
 
 
-hass_f, ctl_f = wired(3600.0, 4000.0)
-# ⚠️ FILTERED BY notification_id, NOT BY SERVICE, since SAFETY-03. This add-on
-# now raises MORE THAN ONE kind of persistent notification, and the scenarios
-# below drive a command to the rail on purpose - which is a winding loop, which
-# legitimately raises the output-clamp alarm too. A bare `svc == "create"` test
-# therefore counts a DIFFERENT guard's alarm as this one's, which is exactly the
-# conflation SAFETY-03's second acceptance criterion exists to forbid. Keying on
-# the id is what makes "this detector said nothing" mean it.
-FROZEN_ID = "goodwe_meter_frozen"
-
-
-def frozen_notes(hass):
-    return [d for dom, svc, d in hass.calls
-            if (dom, svc) == ("persistent_notification", "create")
-            and d.get("notification_id") == FROZEN_ID]
-
-
-notes = frozen_notes(hass_f)
+# ⚠️ SAFETY-03 (#12) arrived at the same diagnosis independently and added a
+# `frozen_notes()` keyed on the id. `notes_of()` above subsumes it - same id,
+# plus the title fallback that keeps the id's own check independently
+# falsifiable, plus `all_notes_of()` for the claims that are about the whole
+# surface rather than about this guard.
+hass_f, ctl_f, trace_f = wired(3600.0, 4000.0)
+notes = notes_of(hass_f)                      # THIS guard's
+all_notes = all_notes_of(hass_f)              # the whole surface
 # The status row lives behind `p1_enabled`, which is what a homewizard_local
 # install has. Flipped here rather than plumbing a whole P1 transport in - the
 # row's own content is what is under test.
@@ -524,9 +687,28 @@ ctl_f.p1_enabled = True
 rows = [c for c in ctl_f.checks() if "METER MAY BE FROZEN" in c["text"]]
 print(f"    frozen at +4000 W: command ran to {ctl_f.target:.0f} W, "
       f"{len(notes)} notification(s), {len(rows)} red status row(s)")
-check("the shipped cycle() raises exactly one HA notification", len(notes) == 1)
+# SURFACE-WIDE, deliberately, and the second guard has now landed. SAFETY-03
+# ships `goodwe_output_clamp`, and this scenario - a runaway loop winding into
+# the rail - raises it legitimately: the loop asks for more than the rating and
+# is clamped. So the answer to "is this scenario still meant to raise exactly
+# one alarm" is NO, it raises exactly two, and BOTH ARE NAMED here rather than
+# the claim being narrowed until it went green again. A third guard has to come
+# back to this line, which is the whole point of keeping one assertion in the
+# file that accounts for the controller's entire alarm output.
+_ids = sorted(d.get("notification_id") for d in all_notes)
+print(f"    the whole notification surface this run: {_ids}")
+check("the whole notification surface carried exactly two alarms this run - "
+      "SAFETY-18's and SAFETY-03's, both named",
+      _ids == ["goodwe_meter_frozen", "goodwe_output_clamp"])
+# THIS GUARD'S, keyed on the id.
+check("...and the freeze detector is the one that raised it", len(notes) == 1)
+# ⚠️ Selected by TEXT, not by id - see notes_by_text_of(). One of the two
+# assertions whose subject IS the id (the other is the stability check in part
+# 5), so it must not be selected by the thing it is checking, and it has to be
+# able to fail while everything else stays green.
 check("...with a fixed notification_id so repeats replace, not stack",
-      notes and notes[0].get("notification_id") == "goodwe_meter_frozen")
+      [d.get("notification_id") for d in notes_by_text_of(hass_f)]
+      == [FREEZE_NOTE_ID])
 check("...and it says the controller is still running",
       bool(notes) and "STILL RUNNING" in notes[0]["message"])
 check("the status page carries a failed row saying so",
@@ -537,17 +719,293 @@ check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
 check("and it does NOT command 0 W - this cut is an alarm only",
       abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
 
-hass_n, ctl_n = wired(3600.0, -14.2)
-check("wired up, TEL-05's converged case still raises nothing",
-      not frozen_notes(hass_n))
-hass_m, ctl_m = wired(2.0, 8000.0)
-check("wired up, a real railed load against a moving meter raises nothing",
-      not frozen_notes(hass_m)
-      and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
-check("an uncovered transport (no unchanged_s) raises nothing, and says so",
-      not frozen_notes(wired(None, 4000.0)[0]))
+# ⚠️ THE THREE SCENARIOS THAT SAFETY-03 FORCED A DECISION ON, decided one at a
+# time and not reflexively narrowed to green. Measured on the merged tree:
+#
+#   scenario                              surface  ids                    freeze
+#   frozen meter, railed command (above)     2      frozen + clamp          1
+#   TEL-05 converged, -14.2 W                0      -                       0
+#   moving meter, a real 8 kW load           1      clamp                   0
+#   uncovered transport, no unchanged_s      1      clamp                   0
+#
+# CONVERGED stays SURFACE-WIDE. At the operating point this controller exists to
+# hold, the correct behaviour is that the whole add-on has nothing to say - not
+# merely that the freeze guard has nothing to say. Nothing about SAFETY-03
+# weakens that claim, and it still holds, so narrowing it would give away a real
+# property for nothing.
+#
+# The other two NARROW to notes_of(), and the reason is the same in both: the
+# clamp alarm there is CORRECT and is SAFETY-03's subject, not this file's. A
+# genuine 8 kW load really does drive the loop past the rating; an uncovered
+# transport really does not stop it doing so. Asserting silence across the whole
+# surface would make this file fail whenever a different guard correctly does
+# its job - a claim about somebody else's behaviour wearing this file's name.
+# What each still owns is the load-bearing half: THE FREEZE DETECTOR SAID
+# NOTHING. The surface is printed rather than asserted, so a reader can still
+# see what else fired without this file grading it.
+hass_n, ctl_n, _ = wired(3600.0, -14.2)
+check("wired up, TEL-05's converged case leaves the WHOLE SURFACE silent - at "
+      "the operating point the add-on should have nothing at all to say",
+      not all_notes_of(hass_n))
+
+hass_m, ctl_m, _ = wired(2.0, 8000.0)
+print(f"    real 8 kW load, moving meter: surface "
+      f"{sorted(d.get('notification_id') for d in all_notes_of(hass_m))}, "
+      f"freeze {len(notes_of(hass_m))}")
+check("wired up, a real railed load against a moving meter raises nothing FROM "
+      "THIS GUARD (the clamp alarm there is SAFETY-03's and is correct)",
+      not notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
+
+hass_u = wired(None, 4000.0)[0]
+print(f"    uncovered transport: surface "
+      f"{sorted(d.get('notification_id') for d in all_notes_of(hass_u))}, "
+      f"freeze {len(notes_of(hass_u))}")
+check("an uncovered transport (no unchanged_s) raises nothing FROM THIS GUARD, "
+      "which is the honest report of a detector that cannot see",
+      not notes_of(hass_u))
+
+
+# --------------------------------------------------------------------------- #
+# 5. WHERE in cycle() the detector is evaluated (TEST-06 / AC1)
+# --------------------------------------------------------------------------- #
+# ⚠️ WHERE THE BOUNDARY ACTUALLY IS - AND IT IS NOT write_setpoint().
+# The first version of this comment said the detector runs "after
+# write_setpoint()" and grades "the command that has just gone on the wire".
+# TEST-06's reviewer disproved both halves by mutation:
+#
+#     N3  move check_meter_frozen() to immediately BEFORE write_setpoint()
+#         -> green, and CORRECTLY green
+#
+# because `self.target = resolution.target_w` is assigned two lines above the
+# write, `write_setpoint()` clamps only a local `value`, and it can skip the
+# write entirely (`if not (changed or force): return`). So what the detector
+# grades is `self.target` - the arbiter's resolved intent, PRE device clamp and
+# possibly never emitted - and the invariant these checks pin is:
+#
+#     check_meter_frozen() runs AFTER `self.target = resolution.target_w`
+#
+# That distinction is not pedantry. A check more robust than its own
+# explanation is how the next person moves the call ABOVE the assignment
+# believing they had preserved the stated rule.
+#
+# Both properties below are asserted by an OBSERVABLE CONSEQUENCE rather than by
+# source order, so semantic-preserving refactors stay green (measured by the
+# reviewer: renaming the method, and calling it from a helper that itself runs
+# last, both survive - the suite drives `await c.cycle()` and never names the
+# method):
+#
+#   (a) BOTH TERMS MUST DESCRIBE THE SAME MOMENT. The command it grades has to
+#       be the one THIS cycle resolved, judged against how long the meter that
+#       chose it has stood still. Evaluated above the assignment it grades the
+#       PREVIOUS cycle's command against THIS cycle's meter - a pairing that was
+#       never true of any moment - and the alarm arrives a whole cycle late.
+#   (b) THE DEAD-POLL-TASK COMPOSITION. When the poll task dies the inputs go
+#       stale, the arbiter resolves 0 W, and the detector - evaluated after that
+#       resolution lands in self.target - correctly sees an unrailed command and
+#       CLEARS. Evaluated above the assignment it would still be looking at the
+#       railed command from before the failsafe, and would keep the alarm
+#       standing over a controller that is already commanding 0 W.
+print("WHERE the detector is evaluated in cycle() - the ordering AC1 protects")
+
+_railed = [i for i, (t, _, _) in enumerate(trace_f)
+           if abs(t) >= rail_w(LIVE_MAX_W)]
+_alarmed = [i for i, (_, _, n) in enumerate(trace_f) if n >= 1]
+_first_railed = _railed[0] if _railed else None
+_first_alarm = _alarmed[0] if _alarmed else None
+print(f"    ramp to the rail: first railed command on cycle {_first_railed}, "
+      f"first notification on cycle {_first_alarm}, over {len(trace_f)} cycles")
+# Without this the equality below could be satisfied on cycle 0, where "the same
+# cycle" and "one cycle later" are indistinguishable.
+check("the command takes several cycles to reach the rail, so 'same cycle' and "
+      "'one cycle later' are different observations",
+      _first_railed is not None and _first_railed >= 1
+      and len(trace_f) > _first_railed + 1)
+check("the alarm lands on the SAME cycle the command reaches the rail - "
+      "evaluated after `self.target = resolution.target_w` it grades the "
+      "command THIS cycle resolved; evaluated above that assignment it grades "
+      "the previous one and arrives a cycle late",
+      _first_alarm is not None and _first_alarm == _first_railed)
+
+
+def dead_poll(after_cycles=20, dead_cycles=4, revived_cycles=20):
+    """Alarm on a railed command, then kill the poll task and keep cycling.
+
+    The real thing: the transport task dies, every input reads None, and after
+    `stale_input_s` the arbiter's P_SAFETY stop resolves 0 W. `inputs_bad_since`
+    is backdated rather than slept through - the clock is wall clock, and the
+    mechanism rather than the wait is what is under test.
+    """
+    async def go():
+        hass = _Hass(4000.0)
+        c = Controller(_opts(), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c.auto = True
+        c.dev_min, c.dev_max = -5000.0, 5000.0
+        c.p1_source = _Source(3600.0)
+        for _ in range(after_cycles):
+            await c.read_inputs()
+            await c.cycle()
+        before = (c.target, c.freeze_wd.active, len(notes_of(hass)))
+        hass.dead = True
+        trace = []
+        for _ in range(dead_cycles):
+            await c.read_inputs()
+            if c.inputs_bad_since is not None:
+                c.inputs_bad_since = (datetime.now(timezone.utc)
+                                      - timedelta(seconds=STALE_INPUT_S + 1))
+            await c.cycle()
+            trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        # ...and then the transport comes back, against a meter that is still
+        # frozen. The loop re-rails and this is a NEW episode - the alarm is
+        # allowed, indeed required, to speak again.
+        hass.dead = False
+        revived = []
+        for _ in range(revived_cycles):
+            await c.read_inputs()
+            await c.cycle()
+            revived.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        # ⚠️ by TEXT: "the second episode REPLACED the first" is a claim about
+        # the id, so the list it is checked against must not be selected by the
+        # id - that would make it true by construction.
+        return before, trace, revived, notes_by_text_of(hass), c
+    return asyncio.run(go())
+
+
+_before, _dead, _revived, _freeze_notes, _ctl_d = dead_poll()
+_zeroed = [i for i, (t, _, _) in enumerate(_dead) if t == 0.0]
+print(f"    poll task killed: alarming at {_before[0]:+.0f} W "
+      f"(active={_before[1]}, {_before[2]} notification(s)) -> "
+      + ", ".join(f"cycle {i}: {t:+.0f} W active={a} notes={n}"
+                  for i, (t, a, n) in enumerate(_dead)))
+check("the episode really was standing before the poll task died - otherwise "
+      "the release below is vacuous",
+      _before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_MAX_W))
+check("a dead poll task drives the command to 0 W",
+      bool(_zeroed) and _zeroed[0] == 0)
+check("...and on THAT SAME cycle the alarm is already clear, because the "
+      "detector ran after 0 W was resolved into self.target, not before it",
+      bool(_zeroed) and not _dead[_zeroed[0]][1])
+check("...and it stays clear for as long as the failsafe holds it there",
+      _dead and not any(a for _, a, _ in _dead))
+# ⚠️ "and it raises nothing new while parked" is NOT asserted here, and the
+# reason matters more than the deletion. It was dropped as REDUNDANT, not as
+# undecidable: TEST-06's reviewer falsified it in one line - move the
+# `hass.call` above `if not fired: return` so the detector notifies every cycle,
+# and notifications appear in exactly these parked cycles (that mutation takes
+# this file RED 9). My first note here claimed no mutation could break it, which
+# was the very inference this file exists to distrust, made about this file.
+# The deletion stands because the exact count below catches that same mutation
+# nine times over - no coverage was lost - and the honest form of the claim is
+# "I tried to break it and here is what happened", never "I could not, so it
+# cannot".
+
+# ⚠️ AND THE OTHER HALF, or "no second notification" is a claim no mutation can
+# break: when the transport comes back against a meter that is STILL frozen the
+# loop re-rails, and that is a new episode which MUST speak. What must not
+# happen is a second notification stacking up beside the first.
+_re_railed = [i for i, (t, _, _) in enumerate(_revived)
+              if abs(t) >= rail_w(LIVE_MAX_W)]
+print(f"    transport revived: command re-rails on cycle "
+      f"{_re_railed[0] if _re_railed else None}, "
+      f"{_ctl_d.freeze_wd.episodes} episode(s), {len(_freeze_notes)} "
+      f"notification(s) from this guard, "
+      f"ids {sorted({n.get('notification_id') for n in _freeze_notes})}")
+# A PRECONDITION, not a discriminating check - the same role as "the command
+# takes several cycles to reach the rail" above, and labelled the same way:
+# without it the two checks below are satisfied by a controller that never came
+# back. The crudest falsifier (never file the loop claim) takes most of this
+# file red, which is the signature of a precondition rather than a check; the
+# narrowest one measured is a stale-input stop that never releases.
+check("PRECONDITION: a revived transport against a still-frozen meter re-rails "
+      "the command",
+      bool(_re_railed))
+check("...which is a SECOND episode, and it alarms again on the cycle it rails",
+      _ctl_d.freeze_wd.episodes == 2
+      and bool(_re_railed) and _revived[_re_railed[0]][2] == _before[2] + 1)
+# ⚠️ STABILITY, not the value - the value is pinned once, above. Renaming the
+# id must fail that check and NOT this one; the two say different things and
+# each has to be able to go red on its own.
+check("...on the same notification_id, so the second REPLACES the first "
+      "instead of stacking a second one beside it",
+      len(_freeze_notes) == 2
+      and len({n.get("notification_id") for n in _freeze_notes}) == 1)
+
+
+# --------------------------------------------------------------------------- #
+# 6. what the status page SAYS about coverage (TEST-06 / AC2)
+# --------------------------------------------------------------------------- #
+# ⚠️ The check above only ever asserted that an uncovered transport is SILENT,
+# which a status page claiming `armed` on a guard that cannot run also satisfies.
+# The AC was that the detector reports uncovered transports *and says so*, so
+# the string is what gets asserted here, in both directions: the uncovered row
+# must not claim `armed`, and the covered row must still claim it - or "does not
+# say armed" would be satisfiable by deleting the armed branch outright.
+print("what the status page SAYS about coverage")
+
+
+def freeze_rows(unchanged_s, grid, **over):
+    """The freeze-detector rows of the real checks(), on a p1 install.
+
+    ⚠️ WHAT THIS DOES NOT COVER. The rows live behind `if self.p1_enabled:`, and
+    the flag is set here by hand rather than by building a real P1 transport -
+    inherited from part 4's fixture. So everything below proves "WHEN the row is
+    produced it says the right thing", and nothing below proves "the row is
+    produced on homewizard_local and not otherwise". That gate is uncovered;
+    AC2 is about the string, and widening the fixture belongs to whoever covers
+    the gate.
+    """
+    _h, ctl, _t = wired(unchanged_s, grid, cycles=3, **over)
+    # p1_enabled is what a homewizard_local install has; the row's own content
+    # is what is under test, not the transport that feeds it.
+    ctl.p1_enabled = True
+    return ctl, [r for r in ctl.checks()
+                 if r["text"].startswith("freeze detector")]
+
+
+_ctl_u, _rows_u = freeze_rows(None, 4000.0, meter_source="ha_dsmr")
+_ctl_c, _rows_c = freeze_rows(2.0, -14.2)
+print(f"    uncovered: {_rows_u[0]['text'] if _rows_u else '(no row at all)'}")
+print(f"    covered:   {_rows_c[0]['text'] if _rows_c else '(no row at all)'}")
+check("an uncovered transport gets exactly one freeze-detector row",
+      len(_rows_u) == 1)
+check("...and it says NOT COVERED",
+      bool(_rows_u) and "NOT COVERED" in _rows_u[0]["text"])
+check("...and it does NOT claim the guard is armed - that would be the status "
+      "page lying on behalf of a guard that cannot run",
+      bool(_rows_u) and "armed" not in _rows_u[0]["text"])
+check("...and it names the meter_source it is not covering, so the operator "
+      "knows which install this is",
+      bool(_rows_u) and "ha_dsmr" in _rows_u[0]["text"])
+check("...and it names homewizard_local as the transport that would be covered",
+      bool(_rows_u) and "homewizard_local" in _rows_u[0]["text"])
+check("...flagged as a warning rather than a silent ok",
+      bool(_rows_u) and not _rows_u[0]["ok"] and _rows_u[0]["warn"])
+
+check("a COVERED transport gets exactly one freeze-detector row",
+      len(_rows_c) == 1)
+check("...and that one DOES say armed - without this, 'the uncovered row does "
+      "not say armed' is satisfied by deleting the armed branch",
+      bool(_rows_c) and "armed" in _rows_c[0]["text"]
+      and "NOT COVERED" not in _rows_c[0]["text"])
+check("...and it quotes the hold and the rail it is armed at, derived from the "
+      "config the controller was actually built with",
+      bool(_rows_c)
+      and f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s unchanged" in _rows_c[0]["text"]
+      and f"{rail_w(LIVE_MAX_W):.0f} W" in _rows_c[0]["text"])
+check("...as an ok row, since a covered armed guard is not a warning",
+      bool(_rows_c) and _rows_c[0]["ok"] and not _rows_c[0]["warn"])
+
+# AC3's other half, through the built object rather than through a literal: the
+# opts dict does NOT carry meter_freeze_s, so this is the number an install that
+# leaves the key at its config.yaml default actually lands on.
+print(f"    a Controller built with meter_freeze_s ABSENT holds "
+      f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s (config.yaml ships "
+      f"{SHIPPED_HOLD_S:.0f} s)")
+check("a Controller built with the key absent lands on the SHIPPED hold, so "
+      "config.yaml and main.py's fallback cannot drift apart unnoticed",
+      _ctl_c.freeze_wd.cfg.hold_s == SHIPPED_HOLD_S)
 
 print()
+print(f"part 3 (the live HTTP transport) ran against: {PART3_TRANSPORT}")
 if fails:
     print(f"{len(fails)} FAILED: {', '.join(fails)}")
     sys.exit(1)

goodwe_controller/test_tariff.py

@@ -0,0 +1,910 @@
+"""Runnable check for TARIFF-01, the capacity-tariff measurement.
+
+    python3 test_tariff.py
+
+Ten parts. What each is for, and - where it matters - what it is deliberately
+NOT allowed to be tested at:
+
+  1. THE BLOCK ARITHMETIC, against a worked example written out segment by
+     segment. The fixture is chosen to be non-degenerate on three axes at once:
+     irregular sample spacing, a sample that STRADDLES the :15 boundary, and a
+     gap longer than `max_hold_s` so the block's coverage is short of 900 s.
+  2. THE 900 s DEGENERACY, named and then avoided. `acc/elapsed`, `acc/900` and
+     the end-of-quarter projection are THREE DIFFERENT FORMULAS that all agree
+     at exactly full coverage of a complete block. A check run only there proves
+     nothing about any of them, so part 2 first shows them agreeing at that
+     point (so the trap is documented rather than asserted around), then shows
+     part 1's fixture separating all three - and every later part uses fixtures
+     that are away from it.
+  3. RESTART MID-BLOCK, through the real `QuarterAverager.snapshot()`/`restore()`
+     and a real SQLite file: a restarted run must land on the same closed block
+     as an uninterrupted one, and a run that DOES NOT restore must land lower.
+     Without that second half, "restore works" is satisfied by a fixture where
+     restoring changes nothing.
+  4. MONTH ROLLOVER, in LOCAL time, including a restart that straddles it.
+  5. AN UNREADABLE STORE IS NOT A ZERO PEAK - the acceptance argument. Three
+     states, three different answers, and no accessor that turns "the database
+     did not answer" into a number.
+  6. THE TIMEZONE FALLBACK, graded against `zoneinfo` day by day.
+  7. THE PER-PHASE REDUCTION on an UNBALANCED load. §17's three-phase case. On a
+     balanced load net/sum/max all coincide, which is why a balanced fixture is
+     worth nothing here.
+  8. `monthly_peak` vs a recompute from `quarter_blocks` - the derived record
+     proven equal to the tier it derives from, and proven never to be lowered.
+  9. AC 3, ON THE REPLAY DATA THAT ACTUALLY EXISTS. See the honesty note below.
+ 10. THE WIRING, through the real `Controller` and its real status page.
+
+------------------------------------------------------------------------------
+⚠️ AC 3 IS NOT MET, AND PART 9 DOES NOT PRETEND OTHERWISE.
+
+AC 3 asks for "the monthly peak held under the soft limit in >= 95 % of
+intervals over a 30-day simulation replay". THERE IS NO 30-DAY DATASET. The
+infra repo's `sim/scenarios/` holds exactly two days of this house's real meter
+- 2026-08-20 (the Illusmart failure day) and 2026-08-23 (the healthy reference)
+- exported on 2026-08-25 because they were about to age out of HA's recorder,
+and nothing longer was ever captured. `sim/scenarios/README.md` lists the whole
+inventory; the rest is RS485 bus frames, OTA witness logs and a 578-row rollback
+poll, none of which carries grid power over calendar time.
+
+So part 9 replays the two days that exist, reports the percentage over them, and
+states the sample size in days next to it. It is a MEASUREMENT, not a verdict:
+nothing in it exits non-zero on the 95 % figure, because a 2-day run cannot pass
+or fail a 30-day criterion and a check that claimed it could would be the more
+expensive of the two mistakes. Both days are also *uncontrolled* - the add-on
+was not shaving peaks on either - so the figure they give is the BASELINE the
+guard has to improve on, which is the useful thing a 2-day replay can say.
+
+The two days are committed here as `scenarios/p1-net-<date>.jsonl.gz`
+(`{"t": iso8601, "w": signed_watts}` per line, derived from the HA recorder
+export) for the reason `scenarios/p1-age-2026-08-26.jsonl.gz` already is: the
+infra repo is private, CI's token is minted for this repository only, and a
+regression fixture that CI cannot see is not a regression fixture.
+
+Closing AC 3 needs 30 days of capture. That is TEST-01's replay harness, and it
+cannot start before somebody starts recording.
+"""
+
+import asyncio
+import gzip
+import json
+import shutil
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from app.main import Controller
+from app.p1 import P1Ingest, QuarterAverager, QUARTER_S, make_sample, split_signed
+from app.tariff import (
+    BILLING_TZ, CHECKPOINT_S, NO_RECORD, NOT_OBSERVED, POLICY_MAX,
+    POLICY_NET, POLICY_SUM, PeakStore, PeakTracker, _EUTimezone, billed_w,
+    month_key,
+)
+
+fails = []
+HERE = Path(__file__).resolve().parent
+TMP = Path(tempfile.mkdtemp(prefix="goodwe-tariff-"))
+
+# The live install's numbers: NOTES.md and freeze.py both record max_w 5000
+# there. `peak_cap_w` ships at 3500 in config.yaml and is the SOFT LIMIT AC 3
+# grades against - read out of the shipped file in part 10 rather than typed.
+LIVE_MAX_W = 5000.0
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def near(a, b, tol=1e-6):
+    return a is not None and b is not None and abs(a - b) <= tol
+
+
+def sample(t: datetime, net_w: float, phases=1, phase_net=None):
+    """One P1 telegram, through the SHIPPED derivation.
+
+    `split_signed` and `make_sample` are what the real transports call, so the
+    sign convention under test is the one that ships rather than a local
+    re-derivation of it.
+    """
+    imp, exp = split_signed(net_w)
+    pi = pe = None
+    if phase_net is not None:
+        pairs = [split_signed(v) for v in phase_net]
+        pi = [a for a, _ in pairs]
+        pe = [b for _, b in pairs]
+    return make_sample("ha_signed", imp, exp, phases=phases,
+                       phase_import_w=pi, phase_export_w=pe,
+                       ingest_ts=t, ingest_mono=t.timestamp())
+
+
+def fresh_store(name: str) -> PeakStore:
+    return PeakStore(str(TMP / f"{name}.db"))
+
+
+# B is a clock-aligned quarter boundary: proved, not assumed, because every
+# other number in part 1 rests on it.
+B = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)
+
+
+# ==============================================================================
+print("1. the block arithmetic, on a worked example")
+
+check(f"the fixture's start {B.isoformat()} really is 900 s aligned",
+      int(B.timestamp()) % QUARTER_S == 0)
+
+# (t seconds after B, net W). Deliberately irregular, with an export, a gap
+# longer than max_hold_s, and a last sample that straddles the boundary.
+MAX_HOLD_S = 30.0
+FIXTURE = [(0, 1000.0), (20, 2000.0), (30, -500.0), (50, 3000.0),
+           (250, 4000.0), (880, 1200.0), (1000, 900.0)]
+
+# The worked example, written as the segments a person would compute by hand:
+# each HELD value and how many seconds of the FIRST block it is evidence for.
+# Two rules do all the work: a sample is held forward until the next one, capped
+# at MAX_HOLD_S; and export is 0 W of offtake, never a negative contribution.
+SEGMENTS = [
+    (1000.0, 20),    # 0 -> 20, a normal telegram interval, held in full
+    (2000.0, 10),    # 20 -> 30, likewise
+    (0.0, 20),       # 30 -> 50: -500 W is EXPORT, so 0 W of offtake for 20 s
+    (3000.0, 30),    # 50 -> 250 is a 200 s GAP; only the first 30 s is evidence
+    (4000.0, 30),    # 250 -> 880 is a 630 s gap; again only 30 s is evidence
+    (1200.0, 20),    # 880 -> 900: the STRADDLE's first half, up to the boundary
+]
+# The straddle's second half: the same held 1200 W, from the boundary until its
+# hold expires at 880 + 30. It must be credited to the NEXT block, not to this
+# one and not to neither.
+SPLIT_TAIL_W, SPLIT_TAIL_S = 1200.0, 10
+EXP_WS = sum(w * s for w, s in SEGMENTS)
+EXP_COVERED_S = sum(s for _, s in SEGMENTS)
+EXP_BLOCK_AVG = EXP_WS / QUARTER_S              # what the utility bills
+EXP_ELAPSED_AVG = EXP_WS / EXP_COVERED_S        # what a /elapsed divisor gives
+NAIVE_MEAN = sum(max(w, 0.0) for _, w in FIXTURE[:-1]) / (len(FIXTURE) - 1)
+
+avg = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+closed = []
+for dt_s, w in FIXTURE:
+    closed += avg.add(sample(B + timedelta(seconds=dt_s), w))
+
+print(f"    {len(SEGMENTS)} held segments -> {EXP_WS:,.0f} Ws over "
+      f"{EXP_COVERED_S:.0f} s of the {QUARTER_S} s block")
+check(f"exactly one block closed", len(closed) == 1)
+check(f"...starting at the fixture's own boundary {B.isoformat()}",
+      bool(closed) and closed[0].start == B)
+check(f"...averaging {EXP_BLOCK_AVG:.2f} W - the worked example's "
+      f"{EXP_WS:,.0f} Ws over the FULL {QUARTER_S} s",
+      bool(closed) and near(closed[0].offtake_avg_w, EXP_BLOCK_AVG, 1e-9))
+
+# The three ways of being wrong, each of which produces a plausible number.
+print(f"    the same data read three other ways: /elapsed = {EXP_ELAPSED_AVG:.2f} W, "
+      f"naive mean of samples = {NAIVE_MEAN:.2f} W, "
+      f"billed = {EXP_BLOCK_AVG:.2f} W")
+check("...and /elapsed is NOT the billed figure - a gap must drag the average "
+      "down, not be averaged away",
+      abs(EXP_ELAPSED_AVG - EXP_BLOCK_AVG) > 1.0)
+check("...and a naive mean over samples is NOT the billed figure either - it "
+      "weights a burst of fast telegrams like a slow one",
+      abs(NAIVE_MEAN - EXP_BLOCK_AVG) > 1.0)
+
+# The straddle: the 4000 W sample at t=400 is held ACROSS 900. Its evidence
+# expires at 430, so the second block gets none of it - and the second block
+# must still have been OPENED at 900, not at 1000.
+check(f"the block after the straddle opens at the boundary, not at the sample",
+      avg.block_start == B + timedelta(seconds=QUARTER_S))
+print(f"    the straddling sample is SPLIT at the boundary: "
+      f"{SEGMENTS[-1][0]:.0f} W x {SEGMENTS[-1][1]} s into the first block, "
+      f"x {SPLIT_TAIL_S} s into the second")
+check(f"...and the second block is credited the tail "
+      f"({SPLIT_TAIL_W * SPLIT_TAIL_S:,.0f} Ws), so a sample spanning :15 is "
+      f"attributed to BOTH blocks in proportion and wholly to neither",
+      near(avg.partial_ws, SPLIT_TAIL_W * SPLIT_TAIL_S, 1e-9))
+
+
+# ==============================================================================
+print()
+print("2. the 900 s degeneracy - where three different formulas agree")
+
+# A complete block at a constant power, fully covered. Here acc/elapsed,
+# acc/900 and the projection are ALL the same number. Anything checked only
+# here is checked against nothing.
+HELD_W = 2400.0
+deg = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+deg_closed = []
+for i in range(0, QUARTER_S + 1, 5):
+    deg_closed += deg.add(sample(B + timedelta(seconds=i), HELD_W))
+d_block = deg_closed[0].offtake_avg_w
+deg2 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+LAST_S = QUARTER_S - 5
+for i in range(0, LAST_S + 1, 5):
+    deg2.add(sample(B + timedelta(seconds=i), HELD_W))
+d_partial = deg2.offtake_avg_w
+# Graded AT the last sample: full coverage behind it, a held value in front.
+d_proj = PeakTracker.project(deg2, B + timedelta(seconds=LAST_S)).projected_w
+print(f"    at full coverage of a whole block: acc/900 = {d_block:.2f} W, "
+      f"acc/elapsed = {d_partial:.2f} W, projection = {d_proj:.2f} W")
+check("all three formulas coincide at exactly 900 s of full coverage - so a "
+      "check placed only there distinguishes none of them",
+      near(d_block, HELD_W, 1e-6) and near(d_partial, HELD_W, 1e-6)
+      and near(d_proj, HELD_W, 1e-6))
+
+# Away from it, on part 1's fixture at a point mid-block, they must separate.
+mid = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+for dt_s, w in FIXTURE[:-1]:
+    mid.add(sample(B + timedelta(seconds=dt_s), w))
+NOW = B + timedelta(seconds=500)
+p = PeakTracker.project(mid, NOW)
+print(f"    at t=500 s on part 1's fixture: acc/elapsed = {p.elapsed_w:.2f} W, "
+      f"projection = {p.projected_w:.2f} W, coverage = {p.covered_frac * 100:.1f} %")
+check("mid-block, with a gap in it, the elapsed average and the projection are "
+      "DIFFERENT numbers", abs(p.elapsed_w - p.projected_w) > 1.0)
+check("...and the projection carries elapsed_s, so a consumer can see it is an "
+      f"extrapolation over {QUARTER_S - p.elapsed_s:.0f} s it has not seen",
+      near(p.elapsed_s, 500.0, 1e-6))
+check("...and a gap shows up as coverage below 1.0 rather than as a confident "
+      "low number", p.covered_frac < 1.0 and not p.complete)
+
+
+# ==============================================================================
+print()
+print("3. restart mid-block: the same block, or a lower one, never a lost one")
+
+RUN_W = 2000.0
+STEP_S = 1
+RESTART_AT_S = 600      # minute 10 of the quarter - the functional note's case
+
+
+def feed(avg, t0_s, t1_s, w=RUN_W):
+    out = []
+    for i in range(t0_s, t1_s + 1, STEP_S):
+        out += avg.add(sample(B + timedelta(seconds=i), w))
+    return out
+
+
+# (a) the control: no restart at all.
+ctrl = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+ctrl_closed = feed(ctrl, 0, QUARTER_S)
+CTRL_W = ctrl_closed[0].offtake_avg_w
+
+# (b) restarted at minute 10, checkpoint restored through a real SQLite file.
+st = fresh_store("restart")
+tr = PeakTracker(st, cap_w=LIVE_MAX_W)
+a1 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+feed(a1, 0, RESTART_AT_S)
+check("the checkpoint round-trips through SQLite", st.save_partial(a1.snapshot()))
+a2 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)   # a brand-new process
+restored = tr.restore(a2, B + timedelta(seconds=RESTART_AT_S + 1))
+print(f"    restore() said: {restored}")
+check("...and restore() reports RESUMING rather than starting fresh",
+      restored.startswith("resumed"))
+r_closed = feed(a2, RESTART_AT_S + STEP_S, QUARTER_S)
+RESTART_W = r_closed[0].offtake_avg_w if r_closed else None
+
+# (c) the mutation guard: the SAME restart with no restore.
+a3 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+n_closed = feed(a3, RESTART_AT_S + STEP_S, QUARTER_S)
+NORESTORE_W = n_closed[0].offtake_avg_w if n_closed else None
+
+print(f"    uninterrupted {CTRL_W:.2f} W | restored {RESTART_W:.2f} W | "
+      f"restart with no restore {NORESTORE_W:.2f} W")
+check(f"a restart at {RESTART_AT_S} s lands on the SAME billed block as an "
+      f"uninterrupted run", near(RESTART_W, CTRL_W, 1e-6))
+check("...and without the restore it lands LOWER - so 'the restore works' is "
+      "not satisfied by a fixture where restoring changes nothing",
+      NORESTORE_W is not None and NORESTORE_W < CTRL_W - 1.0)
+# Derived, not asserted against a literal: the lost part is exactly the seconds
+# before the restart, at the power that was flowing.
+DROPPED_S = RESTART_AT_S + STEP_S     # the resumed feed starts one step later
+check(f"...and it is lower by exactly the {DROPPED_S} s it dropped "
+      f"({RUN_W * DROPPED_S / QUARTER_S:.2f} W)",
+      near(CTRL_W - NORESTORE_W, RUN_W * DROPPED_S / QUARTER_S, 1e-6))
+
+# (d) a LONG outage: restored, but the meter was away past max_hold_s. The
+# uncovered seconds must drag the block DOWN, not be held forward.
+OUTAGE_S = 200
+a4 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+st4 = fresh_store("outage")
+st4.save_partial(a1.snapshot())
+PeakTracker(st4, cap_w=LIVE_MAX_W).restore(a4, B + timedelta(seconds=RESTART_AT_S + OUTAGE_S))
+o_closed = feed(a4, RESTART_AT_S + OUTAGE_S, QUARTER_S)
+OUTAGE_UNCOVERED_S = OUTAGE_S - MAX_HOLD_S
+print(f"    a {OUTAGE_S} s outage inside the quarter, of which "
+      f"{OUTAGE_UNCOVERED_S:.0f} s is past the {MAX_HOLD_S:.0f} s hold: "
+      f"{o_closed[0].offtake_avg_w:.2f} W")
+check(f"...reads lower than the uninterrupted block by exactly the uncovered "
+      f"seconds ({RUN_W * OUTAGE_UNCOVERED_S / QUARTER_S:.2f} W) - a gap is "
+      f"never held forward across the whole outage",
+      near(CTRL_W - o_closed[0].offtake_avg_w,
+           RUN_W * OUTAGE_UNCOVERED_S / QUARTER_S, 1e-6))
+
+# (e) double counting is answered by the primary key, not by care.
+st5 = fresh_store("dedupe")
+st5.record_block(ctrl_closed[0])
+st5.record_block(ctrl_closed[0])
+st5.record_block(ctrl_closed[0])
+rows = st5.blocks_in(month_key(B))
+check("banking the same quarter three times leaves ONE row - quarter_start is "
+      "the primary key, so a replay or a restart cannot double-count",
+      len(rows) == 1 and near(rows[0][1], CTRL_W, 1e-9))
+
+# ...and a later pass with LESS coverage may not lower a banked figure.
+from app.tariff import _ClosedBlock as _CB
+st5.record_block(_CB(start=B, offtake_avg_w=CTRL_W / 2, per_phase_offtake_avg_w=None))
+after = st5.blocks_in(month_key(B))
+check(f"...and re-banking the same quarter at half the figure leaves it at "
+      f"{CTRL_W:.0f} W - a restart's partial re-run must never lower a "
+      f"quarter that was already measured in full",
+      len(after) == 1 and near(after[0][1], CTRL_W, 1e-9))
+
+# (f) the checkpoint's own staleness bound, derived from the constant.
+print(f"    CHECKPOINT_S = {CHECKPOINT_S:.0f} s, so a crash can lose at most "
+      f"{CHECKPOINT_S / QUARTER_S * 100:.1f} % of one quarter's average")
+check("the checkpoint bound is small enough that its worst case is under 2 % "
+      "of a quarter", CHECKPOINT_S / QUARTER_S < 0.02)
+
+
+# ==============================================================================
+print()
+print("4. month rollover, in LOCAL time, and across a restart that straddles it")
+
+# CEST is UTC+2, so the last two hours of 31 August UTC are already September.
+AUG_LATE = datetime(2026, 8, 31, 21, 45, tzinfo=timezone.utc)
+SEP_EARLY = datetime(2026, 8, 31, 22, 15, tzinfo=timezone.utc)
+print(f"    {AUG_LATE.isoformat()} is local "
+      f"{AUG_LATE.astimezone(BILLING_TZ).isoformat()} -> {month_key(AUG_LATE)}")
+print(f"    {SEP_EARLY.isoformat()} is local "
+      f"{SEP_EARLY.astimezone(BILLING_TZ).isoformat()} -> {month_key(SEP_EARLY)}")
+check("a quarter starting 21:45 UTC on 31 Aug bills in AUGUST",
+      month_key(AUG_LATE) == "2026-08")
+check("a quarter starting 22:15 UTC on 31 Aug bills in SEPTEMBER - filing it "
+      "under August would move a whole month's bill",
+      month_key(SEP_EARLY) == "2026-09")
+check("...and the two are genuinely on opposite sides of the boundary, not "
+      "the same answer twice", month_key(AUG_LATE) != month_key(SEP_EARLY))
+# Winter, where the offset is +1 and the boundary sits an hour later.
+JAN_LATE = datetime(2026, 1, 31, 22, 45, tzinfo=timezone.utc)
+check("in WINTER the same trap is an hour later: 22:45 UTC on 31 Jan is still "
+      "January, and 23:15 UTC is February",
+      month_key(JAN_LATE) == "2026-01"
+      and month_key(JAN_LATE + timedelta(minutes=30)) == "2026-02")
+
+st6 = fresh_store("months")
+AUG_PEAK_W, SEP_PEAK_W = 4998.0, 900.0
+st6.record_block(_CB(AUG_LATE, AUG_PEAK_W, None))
+st6.record_block(_CB(SEP_EARLY, SEP_PEAK_W, None))
+r_aug, r_sep = st6.peak(month="2026-08"), st6.peak(month="2026-09")
+print(f"    {r_aug.explain()}")
+print(f"    {r_sep.explain()}")
+check(f"August's peak is the {AUG_PEAK_W:.0f} W quarter (specs.md:583's own "
+      f"measured case) and September's is not",
+      r_aug.is_observed and near(r_aug.peak_w, AUG_PEAK_W)
+      and r_sep.is_observed and near(r_sep.peak_w, SEP_PEAK_W))
+check("...so a new month starts from its OWN quarters, not from the previous "
+      "month's peak", r_sep.peak_w < r_aug.peak_w)
+
+# The restart that straddles the boundary: a checkpoint whose open block is in
+# August, restored when `now` is already September.
+st7 = fresh_store("straddle")
+tr7 = PeakTracker(st7, cap_w=LIVE_MAX_W)
+straddler = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+for i in range(0, 400, STEP_S):
+    straddler.add(sample(AUG_LATE + timedelta(seconds=i), RUN_W))
+st7.save_partial(straddler.snapshot())
+after_boot = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
+said = tr7.restore(after_boot, SEP_EARLY + timedelta(minutes=3))
+print(f"    restore() across the boundary said: {said}")
+aug7, sep7 = st7.peak(month="2026-08"), st7.peak(month="2026-09")
+check("a restart that straddles the month boundary BANKS the dead August "
+      "quarter rather than dropping measured energy",
+      said.startswith("closed a stale block") and aug7.is_observed)
+check(f"...at its partial-coverage figure ({aug7.peak_w:.2f} W = "
+      f"{RUN_W * 399 / QUARTER_S:.2f} W, the seconds it actually covered)",
+      near(aug7.peak_w, RUN_W * 399 / QUARTER_S, 1e-6))
+check("...and September reads NO_RECORD, which is a different answer from "
+      "August's peak AND from a zero",
+      sep7.state == NO_RECORD and sep7.peak_w is None)
+check("...and the checkpoint is cleared, so a second boot cannot re-bank it",
+      st7.load_partial() is None)
+
+
+# ==============================================================================
+print()
+print("5. AN UNREADABLE STORE IS NOT A ZERO PEAK  (the acceptance argument)")
+
+# Three stores, three states, and the three must not answer alike.
+good = fresh_store("states")
+good_tr = PeakTracker(good, cap_w=LIVE_MAX_W)
+empty_r = good_tr.peak(B)
+
+gone = fresh_store("gone")
+gone_tr = PeakTracker(gone, cap_w=LIVE_MAX_W)
+gone.record_block(_CB(B, 3000.0, None))
+gone.close()                       # the database goes away under a live reader
+gone_r = gone_tr.peak(B)
+
+never = PeakStore(str(TMP / "no-such-dir" / "x.db"))
+never_tr = PeakTracker(never, cap_w=LIVE_MAX_W)
+never_r = never_tr.peak(B)
+
+# A store that is OPEN and whose query FAILS is a third code path - a closed
+# handle short-circuits before the SQL ever runs, so without this the
+# `except sqlite3.Error` arm of peak() is never executed by anything.
+broken = fresh_store("broken-table")
+broken.record_block(_CB(B, 3000.0, None))
+broken._db.execute("DROP TABLE monthly_peak")
+broken._db.commit()
+broken_r = PeakTracker(broken, cap_w=LIVE_MAX_W).peak(B)
+
+for label, r in (("fresh store, no quarter yet", empty_r),
+                 ("store that broke under a live reader", gone_r),
+                 ("store that never opened", never_r),
+                 ("store open, query failed", broken_r)):
+    print(f"    {label:38s} -> {r.state:13s} {r.explain()}")
+
+check("a fresh store with no quarter yet says NO_RECORD - a normal state on "
+      "the 1st of the month", empty_r.state == NO_RECORD)
+check("a store that broke says NOT_OBSERVED", gone_r.state == NOT_OBSERVED)
+check("a store that never opened says NOT_OBSERVED", never_r.state == NOT_OBSERVED)
+check("a store whose QUERY fails says NOT_OBSERVED too - a different code path "
+      "from a closed handle, and the one a half-corrupt database takes",
+      broken_r.state == NOT_OBSERVED and broken_r.peak_w is None)
+# ...and losing the DERIVED record does not lose the peak, which is the whole
+# argument for keeping monthly_peak recomputable from the 15-min tier.
+_rec = broken.recompute_month(month_key(B))
+print(f"    ...but with the 15-min tier intact, recompute_month() still "
+      f"answers: {_rec.explain()}")
+check("...while recompute_month() STILL answers from the surviving 15-min "
+      "tier - the permanent record is derived, so losing it is recoverable",
+      _rec.is_observed and near(_rec.peak_w, 3000.0))
+broken._db.execute("DROP TABLE quarter_blocks")
+broken._db.commit()
+check("...and once BOTH tiers are gone it says NOT_OBSERVED rather than "
+      "raising into the control loop",
+      broken.recompute_month(month_key(B)).state == NOT_OBSERVED)
+check("...and NO_RECORD and NOT_OBSERVED are DIFFERENT states, not two "
+      "spellings of the same one", empty_r.state != gone_r.state)
+check("...and they read differently to a human, too",
+      empty_r.explain() != gone_r.explain()
+      and "NOT" in gone_r.explain() and "zero peak" in gone_r.explain())
+
+check("an unreadable peak is None, not 0.0 - there is no number to mistake",
+      gone_r.peak_w is None and never_r.peak_w is None)
+check("...and NO_RECORD carries no number either", empty_r.peak_w is None)
+_typed = False
+try:
+    float(gone_r.peak_w)
+except TypeError:
+    _typed = True
+check("...and a caller who forgets to ask gets a TypeError rather than a "
+      "silently disarming zero", _typed)
+check("PeakReading exposes no accessor that defaults a missing peak to a "
+      "number - no .get(), no _or() ",
+      not [a for a in dir(gone_r)
+           if a.startswith(("get", "value")) or a.endswith("_or")])
+
+# The consequence, which is the reason any of this matters.
+print(f"    headroom under a {LIVE_MAX_W:.0f} W limit: "
+      f"no quarter yet -> {good_tr.headroom_w(B)}, "
+      f"unreadable -> {gone_tr.headroom_w(B)}")
+check("an unreadable store yields NO headroom figure - the disarming default "
+      "(full headroom, guard silently off) is unreachable",
+      gone_tr.headroom_w(B) is None and never_tr.headroom_w(B) is None)
+check(f"...where a genuinely empty month yields the full {LIVE_MAX_W:.0f} W, "
+      f"which is correct and is a different answer",
+      near(good_tr.headroom_w(B), LIVE_MAX_W))
+check("the 80 % warning fraction is likewise None when nothing was observed",
+      gone_tr.warning_frac(B) is None and good_tr.warning_frac(B) == 0.0)
+
+# And a real peak reduces headroom, so "always None" does not pass this part.
+good.record_block(_CB(B, LIVE_MAX_W * 0.9, None))
+print(f"    after one {LIVE_MAX_W * 0.9:.0f} W quarter: "
+      f"headroom {good_tr.headroom_w(B):.0f} W, "
+      f"warning {good_tr.warning_frac(B) * 100:.0f} % of the limit")
+check("a real observed peak gives a real headroom, so this part is not passed "
+      "by a store that answers None to everything",
+      near(good_tr.headroom_w(B), LIVE_MAX_W * 0.1, 1e-6))
+check(f"...and it is over §9.1's 80 % warning threshold",
+      good_tr.warning_frac(B) > 0.8)
+
+# A CORRUPT database file is the third way a store fails, and it must land in
+# NOT_OBSERVED too rather than raising into the control loop.
+corrupt_path = TMP / "corrupt.db"
+corrupt_path.write_bytes(b"this is not a database" * 64)
+corrupt = PeakStore(str(corrupt_path))
+c_r = PeakTracker(corrupt, cap_w=LIVE_MAX_W).peak(B)
+print(f"    a corrupt database file -> {c_r.state}")
+check("a corrupt database reads NOT_OBSERVED and does not raise into the "
+      "control loop (§6 fail-open)", c_r.state == NOT_OBSERVED)
+
+
+# ==============================================================================
+print()
+print("6. the timezone fallback, graded against zoneinfo")
+
+_zi = None
+try:
+    from zoneinfo import ZoneInfo
+    _zi = ZoneInfo("Europe/Brussels")
+except Exception as err:                      # noqa: BLE001
+    print(f"    zoneinfo unavailable here ({err}) - the fallback IS the tz")
+
+if _zi is None:
+    check("zoneinfo is absent, so the encoded rule cannot be graded here - "
+          "reported, not silently passed", True)
+else:
+    fb = _EUTimezone()
+    # ⚠️ HOURLY, NOT 3-HOURLY, ACROSS THE TRANSITION DAYS. Both EU transitions
+    # happen at 01:00 UTC, so a rule that is wrong BY AN HOUR differs from the
+    # right one only between 01:00 and 02:00 UTC. A 3-hourly sweep steps over
+    # that window entirely and grades nothing - measured: a mutation moving the
+    # UTC threshold from 01:00 to 02:00 survived the whole file at 3-hourly.
+    probes = [datetime(2026, 1, 1, 12, tzinfo=timezone.utc)]
+    probes += [datetime(2026, m, d, h, mi, tzinfo=timezone.utc)
+               for m in (3, 10) for d in (28, 29, 30, 31)
+               for h in range(24) for mi in (0, 30)]
+    probes += [datetime(2026, m, 1, 0, 30, tzinfo=timezone.utc) for m in range(1, 13)]
+    # ⚠️ COMPARE THE CONVERTED WALL CLOCK, NOT ITS utcoffset(). `astimezone`
+    # goes through `fromutc`, but the RESULT carries local wall values, so
+    # asking it for `.utcoffset()` routes straight back into the LOCAL arm and
+    # never grades `fromutc` at all. Measured: a mutation moving the UTC
+    # threshold by an hour survived 397 offset-comparing probes and dies
+    # immediately on the wall clock they produce.
+    def _wallclock(t, tz):
+        return t.astimezone(tz).replace(tzinfo=None)
+
+    bad = [t for t in probes if _wallclock(t, fb) != _wallclock(t, _zi)]
+    print(f"    {len(probes)} probes across both DST transitions and every "
+          f"month boundary: {len(bad)} disagreement(s) in the LOCAL WALL CLOCK "
+          f"each one converts to")
+    check("the encoded EU rule converts every probe to the same local wall "
+          "clock as zoneinfo, including the two transition weekends", not bad)
+    # ...and the probes are not all the same offset, or the check is vacuous.
+    offsets = {t.astimezone(_zi).utcoffset() for t in probes}
+    check(f"...and the probe set spans BOTH offsets ({sorted(str(o) for o in offsets)}), "
+          f"so agreeing is not free", len(offsets) == 2)
+    check(f"BILLING_TZ resolved to zoneinfo on this host, not the fallback",
+          not isinstance(BILLING_TZ, _EUTimezone))
+
+    # ⚠️ THE PROBES ABOVE ONLY EXERCISE `fromutc()`. `astimezone` on a UTC
+    # instant never calls `utcoffset()` with a local wall time, so the local
+    # arm of the rule was ungraded - a mutation that moved it by an hour
+    # survived the whole file. These probes drive it directly. Noon is used
+    # because it is unambiguous on every day of the year, including the two
+    # transition days where 02:00-03:00 local is missing or repeated.
+    #
+    # ⚠️ AND THE LOCAL PROBES MUST INCLUDE 03:00-03:59 ON THE OCTOBER SUNDAY.
+    # That hour is where a local rule that ends DST an hour late still claims
+    # CEST while zoneinfo has already returned to CET; at noon both answers
+    # agree and the probe is worth nothing. Measured: at noon-only, a mutation
+    # moving the local thresholds from 02:00/03:00 to 03:00/04:00 survived.
+    #
+    # The MARCH gap (02:00-02:59 local, which does not exist) is deliberately
+    # NOT probed: Python hands a nonexistent local time to `utcoffset` as an
+    # ordinary value and any answer is defensible, so grading it would pin an
+    # arbitrary choice. It cannot move a month boundary - it is the last Sunday
+    # of March, never the 1st of a month - which is the only thing this tz is
+    # used for.
+    wall = [datetime(2026, m, d, 12) for m, d in
+            [(3, 27), (3, 28), (3, 29), (3, 30), (10, 24), (10, 25),
+             (10, 26), (10, 27)]]
+    wall += [datetime(2026, 10, 25, h, mi) for h in range(0, 6) for mi in (0, 30)
+             if h != 2]                       # 02:00 local is the ambiguous fold
+    wall += [datetime(2026, 3, 29, h, mi) for h in (0, 1, 3, 4, 5) for mi in (0, 30)]
+    wall += [datetime(2026, m, 1, 12) for m in range(1, 13)]
+    bad_w = [w for w in wall
+             if w.replace(tzinfo=fb).utcoffset() != w.replace(tzinfo=_zi).utcoffset()]
+    w_offsets = {w.replace(tzinfo=_zi).utcoffset() for w in wall}
+    print(f"    {len(wall)} LOCAL-wall probes through utcoffset(): "
+          f"{len(bad_w)} disagreement(s), spanning "
+          f"{sorted(str(o) for o in w_offsets)}")
+    check("utcoffset() on local wall time matches zoneinfo across both "
+          "transition weekends - the arm astimezone() never reaches", not bad_w)
+    check("...and those probes span both offsets too", len(w_offsets) == 2)
+    bad_n = [w for w in wall
+             if w.replace(tzinfo=fb).tzname() != w.replace(tzinfo=_zi).tzname()]
+    check("...and tzname() agrees with zoneinfo at every one of them (CET/CEST)",
+          not bad_n)
+
+
+# ==============================================================================
+print()
+print("7. the per-phase reduction, on an UNBALANCED three-phase load (§17)")
+
+# One phase exporting is what separates the three policies. On a balanced load
+# they all give the same number, which is why this fixture is not balanced.
+PH = (3000.0, 500.0, -200.0)
+NET = sum(PH)
+u = QuarterAverager(phases=3, max_hold_s=MAX_HOLD_S)
+u_closed = []
+for i in range(0, QUARTER_S + 1, 5):
+    u_closed += u.add(sample(B + timedelta(seconds=i), NET, phases=3, phase_net=PH))
+blk = u_closed[0]
+vals = {p: billed_w(blk, p) for p in (POLICY_NET, POLICY_SUM, POLICY_MAX)}
+print(f"    phases {PH} (net {NET:+.0f} W) -> "
+      + ", ".join(f"{k}={v:.0f} W" for k, v in vals.items()))
+check(f"the NET policy bills the connection's net offtake ({NET:.0f} W)",
+      near(vals[POLICY_NET], NET, 1e-6))
+check(f"the SUM policy clips each phase first, so the exporting phase does not "
+      f"net off ({sum(max(v, 0.0) for v in PH):.0f} W)",
+      near(vals[POLICY_SUM], sum(max(v, 0.0) for v in PH), 1e-6))
+check(f"the MAX policy bills the worst single phase ({max(PH):.0f} W)",
+      near(vals[POLICY_MAX], max(PH), 1e-6))
+check("...and on THIS fixture all three are different numbers - which is the "
+      "whole reason the reduction is a named policy and not an assumption",
+      len({round(v, 6) for v in vals.values()}) == 3)
+check("the shipped default is NET, the meter's own billed register",
+      billed_w(blk) == vals[POLICY_NET])
+
+bal = QuarterAverager(phases=3, max_hold_s=MAX_HOLD_S)
+b_closed = []
+for i in range(0, QUARTER_S + 1, 5):
+    b_closed += bal.add(sample(B + timedelta(seconds=i), 3000.0, phases=3,
+                               phase_net=(1000.0, 1000.0, 1000.0)))
+b_vals = {p: billed_w(b_closed[0], p) for p in (POLICY_NET, POLICY_SUM, POLICY_MAX)}
+check(f"...and on a BALANCED load NET and SUM coincide ({b_vals[POLICY_NET]:.0f} W), "
+      f"so a balanced fixture could not have told them apart",
+      near(b_vals[POLICY_NET], b_vals[POLICY_SUM], 1e-6))
+
+
+# ==============================================================================
+print()
+print("8. monthly_peak is DERIVED, and provably has not diverged")
+
+st8 = fresh_store("derive")
+QUARTERS = [(0, 1200.0), (1, 4998.0), (2, 3357.0), (3, 800.0), (4, 4832.0)]
+for q, w in QUARTERS:
+    st8.record_block(_CB(B + timedelta(seconds=q * QUARTER_S), w, None))
+MAXQ = max(w for _, w in QUARTERS)
+stored = st8.peak(month=month_key(B))
+derived = st8.recompute_month(month_key(B))
+print(f"    {len(QUARTERS)} quarters banked; stored peak {stored.peak_w:.0f} W, "
+      f"recomputed from quarter_blocks {derived.peak_w:.0f} W")
+check(f"the stored monthly peak is the largest quarter ({MAXQ:.0f} W - "
+      f"specs.md:583's own figure)",
+      stored.is_observed and near(stored.peak_w, MAXQ))
+check("...and re-deriving it from the 15-min tier gives the same number and "
+      "the same quarter, so the two records are proven equal rather than "
+      "assumed to be",
+      derived.is_observed and near(derived.peak_w, stored.peak_w)
+      and derived.quarter_start == stored.quarter_start)
+check("...and the two measured regressions from §9.1 land on DIFFERENT peaks - "
+      f"the sustained {MAXQ:.0f} W quarter, not the oscillating 3357 W one",
+      not near(stored.peak_w, 3357.0))
+# A quarter arriving LATER and lower must not lower the month.
+st8.record_block(_CB(B + timedelta(seconds=9 * QUARTER_S), 100.0, None))
+check("a later, smaller quarter does not lower the month's peak",
+      near(st8.peak(month=month_key(B)).peak_w, MAXQ))
+check("...and the recompute agrees with it after that write, too",
+      near(st8.recompute_month(month_key(B)).peak_w, MAXQ))
+# No pruning: §15 says these two tiers are indefinite / permanent.
+src = (HERE / "app" / "tariff.py").read_text(encoding="utf-8")
+check("§15's retention is structural: no DELETE against quarter_blocks or "
+      "monthly_peak exists anywhere in tariff.py",
+      "DELETE FROM quarter_blocks" not in src
+      and "DELETE FROM monthly_peak" not in src
+      and "DROP TABLE" not in src)
+
+
+# ==============================================================================
+print()
+print("9. AC 3, on the replay data that actually exists")
+
+REPLAY = sorted((HERE / "scenarios").glob("p1-net-*.jsonl.gz"))
+print(f"    replay fixtures found: {len(REPLAY)} "
+      f"({', '.join(p.name for p in REPLAY) or 'none'})")
+check("the committed replay fixtures are present - a regression fixture CI "
+      "cannot see is not a regression fixture", len(REPLAY) >= 2)
+
+# ⚠️ `peak_cap_w:` appears TWICE in config.yaml - once under `options:` with the
+# default, once under `schema:` with `int(500,15000)`. Taking the last match
+# would blow up on the schema line, and taking any match that parses would be
+# luck; the FIRST is the options block, which is the number an install lands on.
+SHIPPED_CAP_W = None
+for line in (HERE / "config.yaml").read_text(encoding="utf-8").splitlines():
+    if line.strip().startswith("peak_cap_w:"):
+        try:
+            SHIPPED_CAP_W = float(line.split(":", 1)[1].strip())
+        except ValueError:
+            continue
+        break
+check("config.yaml's shipped peak_cap_w was read out of the file rather than "
+      "typed here", SHIPPED_CAP_W is not None)
+
+replay_days = 0
+total_q = 0
+under = 0
+for path in REPLAY:
+    st9 = fresh_store("replay-" + path.stem)
+    ing = P1Ingest(phases=1, max_age_s=MAX_HOLD_S)
+    tr9 = PeakTracker(st9, cap_w=SHIPPED_CAP_W or LIVE_MAX_W)
+    first = last = None
+    with gzip.open(path, "rt", encoding="utf-8") as fh:
+        for line in fh:
+            row = json.loads(line)
+            t = datetime.fromisoformat(row["t"])
+            first = first or t
+            last = t
+            ing.submit(sample(t, float(row["w"])))
+    tr9.sync(ing)
+    months = {month_key(b.start) for b in ing.blocks}
+    span_h = (last - first).total_seconds() / 3600.0
+    qs = [w for _, w in st9.blocks_in(sorted(months)[0])] if months else []
+    peak = max(qs) if qs else 0.0
+    ok_q = sum(1 for w in qs if w <= (SHIPPED_CAP_W or LIVE_MAX_W))
+    total_q += len(qs)
+    under += ok_q
+    replay_days += span_h / 24.0
+    print(f"    {path.name}: {span_h:.1f} h, {len(qs)} closed quarters, "
+          f"month peak {peak:.0f} W, "
+          f"{ok_q}/{len(qs)} = {100.0 * ok_q / max(len(qs), 1):.1f} % under the "
+          f"{SHIPPED_CAP_W:.0f} W soft limit")
+    check(f"{path.name}: the month's peak equals its own largest quarter",
+          near(st9.peak(month=sorted(months)[0]).peak_w, peak, 1e-9))
+    check(f"{path.name}: a full day of real telegrams produces ~96 quarters, "
+          f"got {len(qs)}", 90 <= len(qs) <= 96)
+
+PCT = 100.0 * under / max(total_q, 1)
+print()
+print(f"    AC 3 MEASURED: {under}/{total_q} quarters = {PCT:.1f} % under the "
+      f"{SHIPPED_CAP_W:.0f} W soft limit, over {replay_days:.1f} DAYS of "
+      f"replay, UNCONTROLLED (no peak shaving was running on either day).")
+print(f"    AC 3 ASKS FOR 30 DAYS. {replay_days:.1f} days exist. "
+      f"AC 3 IS NOT MET and this run does not claim it is - see the module "
+      f"docstring. The figure above is the BASELINE SAFETY-07 has to improve.")
+check("the replay produced a measurable number over a stated sample size, "
+      "rather than a criterion quietly redefined to fit the data",
+      total_q > 100 and replay_days < 30)
+
+
+# ==============================================================================
+print()
+print("10. the wiring: the real Controller, and what its status page says")
+
+
+class _Store:
+    data = {}
+
+    def set(self, *a):
+        pass
+
+    def get_time(self, *a):
+        return None
+
+
+class _Pub:
+    enabled = False
+
+    def publish(self, *a, **k):
+        pass
+
+    def close(self):
+        pass
+
+
+class _Hass:
+    async def limits(self, entity):
+        return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)
+
+    async def number(self, entity, invert=False):
+        return None
+
+    async def set_number(self, entity, value):
+        return True
+
+    async def call(self, domain, service, data):
+        return True
+
+
+def built(db_path, opts=None):
+    return Controller(dict(
+        {"setpoint_entity": "number.sp", "meter_source": "ha_signed",
+         "p1_net_entity": "sensor.net", "meter_phases": 1,
+         "max_w": LIVE_MAX_W, "gain": 0.6, "slew_w": 1000, "deadband_w": 15,
+         "target_grid_w": 0, "step_w": 10, "heartbeat_s": 10,
+         "stale_input_s": 15}, **(opts or {})),
+        _Hass(), _Store(), _Pub(), stop=asyncio.Event(),
+        peak_db_path=str(db_path))
+
+
+c = built(TMP / "wired.db")
+check(f"a Controller built with peak_cap_w ABSENT takes the SHIPPED "
+      f"{SHIPPED_CAP_W:.0f} W, so config.yaml and main.py's fallback cannot "
+      f"drift apart unnoticed", c.peak.cap_w == SHIPPED_CAP_W)
+
+# Drive one real quarter through the real ingest, then the real cycle's sync.
+for i in range(0, QUARTER_S + 1, 5):
+    c.p1.submit(sample(B + timedelta(seconds=i), 4998.0))
+n = c.peak.sync(c.p1)
+wr = c.peak.peak(B)
+print(f"    after one 4998 W quarter through the real ingest: "
+      f"{n} banked, {wr.explain()}")
+check("the closed quarter reached SQLite through the tracker the cycle calls",
+      n == 1 and wr.is_observed and near(wr.peak_w, 4998.0, 1e-6))
+
+# ⚠️ `cycle()` calls sync() ONCE A SECOND against a `blocks` list that keeps a
+# closed quarter for a DAY. Without a dedupe that is ~86 000 re-banks of the
+# same row per block, and `recorded` - the number the status page prints as
+# "quarters banked" - would be a count of loop iterations rather than of
+# quarters. The database stays correct either way, which is exactly why this
+# needs its own check instead of being caught by the peak.
+again = c.peak.sync(c.p1)
+third = c.peak.sync(c.p1)
+print(f"    sync() called three times over the same closed quarter: "
+      f"{n}, {again}, {third} newly banked; tracker reports "
+      f"{c.peak.recorded} quarter(s)")
+check("...and syncing the same quarter again banks nothing new, so the count "
+      "the status page shows is quarters and not cycles",
+      again == 0 and third == 0 and c.peak.recorded == 1)
+check("...and the peak is unchanged by the repeat",
+      near(c.peak.peak(B).peak_w, 4998.0, 1e-6))
+
+rows = [r for r in c.checks() if "capacity-tariff" in r["text"] or "peak" in r["text"]]
+peak_rows = [r for r in rows if "peak" in r["text"] and "quarter(s) banked" in r["text"]]
+print("    status row: " + (peak_rows[0]["text"] if peak_rows else "(none)"))
+check("the status page carries exactly one capacity-tariff peak row",
+      len(peak_rows) == 1)
+check("...naming the peak, the soft limit and the quarters banked, all derived "
+      "from the controller that was actually built",
+      bool(peak_rows) and "4998 W" in peak_rows[0]["text"]
+      and f"{SHIPPED_CAP_W:.0f} W soft limit" in peak_rows[0]["text"]
+      and "1 quarter(s) banked" in peak_rows[0]["text"])
+check("...and it is FLAGGED, because 4998 W is over §9.1's 80 % warning point "
+      f"on a {SHIPPED_CAP_W:.0f} W limit",
+      bool(peak_rows) and not peak_rows[0]["ok"] and peak_rows[0]["warn"])
+
+# The half that matters: the same page with a store that cannot answer.
+c2 = built(TMP / "no-such-dir" / "nope.db")
+u_rows = [r for r in c2.checks() if "capacity-tariff peak" in r["text"]]
+print("    status row (unreadable store): "
+      + (u_rows[0]["text"] if u_rows else "(none)"))
+check("an unreadable store gets its OWN row, saying UNREADABLE", len(u_rows) == 1)
+check("...and it does NOT read as a quiet month at 0 W - without this, a broken "
+      "instrument looks exactly like a month with no peak",
+      bool(u_rows) and "UNREADABLE" in u_rows[0]["text"]
+      and "0 %" not in u_rows[0]["text"] and u_rows[0]["warn"])
+check("...and the healthy row and the unreadable row are different text",
+      bool(u_rows) and bool(peak_rows)
+      and u_rows[0]["text"] != peak_rows[0]["text"])
+
+# THE MEASUREMENT IS OF DELIVERED POWER (SAFETY-20). Graded on the parsed
+# SYNTAX, not on the text: the module docstring argues this at length and names
+# every forbidden word while doing so, so a grep over the source would be
+# satisfied by deleting the explanation and would fail on writing one. What must
+# be true is that no NAME the code evaluates refers to a command.
+import ast as _ast
+
+_tree = _ast.parse(src)
+_names = {n.id for n in _ast.walk(_tree) if isinstance(n, _ast.Name)}
+_names |= {n.attr for n in _ast.walk(_tree) if isinstance(n, _ast.Attribute)}
+_names |= {n.name for n in _ast.walk(_tree)
+           if isinstance(n, (_ast.FunctionDef, _ast.ClassDef))}
+_FORBIDDEN = {"setpoint", "write_setpoint", "target", "Decision", "i_clamped",
+              "clamped", "Claim", "resolve", "P_PEAK", "arbiter"}
+_found = sorted(_names & _FORBIDDEN)
+_imports = sorted(m.module or "" for m in _ast.walk(_tree)
+                  if isinstance(m, _ast.ImportFrom))
+print(f"    tariff.py evaluates {len(_names)} distinct names and imports "
+      f"from {_imports or ['(stdlib only)']}")
+check(f"none of them is a command-side name {sorted(_FORBIDDEN)} - the utility "
+      f"bills what FLOWED, so this reads the meter and nothing else"
+      + (f"; found {_found}" if _found else ""), not _found)
+check("...and it imports nothing from arbiter, clamp or control: it files no "
+      "claim, applies no clamp, and commands nothing - SAFETY-07's job",
+      not [m for m in _imports if m in ("arbiter", "clamp", "control",
+                                        ".arbiter", ".clamp", ".control")])
+# ...and the guard is not vacuous: the same grader run over main.py, which DOES
+# command, must light up.
+_main = {n.attr for n in _ast.walk(
+    _ast.parse((HERE / "app" / "main.py").read_text(encoding="utf-8")))
+    if isinstance(n, _ast.Attribute)}
+check(f"the same grader over main.py finds {sorted(_main & _FORBIDDEN)}, so it "
+      f"is capable of failing", bool(_main & _FORBIDDEN))
+
+
+# ==============================================================================
+print()
+shutil.rmtree(TMP, ignore_errors=True)
+if fails:
+    print(f"{len(fails)} FAILED: {', '.join(fails)}")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_validate.py

@@ -253,6 +253,91 @@ check("...and the shipped charge rate is at or under the shipped ceiling",
 check("...and the shipped discharge rate too, so it warns about nothing",
       _shipped["maintenance_discharge_w"] <= _shipped["max_w"])
 
+# --------------------------------------------------------------------------- #
+print("the shipped meter_freeze_s, pinned (TEST-06 / AC3)")
+# ⚠️ WHY A NUMBER GETS ITS OWN CHECK. `meter_freeze_s: 300` is the threshold
+# SAFETY-18's entire argument is about, and before this it was pinned NOWHERE:
+# 300 -> 3600 left test_freeze.py AND this file green, because the schema
+# accepts int(60,3600) and no check ever read the shipped value. The schema
+# bounds what is CONFIGURABLE; this pins what is SHIPPED, which is a different
+# question and the one the docs answer.
+#
+# Read by regex like _shipped above, for the same reason: no PyYAML in the
+# image, and a check that skips on ImportError is not a check.
+_freeze_s = [float(v) for v in re.findall(
+    r"^  meter_freeze_s: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
+_freeze_schema = re.findall(r"^  meter_freeze_s: int\((\d+),(\d+)\)\s*$",
+                            _txt, re.M)
+check(f"config.yaml carries exactly one meter_freeze_s default {_freeze_s}",
+      len(_freeze_s) == 1)
+check("...and exactly one schema range for it", len(_freeze_schema) == 1)
+_hold_s = _freeze_s[0] if _freeze_s else None
+_lo, _hi = ((float(_freeze_schema[0][0]), float(_freeze_schema[0][1]))
+            if _freeze_schema else (None, None))
+check("the shipped meter_freeze_s is still the measured 300 s that freeze.py, "
+      "config.yaml's own comment and test_freeze.py are all written at",
+      _hold_s == 300.0)
+check(f"...and it is inside its own schema range ({_lo}-{_hi} s)",
+      None not in (_lo, _hi, _hold_s) and _lo <= _hold_s <= _hi)
+
+# The two properties the number has to have, DERIVED rather than restated - so
+# they go red on a value that is merely schema-legal. Both come straight from
+# config.yaml's own comment on the key.
+#
+#   upper: the alert must land inside the 900 s quarter that bills the capacity
+#          tariff, or the notification arrives after the money is spent;
+#   lower: the slowest a HEALTHY meter can tick a 1 Wh register and still look
+#          unchanged for hold_s is 3600/hold_s W, and that must stay inside the
+#          deadband or the conjunction's two terms stop being mutually
+#          exclusive (freeze.py, "WHERE hold_s COMES FROM").
+_QUARTER_S = 900.0
+_shipped_deadband = [float(v) for v in re.findall(
+    r"^  deadband_w: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
+check(f"config.yaml carries exactly one deadband_w default {_shipped_deadband}",
+      len(_shipped_deadband) == 1)
+_db = _shipped_deadband[0] if _shipped_deadband else None
+_slowest_healthy_w = 3600.0 / _hold_s if _hold_s else None
+print(f"    shipped meter_freeze_s {_hold_s} s: alert inside the "
+      f"{_QUARTER_S:.0f} s billing quarter, and a healthy meter slow enough to "
+      f"look unchanged for it is {_slowest_healthy_w} W against a "
+      f"{_db} W deadband")
+check(f"...the alarm still lands inside the {_QUARTER_S:.0f} s billing quarter",
+      _hold_s is not None and _hold_s < _QUARTER_S)
+check("...and a healthy meter that looks unchanged for it is still inside the "
+      "deadband, so the two terms stay mutually exclusive",
+      _slowest_healthy_w is not None and _db is not None
+      and _slowest_healthy_w < _db)
+
+# ⚠️ AND THE RANGE'S OWN BOUNDS. "300 sits inside the range" is worth nothing
+# while the range itself is unpinned: widened to `int(1,86400)` an operator can
+# configure `meter_freeze_s: 86400` and the guard is dead for a day at a time,
+# with every check in this repo still green. That mutation survived the first
+# cut of this PR and is what this block closes. Both numbers are argued for in
+# config.yaml's own comment and in freeze.py, and each is re-derived here:
+#
+#   60 s   the floor freeze.py calls NOT A RECOMMENDATION - there a healthy
+#          meter can look unchanged at 60 W, four times the shipped deadband, so
+#          the conjunction's two terms stop being mutually exclusive and only
+#          the margin is left;
+#   3600 s the ceiling - already four times the billing quarter, so the top of
+#          the range has given up detecting a freeze inside the quarter that
+#          bills it. It is the last bound between a supported configuration and
+#          a disabled guard, and moving it is a decision rather than a tidy-up.
+print(f"    schema range {_lo}-{_hi} s: at the floor a healthy meter that looks "
+      f"unchanged is "
+      + (f"{3600.0 / _lo:.0f} W" if _lo else "?")
+      + f" against a {_db} W deadband; the ceiling is "
+      + (f"{_hi / _QUARTER_S:.1f}x" if _hi else "?")
+      + " the billing quarter")
+check(f"the schema range itself is still the argued 60-3600 s (found "
+      f"{_lo}-{_hi})", (_lo, _hi) == (60.0, 3600.0))
+check("...a floor that gives up the mutual exclusion and keeps only the "
+      "margin, which is why config.yaml calls it not a recommendation",
+      None not in (_lo, _db) and 3600.0 / _lo > _db)
+check("...and a ceiling already past the billing quarter, so no value the "
+      "schema still accepts can alarm inside the quarter that bills it",
+      _hi is not None and _hi > _QUARTER_S)
+
 # --------------------------------------------------------------------------- #
 print("the KNOWN cross-file default divergence, pinned rather than commented")
 # ⚠️ WHAT THIS GUARDS, AND WHY IT ASSERTS THE BUG RATHER THAN THE FIX.