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

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

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

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

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

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

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

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

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

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

goodwe_controller/app/p1.py

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