goodwe-addon/goodwe_controller/test_tariff.py

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

------------------------------------------------------------------------------
⚠️ PART 9 IS A TWO-DAY MEASUREMENT AND SAYS SO. THE 30-DAY ANSWER IS ELSEWHERE
(DATA-01), AND IT DOES NOT LAND WHERE ANYONE EXPECTED.

AC 3 asks for "the monthly peak held under the soft limit in >= 95 % of
intervals over a 30-day simulation replay". When this file was written there
were 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 THE SENTENCE THAT USED TO STAND HERE, "nothing longer was ever
captured", WAS TRUE ABOUT THE PAST AND WAS READ AS A CLAIM ABOUT THE FUTURE.
It was not one. HA's recorder still held the whole month; DATA-01 probed it on
2026-08-31 (230 samples in a 20-minute window at 30 days back, 0 at 45) and
exported thirty days the same day, into `scenarios/p1-net-30d-2026-08/` with
`tools/export_p1_history.py` - which is committed now precisely because the
2026-08-25 one-off was not, and its absence is what let the belief survive.

Over those thirty days the criterion measures 98.89 % UNCONTROLLED and 98.99 %
guarded, so it is met - and that is a finding about the CRITERION as much as
about the guard, because a month with no controller at all passes it. Read the
other way - the monthly PEAK itself under the soft limit - it is not met: the
guard takes August's peak from 5029.8 W to 4107.3 W, still over 3500 W. Both
numbers, and why they part company, are in
`scenarios/p1-net-30d-2026-08/README.md`.

PART 9 IS DELIBERATELY LEFT GRADING THE TWO COMMITTED DAYS. It is a per-PR
regression check and the thirty days cost about fifty seconds of replay; they
also live one directory down precisely so this file's glob does not pick them
up. `tools/replay_ac3.py` is the thing that runs the month, and it is checked
against part 6 of test_peakguard.py rather than trusted:

    python3 tools/replay_ac3.py scenarios --expect 189/190
    python3 tools/replay_ac3.py scenarios/p1-net-30d-2026-08

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.

The recorder's window rolls: 2026-08-01 leaves it around 2026-09-15, which is
why the committed copy is the artefact and not the API call.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

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

# ⚠️ NOT config.yaml's shipped max_w (2000). NOTES.md and freeze.py both record
# that the LIVE INSTALL runs max_w 5000, and that is the number this file uses
# as the device's power ceiling wherever it needs one - it is named
# LIVE_INSTALL_MAX_W, not the shorter LIVE_MAX_W, precisely so it cannot be
# misread as "the current shipped default" (TEST-10; test_freeze.py carries the identical
# constant, under the identical name, for the identical reason - grep it there
# for the pin that stops config.yaml's max_w from drifting under it unnoticed).
# `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_INSTALL_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_INSTALL_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_INSTALL_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_INSTALL_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_INSTALL_MAX_W)
empty_r = good_tr.peak(B)

gone = fresh_store("gone")
gone_tr = PeakTracker(gone, cap_w=LIVE_INSTALL_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_INSTALL_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_INSTALL_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_INSTALL_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_INSTALL_MAX_W:.0f} W, "
      f"which is correct and is a different answer",
      near(good_tr.headroom_w(B), LIVE_INSTALL_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_INSTALL_MAX_W * 0.9, None))
print(f"    after one {LIVE_INSTALL_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_INSTALL_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_INSTALL_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))

# --- TEST-13 item 1: the POLICY ARGUMENT'S PLUMBING, not just billed_w() ------
# `PeakStore.record_block(block, policy)` really does call `billed_w(block,
# policy)` - the product code is correct and is not what this adds. What was
# missing is that NO CHECK COULD TELL that call apart from a hardcoded
# `billed_w(block, POLICY_NET)`. Mutating it to exactly that left all 17 suites
# green, twice measured (TARIFF-01's reviewer, re-run through the purge harness
# for TEST-15 and again here).
#
# The reason is this section's own subject, one level up. `billed_w()`
# short-circuits to `block.offtake_avg_w` whenever `per_phase_offtake_avg_w` is
# absent, and EVERY block that reached a store anywhere in this file was
# single-phase or built with `per_phase_offtake_avg_w=None` - so all three
# policies collapsed onto one number and the argument was unobservable at the
# call site. The checks above grade the FUNCTION on an unbalanced fixture; they
# say nothing about whether its CALLER forwards what it was handed. That is the
# same shape as WITNESS-02's span normalisation and TARIFF-01's own DST window:
# a stimulus evaluated where the quantity it grades is degenerate.
#
# So the stimulus here is `blk` - the unbalanced three-phase block from the top
# of this section, whose three reductions were asserted three lines up to be
# three DIFFERENT numbers. A `record_block()` that ignores its argument stores
# one policy's figure under all three names and at least two of these
# round-trips go red.
#
# Which direction the defect runs in, because that is what makes it worth
# plumbing-testing at all: NET is the SMALLEST of the three on this fixture (the
# exporting phase nets off against the other two). A store that silently billed
# NET while its caller asked for MAX would under-record the month's peak, and an
# under-recorded peak is maximum headroom for SAFETY-07 - the same
# under-guarding direction as item 4's `getattr(..., 0.0)` default.
_banked = {}
for _pol in (POLICY_NET, POLICY_SUM, POLICY_MAX):
    _st = fresh_store(f"policy-{_pol}")
    _st.record_block(blk, _pol)
    _read = _st.peak(month=month_key(blk.start))
    _derived = _st.recompute_month(month_key(blk.start))
    _banked[_pol] = _read.peak_w
    check(f"record_block(block, {_pol!r}) banks {vals[_pol]:.0f} W - the store "
          f"applies the reduction it was HANDED, not one of its own",
          _read.is_observed and near(_read.peak_w, vals[_pol]))
    check(f"...and re-deriving from the 15-min tier gives the same {_pol!r} "
          f"figure, so BOTH writes in that one transaction used that reduction",
          _derived.is_observed and near(_derived.peak_w, vals[_pol]))

_st_def = fresh_store("policy-default")
_st_def.record_block(blk)
check(f"...and record_block() with NO policy argument banks the NET figure "
      f"({vals[POLICY_NET]:.0f} W), matching billed_w()'s own shipped default",
      near(_st_def.peak(month=month_key(blk.start)).peak_w, vals[POLICY_NET]))
# The anti-degeneracy pin, stated over the BANKED numbers rather than over
# `vals`, because it is the banked ones this triple is made of. If a future
# fixture change ever made two of these coincide, the triple above would start
# passing under a hardcoded policy and would be silently back where it started.
print(f"    banked through the store: "
      + ", ".join(f"{k}={v:.0f} W" for k, v in _banked.items()))
check("the three BANKED figures are three distinct numbers, so this triple is "
      "not itself degenerate - it can only pass if the argument is read",
      len({round(v, 6) for v in _banked.values() if v is not None}) == 3)


# ==============================================================================
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_INSTALL_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_INSTALL_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 and this run is {replay_days:.1f}, so it "
      f"does not settle it either way. The 30 days DO now exist - "
      f"scenarios/p1-net-30d-2026-08/, DATA-01 - and "
      f"tools/replay_ac3.py is what runs them; the figure above is the "
      f"BASELINE SAFETY-07 has to improve, on the two days CI replays.")
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_INSTALL_MAX_W, LIVE_INSTALL_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_INSTALL_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))

# ⚠️ `when=B`, AND THAT ARGUMENT IS THE TICKET (TEST-17). Without it `checks()`
# reads the peak for `datetime.now()`'s month while this fixture banked its
# quarter in August, so every check below passed in August and failed from
# 2026-09-01 - on a commit nobody had touched. The suite must grade the month
# it created, not the month the machine happens to be running in.
rows = [r for r in c.checks(when=B) 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(when=B) 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"])

# --- TEST-17: the page must be graded at a stated time, and prove it ----------
# Three ways this can regress, so three checks. The first two would both go
# green again if `checks()` quietly went back to reading the clock, which is
# exactly the mutation this section could not previously detect.

# ⚠️ SELECTED ON "quarter(s) banked", NOT ON "peak". The NO_RECORD row reads
# "2026-10: no completed quarter recorded yet - ..." and contains neither the
# word "peak" nor "capacity-tariff" - only the UNREADABLE branch prefixes the
# family name. Filtering these rows the way the checks above do would drop the
# row and then grade its absence, which is the very mistake this ticket fixes.
# (That asymmetry in the row's own text is worth a look on its own; it is what
# made the original filter brittle. Not changed here - it is product text with
# other suites reading it.)
_peak_rows = lambda rows: [r for r in rows if "quarter(s) banked" in r["text"]]

_later = B + timedelta(days=45)          # same fixture, a later billing month
_other_peak = _peak_rows(c.checks(when=_later))
print(f"    the SAME controller read at {_later.date()}: "
      + (_other_peak[0]["text"] if _other_peak else "(no peak row)"))
check("asked for a LATER month, the same controller reports that month's "
      "absence of a quarter, not August's 4998 W - the row follows `when`",
      bool(_other_peak) and "4998 W" not in _other_peak[0]["text"]
      and "no completed quarter recorded yet" in _other_peak[0]["text"])
check("...and that is NO_RECORD, distinguishable from BOTH an observed peak and "
      "an unreadable store - a month with no quarter is not a broken instrument",
      bool(_other_peak) and bool(peak_rows)
      and "UNREADABLE" not in _other_peak[0]["text"]
      and _other_peak[0]["text"] != peak_rows[0]["text"])
check("...and it is not FLAGGED, because an empty month is not an exceedance",
      bool(_other_peak) and _other_peak[0]["ok"] and not _other_peak[0]["warn"])

# The mutation guard. `checks()` ignoring `when` is the whole defect, and it is
# invisible to every check above whenever the machine's clock happens to sit in
# the fixture's own month - which is how this shipped. Grading two DIFFERENT
# reference months against each other cannot be satisfied by a clock at all.
check("the peak row differs between two reference months, so a `checks()` that "
      "dropped `when` and read datetime.now() would fail HERE regardless of "
      "what day the suite is run on",
      bool(peak_rows) and bool(_other_peak)
      and peak_rows[0]["text"] != _other_peak[0]["text"])

# And the production meaning is unchanged: no argument still means today.
_now_peak = _peak_rows(c.checks())
_today_is_fixture_month = month_key(datetime.now(timezone.utc)) == month_key(B)
check("calling checks() with NO argument still reads the month being billed "
      "today - the live status page's meaning is untouched",
      bool(_now_peak) and (bool(peak_rows)
                           and _now_peak[0]["text"] == peak_rows[0]["text"]
                           if _today_is_fixture_month
                           else "no completed quarter recorded yet"
                                in _now_peak[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)

# ⚠️ THE IMPORT HALF WALKS `Import` TOO, AND READS THE IMPORTED NAMES - TEST-13
# item 2. It used to be `m.module for m in walk(_tree) if isinstance(m,
# ImportFrom)`, matched against a literal list of six spellings, and that is
# blind to TWO OF PYTHON'S THREE import forms:
#
#   * `from . import arbiter as _ab` is an `ImportFrom` whose `.module` is
#     **None** - the dots live in `.level` and the module name lives in
#     `.names` - so the old grader collected `""` and matched nothing;
#   * `import app.arbiter` is an `ast.Import` node, a different class the walk
#     filtered out entirely and therefore never looked at.
#
# Neither was a live exposure: the `_names` half above catches any USE of either
# alias, because reaching anything through it produces an `Attribute` whose
# `.attr` is a forbidden name. The defect was in the SENTENCE. A check that
# prints "it imports nothing from arbiter, clamp or control" while verifying
# that for one spelling out of three is worse than one that claims less, because
# the next reader stops looking. So: collect every module any import statement
# names, in all three forms, and match per dotted SEGMENT - `app.arbiter` and
# `.arbiter` must not be able to hide behind a package prefix either.
_FORBIDDEN_MODULES = {"arbiter", "clamp", "control"}


def _imported_modules(tree):
    """Every module path an import statement in `tree` names, all three forms."""
    out = set()
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Import):
            # `import app.arbiter`, `import app.arbiter as x` - the whole
            # dotted path is the module.
            out |= {a.name for a in node.names}
        elif isinstance(node, _ast.ImportFrom):
            base = node.module or ""      # None for `from . import arbiter`
            if base:
                out.add(base)             # `from .arbiter import resolve`
            # ...and in the `module is None` form the MODULE is in .names, so
            # the names have to be joined on whether or not there is a base.
            out |= {f"{base}.{a.name}" if base else a.name for a in node.names}
    return out


def _forbidden_imports(mods):
    # Segment-wise, not substring: `arbitrage` contains no forbidden segment and
    # must not be flagged, while `app.arbiter` must be.
    return sorted(m for m in mods
                  if _FORBIDDEN_MODULES & {p for p in m.split(".") if p})


_imports = sorted(_imported_modules(_tree))
print(f"    tariff.py evaluates {len(_names)} distinct names and imports "
      f"{sorted({m.split('.')[0] for m in _imports}) or ['(nothing)']}")
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 in ANY of the "
      "three import spellings: it files no claim, applies no clamp, and "
      "commands nothing - SAFETY-07's job",
      not _forbidden_imports(_imports))

# The import grader graded. Every one of these is a real way to reach the
# command side from this module, and the first two are the ones the previous
# version let through; the last two it already caught and must keep catching.
_EVASIONS = ["from .arbiter import resolve",
             "from . import arbiter as _ab",
             "import app.arbiter",
             "from app.control import compute",
             "import app.clamp as _c"]
_missed = [e for e in _EVASIONS
           if not _forbidden_imports(_imported_modules(_ast.parse(e)))]
check(f"...and that import grader is not spelling-blind: all {len(_EVASIONS)} "
      f"ways of importing the command side are caught"
      + (f"; MISSED {_missed}" if _missed else ""), not _missed)
# ...and it is not simply flagging everything, which would make the line above
# free. The last three are the reason the match is per SEGMENT and not `f in m`:
# `clamps`, `arbiters` and `controller` all CONTAIN a forbidden word and none of
# them is the command side. `controller` is not hypothetical - this package is
# `goodwe_controller`, and a substring grader pointed at a path rather than a
# module name would refuse the whole repository.
_INNOCENT = ["import json", "from datetime import datetime, timezone",
             "from . import arbitrage", "from .price import Uncertainty",
             "from zoneinfo import ZoneInfo",
             "import clamps", "from .arbiters import x",
             "from . import controller"]
_false = [i for i in _INNOCENT
          if _forbidden_imports(_imported_modules(_ast.parse(i)))]
check("...and it does not fire on innocent imports whose names merely CONTAIN a "
      "forbidden word (`clamps`, `arbiters`, `controller`) - a substring match "
      "would have"
      + (f"; FALSELY FLAGGED {_false}" if _false else ""), not _false)

# ...and neither half is vacuous: the same two graders run over main.py, which
# DOES command, must both light up. main.py carries all three import forms.
_main_tree = _ast.parse((HERE / "app" / "main.py").read_text(encoding="utf-8"))
_main = {n.attr for n in _ast.walk(_main_tree) if isinstance(n, _ast.Attribute)}
_main_imports = _forbidden_imports(_imported_modules(_main_tree))
check(f"the same NAME grader over main.py finds {sorted(_main & _FORBIDDEN)}, "
      f"so it is capable of failing", bool(_main & _FORBIDDEN))
check(f"...and the same IMPORT grader over main.py finds {_main_imports}, so "
      f"it is capable of failing on a real module and not only on a fixture",
      bool(_main_imports))


# ==============================================================================
print()
shutil.rmtree(TMP, ignore_errors=True)
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")