goodwe-addon/goodwe_controller/app/tariff.py

"""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