Merge release/1.0 (TARIFF-01) into TARIFF-02

TARIFF-01 (#17) landed on release/1.0 first, so this branch was two commits behind and Gitea reported the PR unmergeable. Integrating rather than rebasing: the two reviewed commits stay addressable, and the branch is protected against force-push anyway. ONE CONFLICT, AND IT IS THE BORING HALF OF THE ONE REVIEW PREDICTED. Review of #16 expected collisions in four places - the tail of `options:`, the tail of `schema:`, the rule list inside validate_startup_config(), and the tail of validate.py. None of them happened: TARIFF-01 touched neither config.yaml nor validate.py. It added app/tariff.py, test_tariff.py, two P1 scenario captures, and edits to main.py and p1.py, none of which this branch goes near. The only conflict was CHANGELOG.md, where both branches prepended a block under the same `## Unreleased` heading. Both blocks kept, TARIFF-02 above TARIFF-01 since it lands second. Nothing was dropped or reworded on either side, and the resolution adds no line of its own beyond the blank one between them. Checked rather than assumed after the merge: - all ELEVEN test files pass, including TARIFF-01's own test_tariff.py (which this branch had never run) and the 93 s test_loop.py; - the 24-mutation battery still has 0 survivors; - no semantic overlap in config: TARIFF-01's peak tracker reads the peak_cap_w / peak_forecast_entity options that already existed, so the price options added here neither shadow nor duplicate anything. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
TARIFF-02 review: close the three other doors to the markup, and stop overclaiming
2026-08-30 19:42:16 +02:00 · 2026-08-30 19:35:36 +02:00 · 2026-08-30 19:33:57 +02:00 · 2026-08-30 19:13:58 +02:00 · 2026-08-30 19:05:12 +02:00 · 2026-08-30 18:39:07 +02:00
12 changed files with 4498 additions and 1 deletions
@@ -2,6 +2,138 @@

 ## Unreleased

+**Price provider plugins, and one canonical all-in price format (TARIFF-02).**
+`app/price.py` ships five providers - `generic_dayahead` (wholesale EPEX BE),
+`engie_dynamic`, `totalenergies_dynamic`, `eneco_eddy` and the offline
+`manual_csv` - all emitting exactly `{"prices": [{"start", "eur_per_kwh",
+"provisional"}]}`, tz-aware at the Brussels offset, EUR/kWh, all-in. **Nothing
+consumes it yet**: this is the seam, landed first, so that the arbitrage work
+that follows never learns a supplier's name.
+
+**Double-counting the markup is refused rather than discouraged, given a
+correct declaration.** A plugin does not return a price; it returns the figure
+its source published and declares which of {energy, markup, taxes} that figure
+contains. The framework adds the complement, once. Engie and TotalEnergies
+publish a retail figure with their margin already inside; adding the configured
+3 c/kWh on top would make every future arbitrage decision wrong by that amount
+with nothing reporting it.
+
+Two mechanisms, because one was not enough. `PluginOptions` raises on the
+framework-owned keys - and review of PR #16 then wrote a double-counting plugin
+three more ways, one of them touching nothing but `PRICE_OPTION_DEFAULTS`, a
+public dict in the module every plugin already imports. So `register()` now also
+**refuses at import any plugin whose own code mentions a framework key**
+(`code_leaks()`, docstrings stripped via `ast` so a plugin's mandated
+`SOURCES:` line does not false-positive). Honest wholesale baseline 0.188
+EUR/kWh; every attack landed 0.219.
+
+**The guarantee's precise shape**, stated because an overclaim is worse than no
+claim: the framework adds the complement exactly once *given a correct `sources`
+declaration*. A plugin that declares `{energy}` while its feed is really all-in
+still comes out one markup high while touching no framework name, and no code
+can catch that - the declaration is the plugin's unverifiable claim about
+somebody else's tariff. What is mechanised is reachability; what stays human is
+truthfulness.
+
+**The product type is declared, not inferred from the plugin's name.**
+`eneco_eddy` is a real-time/imbalance product and gets widened arbitrage
+thresholds; that reaches a consumer through `capabilities().product` and
+`volatility_regime()`, never through a name comparison. `test_price.py` asserts
+mechanically that no plugin id appears anywhere in the framework's own code,
+and drives a sixth provider defined inside the test file to the correct regime
+with `price.py` untouched.
+
+**"No price data" is a type, not an empty list.** Following the same rework
+`sim/riglock.py` took, a fetch returns `PriceRead.series(...)` or
+`PriceRead.no_data(reason)`, and only the first carries fields. An empty series
+is not constructible; `no_data` has no `.points` to iterate, no `.at()`, and
+`.get("points", [])` raises rather than returning the default. **A day of free
+electricity and a day with no prices are different instructions** - arbitrage is
+pointless under the first and forbidden under the second - and an implementation
+that returns `[]` on a failed fetch cannot tell them apart. Every site-level
+failure returns `no_data`; nothing raises into the control tick.
+
+**What is not verified, and cannot be by an offline suite:** every fixture is
+synthetic, so the five feed URLs and JSON shapes are checked against a belief
+about what those suppliers publish, not against the feeds. The parse layer is
+exercised; the contract with each supplier is not. Expect the first live fetch
+of each to need a shape fix, and note that a wrong shape degrades to
+no-price-data rather than to a wrong price.
+
+**`manual_csv` is the documented offline fallback** and needs no network, no
+supplier account and no HTTP stack. See DOCS.md for the format. Its `start`
+column must carry a UTC offset - a bare local timestamp is refused, because one
+hour every October names two moments.
+
+**Injection is priced separately** (`export_price_eur_kwh`), a scalar rather
+than a second series, and negative values are accepted. Consumption plugins
+cannot reach it.
+
+**`price_provider` may not be combined with `price_now_entity` /
+`price_avg_entity`** - the add-on refuses to start on that pair. The Home
+Assistant sensors are the existing delegation of pricing to HA and still work
+alone; with both configured they would silently keep deciding while the status
+page named a supplier that decided nothing.
+
+**No timezone database is used.** The shipped Alpine image has no `tzdata`, so
+Brussels offsets are computed from the EU rule and every instant is carried as
+UTC and rendered at the offset that applies to it. The checks drive both 2026
+transitions - the 23-hour and 25-hour days - rather than a normal day, where a
+fixed 24-slot implementation and a correct one agree.
+
+**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
@@ -581,6 +581,97 @@ unchanged is 12 W, which is inside the 15 W deadband — so the command there is
 parked at 0 W and the two halves of the rule cannot both be true. At the schema
 floor of 60 s that figure is 60 W and only the margin is left.

+### Energy price providers
+
+> **Nothing in 0.5.x acts on a price.** This release ships the *plumbing*: the
+> five provider plugins, the canonical all-in format they emit, and the config
+> that selects one. The arbitrage strategy that would use it is a later ticket.
+> Turning a provider on today fetches, validates and publishes prices and
+> changes no charging decision. That is deliberate — the seam lands first.
+
+| option | meaning |
+|---|---|
+| `price_provider` | `off` (default) or one of the five plugins below |
+| `price_markup_eur_kwh` | Your supplier's margin, EUR/kWh. **Added only when the chosen plugin's feed does not already contain it** |
+| `price_taxes_eur_kwh` | Levies and VAT, EUR/kWh. Same rule |
+| `export_price_eur_kwh` | The **injection** price. A separate scalar, not a series — see below |
+| `price_csv_path` | Where `manual_csv` reads its file |
+| `price_feed_url` | Override the plugin's default feed. Only a host the plugin itself declares is accepted |
+| `price_horizon_h` | How far ahead to ask for, hours |
+| `price_timeout_s` | Per-request timeout |
+
+**Which plugin sources what.** This is the one thing to get right, because
+adding your markup to a price that already contains it makes every future
+arbitrage decision wrong by that amount, in a direction nothing reports.
+
+| plugin | product | the published figure contains | the add-on adds |
+|---|---|---|---|
+| `generic_dayahead` | day-ahead | energy only (wholesale EPEX BE) | markup **and** taxes |
+| `engie_dynamic` | day-ahead | energy + Engie's markup | taxes |
+| `totalenergies_dynamic` | day-ahead | energy + TotalEnergies' markup | taxes |
+| `eneco_eddy` | **real-time / imbalance** | everything, all-in | nothing |
+| `manual_csv` | day-ahead | everything, all-in | nothing |
+
+You cannot get this wrong by editing the markup: each plugin *declares* what its
+feed contains and the add-on adds only the rest. Set a markup under a plugin
+that already carries one and it is **ignored, with a line in the log saying so**
+— never applied twice.
+
+That holds **given a correct declaration**, and it is worth knowing where the
+line is. What the add-on enforces mechanically is that a plugin cannot obtain
+the markup or apply it: the numbers are not reachable from plugin code, and a
+plugin whose code mentions them fails to load at all. What it cannot check is
+whether a plugin described its supplier's feed truthfully — that is a claim
+about somebody else's tariff, and no code can verify it. If you write a plugin,
+the `sources` declaration is the part a reviewer has to read.
+
+`eneco_eddy` is a real-time/imbalance product rather than a day-ahead one. It is
+flagged as high-volatility, which widens arbitrage thresholds once the
+optimiser lands. That comes from the plugin's own declaration, not from its
+name.
+
+**`manual_csv` — the offline fallback.** It needs no internet, no supplier
+account and no HTTP stack at all, and it is the supported way to run this
+add-on price-aware on a site with no outbound connectivity. Write a file and
+point `price_csv_path` at it:
+
+```csv
+start,eur_per_kwh,provisional
+2026-01-15T00:00:00+01:00,0.1834,false
+2026-01-15T01:00:00+01:00,-0.0210,true
+2026-01-15T02:00:00+01:00,0.3115,false
+```
+
+- `start` **must carry its UTC offset** (`+01:00` in winter, `+02:00` in
+  summer). A bare `2026-01-15 00:00` is refused, because for one hour every
+  October a local wall clock names two different moments.
+- `eur_per_kwh` is **all-in** — energy *plus* markup *plus* taxes. This plugin
+  has no way to know what your numbers already include, so it assumes
+  everything and adds nothing.
+- Negative prices are valid and are kept. They are the best charging hours of
+  the year.
+- `provisional` is optional; omit the column and every hour is treated as
+  settled.
+
+**When there are no prices at all** — provider off, feed unreachable, CSV
+missing — the add-on does not guess. It reports *no price data*, which is a
+distinct state from "prices, all of them zero": the plan is marked provisional
+and arbitrage is disabled, per the degradation matrix. It never falls back to a
+price of zero, which would read as free electricity.
+
+**Injection is priced separately** (`export_price_eur_kwh`). In Belgium it is
+often near zero, which by itself strongly favours self-consumption over export.
+It is a single number rather than a curve because no supported supplier
+publishes a dynamic injection tariff; negative values are accepted.
+
+**The old Home Assistant price sensors.** `price_now_entity` and
+`price_avg_entity` still work and are unchanged: the add-on compares them and
+uses the result as a cheap-window hint for the maintenance charge. They are
+**not compatible with `price_provider`** — configure both and the add-on
+refuses to start, naming the conflict, because the Home Assistant pair would
+silently keep deciding while the status page named your supplier. Use one or
+the other.
+
 ### Site

 | option | meaning |
@@ -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.
@@ -500,6 +525,14 @@ class Controller:
        # 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.

@@ -837,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:
@@ -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
@@ -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
@@ -116,6 +116,21 @@ NOT READY banner already names the problem, (b) does not apply.
     (13.8 kWh usable at 2000 W is ~6.9 h against the 720 min DRAIN_TIMEOUT).
     It is warned about below, not refused.

+  7. A price configuration that cannot mean what it says.  [(a)]
+     Three cases, all in `_price_problems()`: an unknown `price_provider` id
+     (refused rather than defaulted to 'off', per the §14 warning box);
+     `price_provider` set alongside `price_now_entity`/`price_avg_entity`,
+     which is TWO PRICE SOURCES ON ONE SITE where the Home Assistant pair
+     silently wins; and `manual_csv` with no `price_csv_path`, which turns the
+     documented offline fallback into a permanent no-price-data condition
+     reached by accident. Same shape as 1, 4 and 5 every time - a source the
+     operator chose, overridden by a layer they cannot see.
+
+     ⚠️ `price_provider: "off"` IS NOT REFUSED AND MUST NOT BECOME SO. §11.4
+     makes no-price-data a supported state, and §9.2 says a large share of the
+     installed base is on a fixed or static tariff. Refusing to start without a
+     price feed would lock those sites out of the add-on entirely.
+
 WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:

  - `integrator_max_w` above `max_w`. DOCS.md is right that the only thing the
@@ -141,6 +156,7 @@ WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:
 import logging

 from .p1 import is_enabled, validate_meter_options
+from .price import PROVIDERS

 _LOG = logging.getLogger("goodwe.config")

@@ -349,6 +365,9 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None) -> list[str]
            f"two numbers. Lower maintenance_charge_w to {max_w:.0f} W or less, "
            f"or raise max_w to at least {charge_w:.0f} W.")

+    # --- 7. two price sources on one site ------------------------------------
+    problems.extend(_price_problems(opts))
+
    # --- warnings: real, but not worth refusing over -------------------------
    integ = opts.get("integrator_max_w")
    integ_w = None if not integ else _number(opts, "integrator_max_w", None)
@@ -382,6 +401,75 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None) -> list[str]
    return problems


+def _price_problems(opts: dict) -> list[str]:
+    """Rule 7 (TARIFF-02): the price configuration, refused where it is untrue.
+
+    ⚠️ NOTE WHAT IS *NOT* HERE. `price_provider: "off"` is not refused and never
+    will be: §11.4 makes no-price-data a first-class supported state (static
+    profile, provisional plan, no arbitrage), and an add-on that will not start
+    without a price feed is an add-on that a Belgian fixed-tariff site - a large
+    share of the installed base, per §9.2 - cannot run at all.
+
+    Three refusals, all case (a) from this module's docstring: a source the
+    operator SET being overridden by a layer they cannot see.
+    """
+    problems: list[str] = []
+    pid = str(opts.get("price_provider", "") or "off").strip() or "off"
+
+    if pid != "off" and pid not in PROVIDERS:
+        problems.append(
+            f"price_provider is {pid!r}, which is not a shipped plugin. The "
+            f"ids that exist are: off, {', '.join(sorted(PROVIDERS))}. "
+            "Refused rather than defaulted, per the §14 warning box - a "
+            "silent fall back to 'off' would look exactly like a working "
+            "provider that happens never to find a cheap hour.")
+        return problems          # nothing below can be judged against an unknown id
+
+    # ⚠️ THE HA-SIDE OVERLAP THE PM TOLD US TO DECIDE ABOUT, DECIDED HERE.
+    # price_now_entity/price_avg_entity are the existing delegation of pricing
+    # to Home Assistant: main.py reduces them to ONE BOOLEAN, `cheap`, read only
+    # by the maintenance cheap-window charge floor. They are KEPT (TARIFF-03 and
+    # OPT-04 are what replace them, and neither has landed), but they may not
+    # run alongside a provider. With both configured the provider series drives
+    # nothing at all while the two entities quietly keep deciding, and the
+    # operator's evidence that they picked a supplier is a status line saying
+    # they picked a supplier. That is (a), and it is the "silently double up"
+    # the PM named on 2026-08-30.
+    # ⚠️ THIS REFUSAL'S REASON EXPIRES, AND THE MESSAGE BELOW MUST BE EDITED IN
+    # THE SAME COMMIT THAT LANDS THE FIRST CONSUMER (TARIFF-03 or OPT-04).
+    # Today the message says the provider "would decide nothing" - true, because
+    # nothing reads the series. Once something does, both sources really decide
+    # and this becomes a straightforward two-writer conflict; the refusal stays
+    # right and its explanation goes wrong, which is worse than either. There is
+    # a tripwire on exactly this in test_price.py section 12: it greps app/ for
+    # a `load_prices` consumer and goes RED the moment one appears. Update the
+    # message, not the check.
+    ha_price = [k for k in ("price_now_entity", "price_avg_entity")
+                if str(opts.get(k, "") or "").strip()]
+    if pid != "off" and ha_price:
+        problems.append(
+            f"price_provider is {pid!r} AND {' and '.join(ha_price)} "
+            f"{'is' if len(ha_price) == 1 else 'are'} set, which is two price "
+            "sources on one site. THE HOME ASSISTANT ENTITIES WIN: they are "
+            "what main.py reduces to the `cheap` flag the maintenance cycle "
+            "reads, and nothing yet consumes the provider series, so the "
+            "supplier you configured would decide nothing while the status "
+            "page named it. Clear price_now_entity and price_avg_entity to use "
+            f"the {pid!r} plugin, or set price_provider to 'off' to keep the "
+            "Home Assistant sensors.")
+
+    if pid == "manual_csv" and not str(opts.get("price_csv_path", "") or "").strip():
+        problems.append(
+            "price_provider is 'manual_csv' but price_csv_path is empty, so "
+            "the offline fallback has no file to read. Every fetch would "
+            "return no-price-data with a filesystem error as its reason - "
+            "which is §11.4 behaviour arrived at by accident rather than by "
+            "configuration. Point it at the CSV, or set price_provider to "
+            "'off', which says the same thing on purpose.")
+
+    return problems
+
+
 def _number(opts: dict, key, default):
    raw = opts.get(key, default)
    if raw is None or raw == "":
@@ -162,9 +162,53 @@ options:
  # --- tariff / capacity tariff (all optional) ------------------------------
  peak_forecast_entity: ""
  peak_cap_w: 3500
+
+  # ⚠️ THE OLD PRICE PATH, AND IT IS NOT DEAD YET. These two entities are the
+  # add-on delegating the whole price question to Home Assistant: main.py reads
+  # both and derives ONE BOOLEAN, `cheap = now <= avg`, which is consumed by
+  # nothing except the maintenance cycle's cheap-window charge floor. That is
+  # all they have ever done. They are kept, because retiring them would take
+  # the maintenance cycle's only price signal away on a live install before
+  # anything replaces it - TARIFF-03 (static profile) and OPT-04 (arbitrage)
+  # are the cards that replace it, and neither has landed.
+  #
+  # ⚠️ They may NOT be combined with price_provider below. Two price sources on
+  # one site, one of which silently wins, is exactly the "double up" the PM
+  # named; validate.py rule 7 refuses that configuration at startup rather than
+  # letting an operator believe the provider they configured is driving
+  # something. See app/validate.py.
  price_now_entity: ""
  price_avg_entity: ""

+  # --- price providers (TARIFF-02, specs §9.3 / §14 `tariffs:`) --------------
+  # `off` is the shipped default and means the §11.4 no-price-data row applies:
+  # static profile, plan provisional, NO ARBITRAGE. Nothing in 0.5.x consumes
+  # the series yet - OPT-04 is the consumer - so turning this on today fetches,
+  # validates and publishes prices and changes no energy decision.
+  #
+  # ⚠️ THESE DEFAULTS ARE MIRRORED ONCE, IN app/price.py's
+  # PRICE_OPTION_DEFAULTS, and test_price.py reads THIS FILE and asserts the two
+  # agree key for key. Do not add a third copy in main.py - CFG-03 is open
+  # because that has already happened three times.
+  #
+  # markup/taxes are the §14 `tariffs.energy_price.*` figures and are added by
+  # the FRAMEWORK, only for the components the chosen plugin declares its feed
+  # does not already contain. engie_dynamic, totalenergies_dynamic, eneco_eddy
+  # and manual_csv publish a figure that already carries the markup, so setting
+  # it under those is ignored - and logged as ignored, never applied twice.
+  price_provider: "off"
+  price_markup_eur_kwh: 0.03
+  price_taxes_eur_kwh: 0.06
+  price_csv_path: "/share/goodwe_prices.csv"
+  price_feed_url: ""
+  price_horizon_h: 48
+  price_timeout_s: 10
+
+  # §9.3: injection is priced SEPARATELY, and in Belgium it is often near zero -
+  # which by itself strongly favours self-consumption over export. A scalar, not
+  # a second series: no shipped supplier publishes a dynamic injection curve.
+  export_price_eur_kwh: 0.04
+
  # --- site ------------------------------------------------------------------
  estop_fitted: false
  log_level: info
@@ -239,5 +283,23 @@ schema:
  price_now_entity: str?
  price_avg_entity: str?

+  # ⚠️ The provider list is written out rather than left as `str`, so the
+  # Supervisor rejects a typo in the UI. validate.py ALSO refuses an unknown id
+  # against app/price.py's PROVIDERS registry, because sim/inject.py and the
+  # checks reach validate_startup_config() with a dict the Supervisor never saw.
+  # Adding a sixth plugin means adding it here and to PROVIDERS - and nowhere
+  # else, which is AC2.
+  price_provider: list(off|generic_dayahead|engie_dynamic|totalenergies_dynamic|eneco_eddy|manual_csv)
+  price_markup_eur_kwh: float(0,1)
+  price_taxes_eur_kwh: float(0,1)
+  price_csv_path: str?
+  price_feed_url: str?
+  price_horizon_h: int(1,72)
+  price_timeout_s: int(2,60)
+  # ⚠️ The lower bound is NEGATIVE on purpose. Belgian injection tariffs have
+  # gone below zero (you pay to export at midday in summer), and a schema that
+  # bottoms out at 0 would make that site inexpressible.
+  export_price_eur_kwh: float(-1,1)
+
  estop_fitted: bool
  log_level: list(trace|debug|info|warning|error)
@@ -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")
Author	SHA1	Message	Date
adminandClaude Opus 5	862f0fa574	Merge release/1.0 (TARIFF-01) into TARIFF-02 checks / unit (pull_request) Successful in 2m39s Details TARIFF-01 (#17) landed on release/1.0 first, so this branch was two commits behind and Gitea reported the PR unmergeable. Integrating rather than rebasing: the two reviewed commits stay addressable, and the branch is protected against force-push anyway. ONE CONFLICT, AND IT IS THE BORING HALF OF THE ONE REVIEW PREDICTED. Review of #16 expected collisions in four places - the tail of `options:`, the tail of `schema:`, the rule list inside validate_startup_config(), and the tail of validate.py. None of them happened: TARIFF-01 touched neither config.yaml nor validate.py. It added app/tariff.py, test_tariff.py, two P1 scenario captures, and edits to main.py and p1.py, none of which this branch goes near. The only conflict was CHANGELOG.md, where both branches prepended a block under the same `## Unreleased` heading. Both blocks kept, TARIFF-02 above TARIFF-01 since it lands second. Nothing was dropped or reworded on either side, and the resolution adds no line of its own beyond the blank one between them. Checked rather than assumed after the merge: - all ELEVEN test files pass, including TARIFF-01's own test_tariff.py (which this branch had never run) and the 93 s test_loop.py; - the 24-mutation battery still has 0 survivors; - no semantic overlap in config: TARIFF-01's peak tracker reads the peak_cap_w / peak_forecast_entity options that already existed, so the price options added here neither shadow nor duplicate anything. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC	2026-08-30 19:42:16 +02:00
adminandClaude Opus 5	e102cde841	TARIFF-02 review: close the three other doors to the markup, and stop overclaiming checks / unit (pull_request) Successful in 2m34s Details Review of PR #16 wrote a double-counting plugin four ways. PluginOptions blocked one. The one that decided it touched nothing but PRICE_OPTION_DEFAULTS - a public dict in the module every plugin must already import for register() - so "the number is not in scope" was simply false, and an absolute guarantee a reviewer trusts is worse than a stated limit, because they stop checking. register() now refuses, AT IMPORT, any plugin whose own code mentions a framework-owned name: FRAMEWORK_KEYS plus PRICE_OPTION_DEFAULTS, price_option and _o. code_leaks() parses the class and drops every docstring node before searching, because every plugin is REQUIRED to name its sourced components in prose - engie_dynamic says "markup" in its docstring and a raw text scan false-positives on the documentation the same file mandates. A check that fires on required documentation is a check somebody deletes. test_price.py asserts that a raw scan really would have false-positived, so the strip is proven load-bearing rather than asserted. It fails OPEN if inspect.getsource() cannot find the .py, which the add-on image always can. Refusing to start a battery controller because a static analysis could not run would trade a real outage for a hypothetical accounting error - so instead the checks assert source was found for all five shipped plugins, and a build that lost it is a red check rather than a quietly disarmed guard. All three reachable attacks are written out as plugins in the suite and each must be refused at registration. The cost is measured, not asserted: honest wholesale 0.188, every attack 0.219 - one markup high, every hour, invisible in every log this add-on writes. The guarantee is now stated at its true size everywhere it appears: the framework adds the complement exactly once, GIVEN A CORRECT sources DECLARATION. A plugin that declares {ENERGY} while its feed is all-in still doubles while touching no framework name, and no code can catch that - the declaration is an unverifiable claim about somebody else's tariff. Measured in the suite (0.276) so the limit is a number rather than a caveat. The docstring/declaration check verifies that a plugin's two claims AGREE, not that either is true; that now says so. The three surviving mutations: M3 - _NoData's reason taking a default. This is the identical survivor left open on RIG-01's _Unread.cause: the shape was borrowed from sim/riglock.py including its blind spot, which is what borrowing a shape does if nobody re-runs the mutations against the copy. Closed by SIGNATURE - no constructor or handler of an outcome type may carry a default - so the next one is red without anyone writing a new case. M5 - the ENERGY-required guard was untested. Its consequence is that adds becomes {ENERGY, TAXES} and the emitted price silently omits the energy it is pricing. M7 - deleting the TLS refusal went green AND the check named for it still passed, because the old parser stripped a fixed len("https://") and mangled http://api.energy-charts.info/p into the host pi.energy-charts.info, which then failed the ALLOW-LIST. An assertion that does not test what it names, inside a security check. Fixed on both sides: the scheme is parsed as a scheme, so an http URL now resolves to its real and declared host and only the TLS rule can refuse it; and the check asserts the REASON. Fixing it surfaced M7b - the no-scheme guard was survivable the same way - found by mutating my own fix, and closed the same way. Also carried forward: rule 7's refusal is correct forever but its stated reason ("the provider decides nothing") expires the moment TARIFF-03 or OPT-04 lands. test_price.py section 12 now greps app/ by AST for a load_prices consumer and goes red when one appears, telling whoever lands it to update the message in the same commit. The first draft of that tripwire used a substring and matched the comment in validate.py explaining the tripwire - the same false-positive class as the plugin docstrings, hit twice in one ticket, and fixed the same way. Documented rather than coded around: vars(read).get("points", []) and bool(read) both still get past the absence type, and neither is worth code - defining __bool__ would turn a line that reads like a guard into one that IS a guard by accident, which is how .get(k, default) got here in the first place. Stated plainly in DOCS/CHANGELOG/PR: every fixture is synthetic, so the five feed URLs and JSON shapes are verified against a belief about what those suppliers publish rather than against the feeds. A wrong shape degrades to no-price-data, not to a wrong price. Review's own DST verification recorded: 149,040 hourly instants, 2020-2036, zero mismatches against zoneinfo. The residual is political, not arithmetic, and the docstring now says so. 24 mutations driven through the suite, 0 survivors. Whole add-on suite green. No deleted lines outside this ticket's own files; main.py still untouched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC	2026-08-30 19:35:36 +02:00
admin	0dbd485b5a	TARIFF-01: measure the month's billed capacity peak, and make it survive a restart (#17 ) The instrument SAFETY-07 will read: the highest clock-aligned 15-minute average offtake of the calendar month, persisted in /data/tariff.db across two retention tiers, with a monthly peak that recompute_month() proves has not diverged from the 15-minute tier. Files no claim, applies no clamp, commands nothing. AC 1 was already built and unwired - p1.QuarterAverager on the base already does clock-aligned 900 s blocks, time-weighted over irregular spacing, offtake-only, per-phase, with the max_hold_s gap rule; verified on `9aadf46` rather than taken on the PR's word, and the p1.py diff removes zero lines. So this is persistence and reduction rather than a second source of truth for the quantity the money depends on. Block rather than rolling, and the reading is provable: a 900 s burst centred on a :15 boundary bills at 2000 W and reads 4000 W through a trailing window - exactly 2.00x, the supremum - and since the aligned windows are a subset of all trailing windows, a rolling figure can only ever read HIGH, never low. On the real captures, 1.09x and 1.05x. Reviewed independently: the encoded EU DST rule graded against zoneinfo at minute resolution over eight transitions, all of 2026 hourly and every month boundary - 0 mismatches - and the wrong-arm fromutc() mutation that differs only inside the one-hour transition window now goes red, which is the class that escaped the first draft. All four persistence behaviours produced rather than read, including a hard os._exit with no close() losing neither a banked quarter nor the open checkpoint. The SAFETY-20 syntax grader discriminates on names; two evasions remain in its import half. AC 3 is NOT met and the check says so: no 30-day dataset exists. The two days that do exist replay to 189/190 quarters = 99.5 % under the 3500 W soft limit, uncontrolled - the baseline SAFETY-07 must improve on - and 2026-08-20's peak reproduces independently at 5029.8 W, a billed quarter over the live install's own 5000 W rating. Do not close the card as AC 3 met; start the 30-day capture. Four reviewer mutations survive, recorded on the PR. The one to act on: record_block() ignoring its policy argument stays green, because every block reaching the store in the check is single-phase where NET, SUM and MAX collapse - the third instance of the degeneracy trap this file already documents twice, sitting on the exact line the pending per-phase product decision will change. No conflict with TARIFF-02 (#16) beyond CHANGELOG.md, verified from both sides.	2026-08-30 19:33:57 +02:00
glenn schrooyenandClaude Opus 5	f4914a174a	TARIFF-01: measure the month's billed peak, and make it survive a restart checks / unit (pull_request) Successful in 2m38s Details 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	2026-08-30 19:13:58 +02:00
adminandClaude Opus 5	852fc7ced6	TARIFF-02: one canonical price format, and a seam that keeps supplier names out of the optimiser checks / unit (pull_request) Successful in 2m33s Details Five provider plugins - generic_dayahead, engie_dynamic, totalenergies_dynamic, eneco_eddy, manual_csv - all emitting exactly {"prices": [{"start", "eur_per_kwh", "provisional"}]}, tz-aware at the Brussels offset, EUR/kWh, all-in. Nothing consumes it yet; the seam lands first, per the PM's ordering. It is built in goodwe_controller/app/ rather than at specs.md 2.2's ems/ingest/prices/, per the product owner's decision of 2026-08-30. The behaviour 9.3 / 8.3 / 11.4 specifies is unchanged. Three things shape the design, and each is the mechanism rather than a comment saying to be careful: Double-counting the markup is unreachable. A plugin does not return a price - it returns the figure its source published and declares which of {energy, markup, taxes} that figure already contains. assemble() adds the complement, once, in the framework. A plugin is never handed price_markup_eur_kwh: PluginOptions raises FrameworkOption on it, so the line that adds the margin a second time cannot be written. Engie and TotalEnergies publish a retail figure with their margin inside; adding the configured 3 c/kWh on top would have made every arbitrage decision wrong by that amount with nothing reporting it. The product type is declared, not inferred from the plugin's name. eneco_eddy is real-time/imbalance and gets 8.3's widened thresholds through capabilities().product and volatility_regime(), never through a name comparison - which is AC2. test_price.py asserts mechanically that no plugin id appears anywhere in the framework's code, and drives a sixth provider defined inside the test file to the right regime with price.py untouched. "No price data" is a type, not an empty list - the same rework sim/riglock.py took under TEST-04. A fetch returns PriceRead.series(...) or PriceRead.no_data(reason), and only the first carries fields. An empty series is not constructible; no_data has no .points, no .at(), and .get("points", []) raises rather than returning the default. A day of free electricity and a day with no prices are different instructions - arbitrage is pointless under one and forbidden under the other - and an implementation returning [] cannot tell them apart. Every site-level failure returns no_data; nothing raises into the control tick. Also: injection stays a separate scalar (export_price_eur_kwh, AC4), negative prices survive the boundary validator (17's regression case), egress is allow-listed against each plugin's own declaration (16), and no tzdata is used - the shipped Alpine image has none, so Brussels offsets come from the EU rule and instants are carried as UTC. validate.py grows rule 7: an unknown provider id, manual_csv with no path, and - the HA overlap the PM told us to decide about - price_provider set alongside price_now_entity/price_avg_entity, which is two price sources where the HA pair silently wins. The HA sensors are kept and still work alone; TARIFF-03 and OPT-04 are what replace them. CFG-03 is not widened: the price defaults have ONE in-code mirror, PRICE_OPTION_DEFAULTS, which validate.py imports rather than restating and which test_price.py pins to config.yaml by reading the file. test_price.py is runnable, assert-free in the house style, and every verdict it prints is derived. It grades the markup arithmetic with 0.031/0.057 rather than the shipped 0.03/0.06, because 0.06 is exactly twice 0.03 and a degenerate pair would make double-counting indistinguishable from correct addition - and it asserts that non-degeneracy as its first section. The DST checks run on the 23-hour and 25-hour days, not on a normal day where a fixed 24-slot implementation and a correct one agree. Thirteen mutations were driven through it; all thirteen go red. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC	2026-08-30 19:05:12 +02:00
admin	9aadf469bc	TEST-06: pin the cycle() ordering, the NOT COVERED string and the shipped meter_freeze_s (#14 ) Closes the three mutations that survived SAFETY-18 suite green. Ordering pinned by consequence (N8 red above the assignment, N3 green below it); NOT COVERED asserted in both directions; shipped meter_freeze_s and its schema bounds pinned. notes_of()/all_notes_of() split so this guard and the whole notification surface are separately nameable, with the three surface-wide claims decided per scenario after SAFETY-03. app/ change is comment-only (AST-identical).	2026-08-30 18:39:07 +02:00