goodwe-addon/goodwe_controller/test_peakguard.py

"""Runnable check for SAFETY-07, the quarter-peak projection guard.

    python3 test_peakguard.py

Eight parts. What each is for, and - where it matters - what it is deliberately
NOT allowed to be tested at:

  1. THE CONTROL LAW, `excess_w`, graded AWAY from the two points where it is
     degenerate. At the start of a quarter the 900/remaining gain is exactly 1.0
     and `excess` collapses onto the plain difference `projected - ceiling`; a
     check placed only there cannot tell the shipped law from that difference.
     So part 1 first shows them coinciding (documenting the trap), then separates
     them, then shows the two properties the whole design rests on: the law is
     ZERO at the engagement boundary (so the claim is continuous and needs no
     hysteresis constant) and its gain GROWS through the quarter (so the guard is
     gentlest where its evidence is weakest).
  2. THE CLAIM, through the REAL `arbiter.resolve()`. That it binds the loop and
     the maintenance charge shaping, that it does NOT bind a maintenance phase
     that owns the actuator or any safety stop, that it can never cap discharge
     or export, and - the load-bearing one - that it can never produce the
     `lo > hi` contradiction the arbiter resolves to 0 W.
  3. THE CEILING, AND A FAILED READ. Only an `OBSERVED` reading may raise it.
     `NO_RECORD` and `NOT_OBSERVED` both leave it at the cap, so an unreadable
     store makes the guard STRICTER. Graded against a REAL `PeakStore` that
     genuinely cannot open, not a stub - and the acceptance half is that the
     guard still ENGAGES with that store, because "fail open on an unreadable
     peak" is the one bug that would leave every other check in this file green.
  4. BLINDNESS, all three kinds, each one evaluated exactly where the number it
     replaces is degenerate. A fresh averager projects 0 W, which is under every
     ceiling there is; a frozen meter projects confidently; a meter gap projects
     LOW. None of the three may be graded as a number.
  5. `project()`'s DEFAULTED READ - TEST-13 item 4, fixed here because this card
     is the consumer that would have inherited it. The old form failed quiet AND
     in the disarming direction; part 5 computes what it would have concluded on
     the same fixture the shipped form engages on.
  6. THE REPLAY. 2026-08-20's 5029.8 W billed quarter, closed-loop, and the
     healthy day's baseline. Every figure re-derived from the committed capture.
  7. THE WIRING, through the real `Controller.cycle()` and its real status page.
  8. MUTATION. Six one-line defeats of this guard, each run against the checks
     above to show they are not vacuous.

------------------------------------------------------------------------------
WHAT THE REPLAY IS AND IS NOT

The captures carry ONE series: the signed net power at the meter
(`{"t": iso8601, "w": signed_watts}`). They do not carry battery power, so the
replay cannot reconstruct how much of 2026-08-20's offtake was the house and how
much was an inverter. The model is therefore stated rather than hidden:

  * the captured `w` is treated as the load presented to the meter, and
  * the guard's battery is modelled as an ADDITIONAL discharge subtracted from
    it, delivered on the next telegram (a one-sample actuation delay), and
  * when the guard is idle the modelled battery delivers 0 W.

That last line is deliberately PESSIMISTIC and it is what makes the number mean
something. The real grid-following loop would itself have been discharging on
2026-08-20 - it was not, because that is the day the vendor controller failed -
so crediting the guard with zero baseline discharge isolates SAFETY-07's own
contribution instead of flattering it with the loop's.

Two days is also not thirty. AC 3's 95 %-over-30-days criterion is TARIFF-01's
and is still not met; `test_tariff.py`'s part 9 says so at length and this file
does not re-litigate it. What part 6 reports is the comparison that two days CAN
support: the same quarters, with and without the guard.
"""

import asyncio
import ast
import gzip
import json
import re
import shutil
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from math import inf, isinf
from pathlib import Path

from app import peakguard
from app.arbiter import (
    P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, Claim, resolve,
)
from app.freeze import FreezeWatchdogConfig, looks_frozen, rail_w
from app.main import Controller
from app.p1 import (
    PLAUSIBLE_MAX_W, P1Ingest, QUARTER_S, QuarterAverager, make_sample,
    split_signed,
)
from app.peakguard import (
    BLIND_COVERAGE, BLIND_FROZEN, BLIND_NO_BLOCK, CEILING_CAP,
    CEILING_MONTH_PEAK, PeakDecision, blindness, decide, excess_w,
    quarter_ceiling_w, uncovered_s,
)
from app.tariff import (
    NOT_OBSERVED, NO_RECORD, OBSERVED, PeakReading, PeakStore, PeakTracker,
    month_key,
)

fails = []
HERE = Path(__file__).resolve().parent
TMP = Path(tempfile.mkdtemp(prefix="goodwe-peakguard-"))


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 - test_tariff.py's helper,
    for its reason: the sign convention under test must be the one that ships."""
    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"))


def read_option(key: str):
    """The shipped default for `key`, out of config.yaml.

    ⚠️ `peak_cap_w:` appears TWICE - once under `options:`, once under `schema:`.
    test_tariff.py's part 9 documents the trap; this takes the FIRST match, which
    is the options block, i.e. the number an install actually lands on.
    """
    for line in (HERE / "config.yaml").read_text(encoding="utf-8").splitlines():
        if line.strip().startswith(f"{key}:"):
            try:
                return float(line.split(":", 1)[1].strip())
            except ValueError:
                continue
    return None


def read_schema_max(key: str):
    """The schema's upper bound for `key` - `int(lo,hi)` -> hi."""
    body = (HERE / "config.yaml").read_text(encoding="utf-8")
    marker = body.index("\nschema:")
    for line in body[marker:].splitlines():
        st = line.strip()
        if st.startswith(f"{key}:") and "(" in st and ")" in st:
            return float(st[st.index("(") + 1:st.rindex(")")].split(",")[1])
    return None


# Everything numeric below is read out of the shipped file, not typed here.
CAP_W = read_option("peak_cap_w")
MAX_HOLD_S = read_option("meter_max_age_s")
FREEZE_S = read_option("meter_freeze_s")
CYCLE_S = read_option("cycle_floor_s")
# The live install runs max_w 5000 (NOTES.md, freeze.py). That is also the schema
# ceiling and the GW5000S-BP's rating, which is why the two agree - so the
# device figure is taken from the schema rather than typed.
LIVE_MAX_W = read_schema_max("max_w")

print("SAFETY-07 - the quarter-peak projection guard")
print(f"    read from config.yaml: peak_cap_w {CAP_W:.0f} W, "
      f"meter_max_age_s {MAX_HOLD_S:.0f} s, meter_freeze_s {FREEZE_S:.0f} s, "
      f"cycle_floor_s {CYCLE_S:.0f} s, max_w schema ceiling {LIVE_MAX_W:.0f} W")
check("every threshold this check grades against came out of config.yaml",
      None not in (CAP_W, MAX_HOLD_S, FREEZE_S, CYCLE_S, LIVE_MAX_W))

B = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)   # a quarter boundary


def averager_at(net_w, seconds, *, start=B, step=5.0, gap_at=None, gap_s=0.0):
    """A real `QuarterAverager` driven `seconds` into the block at `net_w`.

    Optionally with a real hole in the telegram stream, which is how a coverage
    shortfall is produced here: manufactured by NOT sending telegrams, exactly as
    a dead meter produces it, rather than by writing to a private field.
    """
    a = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
    t = 0.0
    while t <= seconds:
        if gap_at is not None and gap_at <= t < gap_at + gap_s:
            t += step
            continue
        a.add(sample(start + timedelta(seconds=t), net_w))
        t += step
    return a


# ==============================================================================
print()
print("1. the control law, and the point where it is indistinguishable")

# THE DEGENERACY, NAMED FIRST. At elapsed 0 the gain 900/remaining is exactly
# 1.0, so `excess_w` and the naive difference `projected - ceiling` are the same
# number. Anything graded only here separates neither.
CEIL = CAP_W
top = averager_at(5000.0, 0.0)
p_top = PeakTracker.project(top, B)
e_top = excess_w(p_top, CEIL)
naive_top = p_top.projected_w - CEIL
print(f"    at the top of the quarter: projected {p_top.projected_w:.0f} W, "
      f"excess {e_top:.1f} W, naive difference {naive_top:.1f} W")
check("at elapsed 0 the shipped law and the plain difference coincide - so a "
      "check placed only there proves nothing about either",
      near(e_top, naive_top, 1e-6))

# Away from it they must separate, and the direction is the design: the later
# the guard engages, the harder it has to pull to recover the same average.
for el in (300.0, 600.0, 800.0):
    a = averager_at(5000.0, el)
    p = PeakTracker.project(a, B + timedelta(seconds=el))
    e = excess_w(p, CEIL)
    gain = QUARTER_S / (QUARTER_S - p.elapsed_s)
    print(f"    at {el:.0f} s elapsed: projected {p.projected_w:.0f} W, "
          f"excess {e:.0f} W, gain {gain:.2f}x")
    check(f"at {el:.0f} s the law is the difference times {gain:.2f} - the "
          f"900/remaining gain, not a constant offset",
          near(e, (p.projected_w - CEIL) * gain, 1e-6) and e > (p.projected_w - CEIL))

a300 = averager_at(5000.0, 300.0)
a800 = averager_at(5000.0, 800.0)
e300 = excess_w(PeakTracker.project(a300, B + timedelta(seconds=300)), CEIL)
e800 = excess_w(PeakTracker.project(a800, B + timedelta(seconds=800)), CEIL)
check("...and it grows monotonically through the quarter, so the guard is at "
      "its GENTLEST when its projection rests on the least evidence",
      e800 > e300 > 0.0)

# CONTINUITY AT THE BOUNDARY. This is what stands in for §10.2's hysteresis.
BATT_NOW = 400.0
a_edge = averager_at(CEIL, 450.0)
p_edge = PeakTracker.project(a_edge, B + timedelta(seconds=450))
d_edge = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
e_edge = excess_w(p_edge, CEIL)
print(f"    held exactly AT the ceiling for 450 s: projected "
      f"{p_edge.projected_w:.2f} W, excess {e_edge:.3f} W, "
      f"floor {'(none)' if d_edge.floor_w is None else f'{d_edge.floor_w:.1f} W'}, "
      f"battery measured {BATT_NOW:.0f} W")
check("at the engagement boundary the excess is zero, so the floor is exactly "
      "the battery's own measured power - i.e. 'carry on'. The law is CONTINUOUS "
      "and there is no step for a hysteresis constant to damp",
      near(e_edge, 0.0, 1e-6) and near(d_edge.floor_w, BATT_NOW, 1e-6))
# THE OSCILLATION, and it is the reason the threshold is on the floor rather
# than on the excess. Re-derived here rather than described: a guard that
# released as soon as `excess <= 0` hands the actuator back at the exact instant
# it is working, and the load springs straight back.
released = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
would_release_on_excess = excess_w(p_edge, CEIL) <= 0.0
print(f"    at the boundary: excess {e_edge:.3f} W (an `excess > 0` test would "
      f"RELEASE: {would_release_on_excess}), floor {released.floor_w:.0f} W "
      f"(the shipped `floor > 0` test holds: {released.engaged})")
check("...and the claim is held there, not released. `excess` falls to zero as "
      "soon as the battery is already taking the watts, so thresholding IT is "
      "the bang-bang §10.2 forbids; thresholding the FLOOR is the same "
      "arithmetic read as a level", would_release_on_excess and released.engaged)
d_gone = decide(PeakTracker.project(averager_at(500.0, 450.0),
                                    B + timedelta(seconds=450)),
                batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
check("...and it DOES release once the load falls under what the quarter can "
      "afford, so 'threshold the floor' is not 'never let go'",
      not d_gone.engaged and d_gone.floor_w is None)

# A hair over, and the floor is the battery plus the excess. Not a constant.
a_over = averager_at(CEIL + 200.0, 450.0)
p_over = PeakTracker.project(a_over, B + timedelta(seconds=450))
d_over = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
print(f"    200 W over for 450 s: {d_over.explain()}")
check("a quarter projected over the ceiling engages, and the floor is the "
      "MEASURED battery power plus the derived excess",
      d_over.engaged
      and near(d_over.floor_w, BATT_NOW + excess_w(p_over, CEIL), 1e-6))
check("...and the floor moves with the battery measurement, so it is a claim "
      "about delivered power rather than a fixed setpoint",
      not near(decide(p_over, batt_w=BATT_NOW + 250.0, cap_w=CAP_W, reading=None,
                      achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S).floor_w,
               d_over.floor_w, 1.0))

# THE WIND-UP ARGUMENT, made concrete. A floor built on INTENT re-adds its own
# unmet ask every cycle; one built on delivered power does not.
stuck_batt = 0.0            # the inverter is not following
intent = 0.0
delivered_floor = []
intent_floor = []
for _ in range(5):
    d_del = decide(p_over, batt_w=stuck_batt, cap_w=CAP_W, reading=None,
                   achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
    delivered_floor.append(d_del.floor_w)
    intent = intent + excess_w(p_over, CEIL)     # what the intent form would do
    intent_floor.append(intent)
print(f"    five cycles with the inverter not following: delivered-power floor "
      f"{[f'{v:.0f}' for v in delivered_floor]}, intent floor "
      f"{[f'{v:.0f}' for v in intent_floor]}")
check("a floor sized from DELIVERED battery power is stationary while the "
      "inverter fails to follow; one sized from the standing command winds up",
      len(set(f"{v:.3f}" for v in delivered_floor)) == 1
      and intent_floor[-1] > intent_floor[0] * 4)


# ==============================================================================
print()
print("2. the claim, through the real arbiter")

SAFE = [Claim.limit("device", P_SAFETY, -LIVE_MAX_W, LIVE_MAX_W, "rating"),
        Claim.limit("supervised", P_SAFETY, -LIVE_MAX_W, LIVE_MAX_W, "max_w")]
pc = peakguard.claim(d_over)
print(f"    claim: {pc.source} {pc.kind} lo={pc.lo:.0f} hi={pc.hi} "
      f"priority={pc.priority}")
check("it is a LIMIT at P_PEAK, not a SET - so no safety stop can be dragged "
      "off zero by it", pc.kind == "limit" and pc.priority == P_PEAK)
check("...with only a LOWER bound. `hi` is +inf, which is §9.1's 'discharge and "
      "export are never capped by the peak guard' as a property of the object "
      "rather than a rule to remember", isinf(pc.hi) and pc.hi > 0)

# It binds the loop.
LOOP_ASK = -1500.0          # the loop wants to CHARGE at 1500 W
r = resolve(SAFE + [Claim.set("loop", P_LOOP, LOOP_ASK, "grid"), pc])
print(f"    loop asks {LOOP_ASK:+.0f} W (charging): {r.explain()}")
check("the floor binds the grid-following loop and turns a charge into the "
      "commanded discharge", near(r.target_w, pc.lo, 1e-9) and r.target_w > 0)
check("...and the arbiter names the peak claim as what bound it, so an operator "
      "reading one line knows why the battery is discharging",
      r.bound_by is not None and "peak" in r.bound_by)

# TWO LIMITS THAT CROSS. This is the defect this check was written to assume
# away and instead found: a LIMIT does not out-vote another LIMIT. Clause 2 takes
# the MOST RESTRICTIVE of every limit at or above the winner's priority, with no
# priority between them - which is what makes the safety bounds unconditional.
r2 = resolve(SAFE + [Claim.set("loop", P_LOOP, LOOP_ASK, "grid"),
                     Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0,
                                 reason="charge-only"), pc])
print(f"    peak floor AND maintenance charge-only shaping both filed: "
      f"{r2.explain()}")
check("filing both is a CONTRADICTION, not a P_PEAK win - `lo > hi` resolves to "
      "0 W, i.e. the battery parked in the one quarter it should be shaving. "
      "P_PEAK 50 > P_MAINT_SHAPE 40 does NOT save it",
      r2.contradiction and near(r2.target_w, 0.0))
check("...so the rule for two limits that can cross is SUPPRESSION, and "
      "removing the shaping gives the floor, which is what main.py now does",
      near(resolve(SAFE + [Claim.set("loop", P_LOOP, LOOP_ASK, "grid"), pc]
                   ).target_w, pc.lo, 1e-9))

# It does NOT bind a maintenance phase that owns the actuator, nor any stop.
r3 = resolve(SAFE + [Claim.set("maintenance", P_MAINTENANCE, 0.0, "hold"), pc])
r4 = resolve(SAFE + [Claim.set("safety", P_SAFETY, 0.0, "stopped"), pc])
print(f"    maintenance HOLD owns the setpoint: {r3.explain()}   |   "
      f"control stopped: {r4.explain()}")
check("a maintenance phase that OWNS the actuator (P_MAINTENANCE 60) is not "
      "dragged off its own setpoint by the peak floor", near(r3.target_w, 0.0))
check("...and neither is any P_SAFETY stop - a controller pushing power the "
      "wrong way is wrong however much money it is saving",
      near(r4.target_w, 0.0))
check("...and neither resolution is a contradiction, which would ALSO have "
      "produced 0.0 and would have meant something entirely different",
      not r3.contradiction and not r4.contradiction)

# Discharge is never capped: the loop asking for MORE discharge than the floor
# must be left alone.
BIG = pc.lo + 900.0
r5 = resolve(SAFE + [Claim.set("loop", P_LOOP, BIG, "grid"), pc])
check(f"a loop already discharging harder ({BIG:.0f} W) than the floor is not "
      f"pulled back to it - a floor cannot cap discharge",
      near(r5.target_w, BIG, 1e-9))

# THE CONTRADICTION, which is the one that would park the battery.
absurd = averager_at(LIVE_MAX_W * 4, 870.0)
p_absurd = PeakTracker.project(absurd, B + timedelta(seconds=870))
DEV_MAX = LIVE_MAX_W - 500.0        # a device rating tighter than max_w
d_absurd = decide(p_absurd, batt_w=1000.0, cap_w=CAP_W, reading=None,
                  achievable_w=min(LIVE_MAX_W, DEV_MAX), max_hold_s=MAX_HOLD_S)
r6 = resolve([Claim.limit("device", P_SAFETY, -DEV_MAX, DEV_MAX, "rating"),
              Claim.limit("supervised", P_SAFETY, -LIVE_MAX_W, LIVE_MAX_W, "max_w"),
              Claim.set("loop", P_LOOP, -200.0, "grid"),
              peakguard.claim(d_absurd)])
print(f"    an absurd projection ({p_absurd.projected_w:.0f} W with 30 s left) "
      f"wanted {d_absurd.wanted_w:.0f} W, floor clamped to "
      f"{d_absurd.floor_w:.0f} W; {r6.explain()}")
check("the floor is clamped to what the arbiter could actually grant, so the "
      "claim can NEVER produce lo > hi - a contradiction resolves to 0 W, i.e. "
      "the battery parked in the one minute of the month it should be pulling "
      "hardest", not r6.contradiction and near(r6.target_w, DEV_MAX, 1e-9))
check("...and the decision SAYS it saturated rather than reporting the clamped "
      "number as if it were the answer",
      d_absurd.saturated and d_absurd.wanted_w > d_absurd.floor_w)
check("...while the un-clamped case does not claim saturation",
      not d_over.saturated)


# ==============================================================================
print()
print("3. the ceiling, and what an unreadable peak store does to it")

obs_hi = PeakReading(OBSERVED, "2026-08", peak_w=CAP_W + 1513.0,
                     quarter_start=B)
obs_lo = PeakReading(OBSERVED, "2026-08", peak_w=CAP_W - 800.0,
                     quarter_start=B)
norec = PeakReading(NO_RECORD, "2026-08")
unread = PeakReading(NOT_OBSERVED, "2026-08", detail="disk gone")
zero = PeakReading(OBSERVED, "2026-08", peak_w=0.0, quarter_start=B)

for name, rd in (("OBSERVED above the cap", obs_hi),
                 ("OBSERVED below the cap", obs_lo),
                 ("NO_RECORD", norec), ("NOT_OBSERVED", unread),
                 ("OBSERVED at 0 W", zero)):
    c_w, c_from = quarter_ceiling_w(CAP_W, rd)
    print(f"    {name:<24} -> ceiling {c_w:.0f} W ({c_from})")

check("a month whose peak ALREADY stands above the cap raises the ceiling to "
      "it - the bill for that month is already that number, and holding every "
      "later quarter to the cap buys nothing while costing real charging",
      quarter_ceiling_w(CAP_W, obs_hi) == (CAP_W + 1513.0, CEILING_MONTH_PEAK))
check("a month peak UNDER the cap does not lower the ceiling below the cap",
      quarter_ceiling_w(CAP_W, obs_lo) == (CAP_W, CEILING_CAP))
check("NO_RECORD leaves the ceiling at the cap",
      quarter_ceiling_w(CAP_W, norec) == (CAP_W, CEILING_CAP))
check("...and so does NOT_OBSERVED: a store that did not answer makes this "
      "guard STRICTER, never looser. That direction is the whole point - the "
      "number a failed read would have become is a zero peak, and a zero peak "
      "reads as 'quiet month, full freedom'",
      quarter_ceiling_w(CAP_W, unread)[0] <= quarter_ceiling_w(CAP_W, norec)[0]
      and quarter_ceiling_w(CAP_W, unread) == (CAP_W, CEILING_CAP))
# ⚠️ GRADED ON THE STATE, NOT ON `peak_w is not None`. Today a NOT_OBSERVED
# reading always carries `peak_w=None`, so a `peak_w is not None` test happens to
# behave - and dropping `is_observed` from quarter_ceiling_w survives every other
# check in this file. That coincidence is exactly the thing not to depend on: a
# future PeakReading that carried a LAST-KNOWN peak alongside NOT_OBSERVED - the
# obvious next feature - would silently raise the ceiling off a read that failed.
# So the fixture is that reading, built here, and the rule graded is the state.
stale = PeakReading(NOT_OBSERVED, "2026-08", peak_w=CAP_W + 1500.0,
                    quarter_start=B, detail="disk gone, last known peak kept")
print(f"    NOT_OBSERVED CARRYING a {stale.peak_w:.0f} W last-known peak "
      f"-> ceiling {quarter_ceiling_w(CAP_W, stale)[0]:.0f} W "
      f"({quarter_ceiling_w(CAP_W, stale)[1]})")
check("a NOT_OBSERVED reading that CARRIES a peak still does not raise the "
      "ceiling - the gate is the reading's STATE, not whether a number happens "
      "to be attached to it. A failed read is not an observation",
      quarter_ceiling_w(CAP_W, stale) == (CAP_W, CEILING_CAP))
check("...and the relaxation is granted ONLY by an observation: the three "
      "non-observations agree, and only the OBSERVED-above-cap case differs",
      len({quarter_ceiling_w(CAP_W, r)[0]
           for r in (norec, unread, zero)}) == 1
      and quarter_ceiling_w(CAP_W, obs_hi)[0] > CAP_W)

# THE ACCEPTANCE HALF, and it is graded on a REAL store that cannot open.
broken = PeakStore(str(TMP / "no-such-volume" / "tariff.db"))
rd_broken = broken.peak(B)
tr_broken = PeakTracker(broken, cap_w=CAP_W)
print(f"    a real PeakStore on a missing volume: state={rd_broken.state}, "
      f"headroom_w()={tr_broken.headroom_w(B)}")
check("the real store answers NOT_OBSERVED rather than a peak of 0 W",
      rd_broken.state == NOT_OBSERVED and rd_broken.peak_w is None)
d_broken = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=rd_broken,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
print(f"    {d_broken.explain()}")
check("AND THE GUARD STILL ENGAGES ON IT. An unreadable peak must not disarm "
      "the cap - 'no peak recorded' is precisely the reading that would, and it "
      "is the one bug that would leave every other check in this file green",
      d_broken.engaged and near(d_broken.floor_w, d_over.floor_w, 1e-9))
check("...and it does so at the CAP ceiling, not at some number derived from "
      "the store that could not answer",
      d_broken.ceiling_w == CAP_W and d_broken.ceiling_from == CEILING_CAP)
check("`headroom_w()` is None on that same store, which is why this module does "
      "not consume it: a consumer that defaulted that None to the cap would "
      "have full headroom and no guard", tr_broken.headroom_w(B) is None)
src = (HERE / "app" / "peakguard.py").read_text(encoding="utf-8")
_calls = {n.func.attr for n in ast.walk(ast.parse(src))
          if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)}
check("...and peakguard.py calls no headroom accessor at all - graded on the "
      "parsed call syntax, so deleting the paragraph above would not satisfy it",
      "headroom_w" not in _calls and "warning_frac" not in _calls)


# ==============================================================================
print()
print("4. blindness: three ways a projection is not evidence")

# (a) NO OPEN BLOCK. A fresh averager projects 0 W - under every ceiling.
fresh = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
p_fresh = PeakTracker.project(fresh, B)
d_fresh = decide(p_fresh, batt_w=0.0, cap_w=CAP_W, reading=None,
                 achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
print(f"    fresh averager: projected {p_fresh.projected_w:.0f} W, "
      f"block_start={p_fresh.block_start}; {d_fresh.explain()}")
check("a fresh averager's projection is 0 W, i.e. comfortably under any "
      "ceiling - the degenerate value, present and quiet",
      near(p_fresh.projected_w, 0.0) and p_fresh.projected_w < CAP_W)
check("...and the guard grades block_start BEFORE the number, so it reports "
      "BLIND rather than 'the quarter is fine'",
      d_fresh.blind == BLIND_NO_BLOCK and d_fresh.projected_w is None)
check("...and the blind action is a floor of 0 W - no charging while the "
      "billed quantity cannot be measured - which is one-directional: "
      "forbidding import can only lower delivered offtake, never raise it",
      d_fresh.engaged and near(d_fresh.floor_w, 0.0))
r_blind = resolve(SAFE + [Claim.set("loop", P_LOOP, -2500.0, "grid"),
                          peakguard.claim(d_fresh)])
print(f"    a loop asking to charge 2500 W while blind: {r_blind.explain()}")
check("...and through the real arbiter that stops the charge dead at 0 W "
      "without forcing any discharge", near(r_blind.target_w, 0.0)
      and not r_blind.contradiction)

# (b) FROZEN - the case SAFETY-07 exists to backstop, and the worst of the three:
# coverage is PERFECT and the projection is confident and wrong.
frozen_av = averager_at(-4000.0, 600.0)      # meter stuck reading 4 kW of EXPORT
p_frozen = PeakTracker.project(frozen_av, B + timedelta(seconds=600))
RAILED = LIVE_MAX_W                          # the loop has wound to the rail
is_frozen = looks_frozen(FREEZE_S + 1.0, -RAILED, LIVE_MAX_W,
                         FreezeWatchdogConfig(hold_s=FREEZE_S))
d_frozen = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                  frozen=is_frozen)
print(f"    meter stuck at -4000 W with the command at {-RAILED:+.0f} W: "
      f"projected {p_frozen.projected_w:.0f} W, coverage "
      f"{p_frozen.covered_frac * 100:.0f} %, looks_frozen={is_frozen}")
print(f"    {d_frozen.explain()}")
check("a frozen meter reading export projects a CONFIDENT ZERO at full "
      "coverage - the projection cannot see this failure at all",
      near(p_frozen.projected_w, 0.0) and p_frozen.covered_frac >= 1.0)
check("...so blindness is decided by freeze.py's SHIPPED predicate, not by the "
      "projection, and not by a second copy of the conjunction here", is_frozen)
check("...and the guard goes blind and forbids the import that is railing the "
      "meter", d_frozen.blind == BLIND_FROZEN and near(d_frozen.floor_w, 0.0))
r_fz = resolve(SAFE + [Claim.set("loop", P_LOOP, -RAILED, "grid"),
                       peakguard.claim(d_frozen)])
print(f"    the railed loop, arbitrated: {r_fz.explain()}")
check(f"THE BACKSTOP: SAFETY-18's runaway - a {-RAILED:+.0f} W grid charge held "
      f"for a quarter - is cut to 0 W. freeze.py's table puts the rail at 75 s; "
      f"the alarm alone arrives at {FREEZE_S:.0f} s of a {QUARTER_S} s billing "
      f"quarter and stops nothing", near(r_fz.target_w, 0.0))
check("...and the same command NOT frozen is left alone, so the blind branch is "
      "not simply always on",
      near(resolve(SAFE + [Claim.set("loop", P_LOOP, -RAILED, "grid"),
                           peakguard.claim(decide(
                               p_frozen, batt_w=-RAILED, cap_w=CAP_W,
                               reading=None, achievable_w=LIVE_MAX_W,
                               max_hold_s=MAX_HOLD_S, frozen=False))
                           or Claim.limit("noop", P_PEAK)]).target_w,
           -RAILED, 1e-9))

# (c) COVERAGE. A real hole in the telegram stream, longer than max_hold_s.
GAP = MAX_HOLD_S * 3
gapped = averager_at(4200.0, 600.0, gap_at=200.0, gap_s=GAP)
p_gap = PeakTracker.project(gapped, B + timedelta(seconds=600))
clean = averager_at(4200.0, 600.0)
p_clean = PeakTracker.project(clean, B + timedelta(seconds=600))
d_gap = decide(p_gap, batt_w=0.0, cap_w=CAP_W, reading=None,
               achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
print(f"    a {GAP:.0f} s hole at 4200 W: uncovered {uncovered_s(p_gap):.0f} s, "
      f"projected {p_gap.projected_w:.0f} W vs {p_clean.projected_w:.0f} W with "
      f"no hole")
check("a gap makes the projection read LOW - the under-guarding direction, "
      "which is why it may not simply be graded as a number",
      p_gap.projected_w < p_clean.projected_w - 1.0)
check("...and `uncovered_s` is derived from what Projection already carries",
      near(uncovered_s(p_gap), GAP - MAX_HOLD_S, MAX_HOLD_S))
check("...and the guard goes blind on it, at the threshold p1.py already uses "
      f"to stop trusting a held sample ({MAX_HOLD_S:.0f} s) rather than at a "
      f"new number", d_gap.blind is not None
      and d_gap.blind.startswith(BLIND_COVERAGE))
small = averager_at(4200.0, 600.0, gap_at=200.0, gap_s=MAX_HOLD_S / 2)
check("...while a hole SHORTER than one hold window is not blindness - the "
      "averager covered it, so there is nothing unmeasured",
      blindness(PeakTracker.project(small, B + timedelta(seconds=600)),
                MAX_HOLD_S, False) is None)

check("blindness is decided before the projection is compared to anything: the "
      "frozen case would otherwise have read 0 W and idled, and the fresh case "
      "would have read 0 W and idled",
      d_fresh.engaged and d_frozen.engaged
      and d_fresh.projected_w is None and d_frozen.projected_w is None)


# ==============================================================================
print()
print("5. project()'s defaulted read - TEST-13 item 4, fixed here")

HELD = 4800.0
held_av = averager_at(HELD, 300.0)
p_held = PeakTracker.project(held_av, B + timedelta(seconds=300))
check("QuarterAverager exposes the held value as a documented public property",
      hasattr(held_av, "held_net_w") and near(held_av.held_net_w, HELD))
check("...and Projection carries it, so a consumer can see WHAT was "
      "extrapolated and not only the extrapolation", near(p_held.held_w, HELD))


class _Renamed:
    """An averager whose held value has been renamed - the TEST-13 scenario."""

    block_start = B
    elapsed_s = 300.0
    partial_ws = HELD * 300.0
    max_hold_s = MAX_HOLD_S


raised = None
try:
    PeakTracker.project(_Renamed(), B + timedelta(seconds=300))
except AttributeError as err:
    raised = err
print(f"    project() against a renamed averager: "
      f"{'AttributeError: ' + str(raised) if raised else 'NO ERROR - silent'}")
check("a renamed held value now raises AttributeError on the first cycle. The "
      "shipped read was `getattr(averager, '_last_net', 0.0)`: a private name "
      "with a fallback, which fails QUIET", raised is not None)

# ...and the direction of the quiet failure is the reason it mattered here.
OLD = (p_held.partial_ws if hasattr(p_held, "partial_ws")
       else HELD * 300.0)   # acc, which the old form would still have had
old_projected = (HELD * 300.0 + 0.0 * (QUARTER_S - 300.0)) / QUARTER_S
d_true = decide(p_held, batt_w=0.0, cap_w=CAP_W, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
print(f"    same fixture: shipped projection {p_held.projected_w:.0f} W "
      f"(engaged={d_true.engaged}), the defaulted form's "
      f"{old_projected:.0f} W (would engage={old_projected > CAP_W})")
check("...and it fails in the DISARMING direction: crediting 0 W held reads "
      f"{p_held.projected_w - old_projected:.0f} W LOW on this fixture, which "
      f"is the difference between a guard that engages and one that does not",
      p_held.projected_w > CAP_W >= old_projected and d_true.engaged)


# ==============================================================================
print()
print("6. the replay: what the guard does to the two days that exist")

REPLAY = sorted((HERE / "scenarios").glob("p1-net-*.jsonl.gz"))
check("the committed replay fixtures are present", len(REPLAY) >= 2)


def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W):
    """Drive one capture through the real ingest, tracker and guard.

    See the module docstring for the model and why its baseline is pessimistic.
    """
    ing = P1Ingest(phases=1, max_age_s=MAX_HOLD_S)
    store = fresh_store(f"{path.stem}-{'g' if guarded else 'u'}")
    tracker = PeakTracker(store, cap_w=cap_w)
    batt = 0.0                      # delivered discharge, W (positive)
    next_cycle = None
    graded_engaged = blind_engaged = saturated = cycles = 0
    wh = 0.0
    prev_t = None
    for line in gzip.open(path, "rt", encoding="utf-8"):
        row = json.loads(line)
        t = datetime.fromisoformat(row["t"])
        if prev_t is not None:
            wh += batt * (t - prev_t).total_seconds() / 3600.0
        prev_t = t
        ing.submit(sample(t, float(row["w"]) - batt))
        tracker.sync(ing)
        if next_cycle is not None and t < next_cycle:
            continue
        next_cycle = t + timedelta(seconds=CYCLE_S)
        cycles += 1
        d = decide(PeakTracker.project(ing.averager, t),
                   batt_w=batt, cap_w=cap_w, reading=tracker.peak(t),
                   achievable_w=max_w, max_hold_s=MAX_HOLD_S, frozen=False)
        if d.engaged:
            (globals(), None)
            if d.blind:
                blind_engaged += 1
            else:
                graded_engaged += 1
            saturated += 1 if d.saturated else 0
        if not guarded:
            continue
        batt = max(0.0, d.floor_w) if (d.engaged and d.floor_w is not None) else 0.0
    month = month_key(datetime.fromisoformat(json.loads(
        gzip.open(path, "rt", encoding="utf-8").readline())["t"]))
    return {
        "blocks": dict(store.blocks_in(month)), "cycles": cycles,
        "graded": graded_engaged, "blind": blind_engaged,
        "saturated": saturated, "wh": wh, "month": month,
    }


totals = {"u_q": 0, "u_under": 0, "g_q": 0, "g_under": 0}
worst_start = None
for path in REPLAY:
    u = replay(path, guarded=False)
    g = replay(path, guarded=True)
    uq, gq = u["blocks"], g["blocks"]
    u_peak_start = max(uq, key=lambda k: uq[k])
    u_under = sum(1 for v in uq.values() if v <= CAP_W)
    g_under = sum(1 for v in gq.values() if v <= CAP_W)
    totals["u_q"] += len(uq); totals["u_under"] += u_under
    totals["g_q"] += len(gq); totals["g_under"] += g_under
    print(f"    {path.name}: {len(uq)} quarters, "
          f"uncontrolled peak {max(uq.values()):.1f} W -> "
          f"guarded peak {max(gq.values()):.1f} W; "
          f"under {CAP_W:.0f} W: {u_under}/{len(uq)} -> {g_under}/{len(gq)}")
    print(f"      guard cycles {g['cycles']}, engaged on the measurement "
          f"{g['graded']}, blind {g['blind']}, saturated {g['saturated']}; "
          f"modelled discharge {g['wh']:.0f} Wh")
    check(f"{path.name}: the guard never RAISES a quarter - it caps charging "
          f"and forces discharge, and neither can add offtake",
          all(gq.get(k, 0.0) <= v + 1e-6 for k, v in uq.items()))
    check(f"{path.name}: no quarter that was under the limit is pushed over it",
          g_under >= u_under)
    if max(uq.values()) > CAP_W:
        worst_start = u_peak_start
        WORST_U = uq[u_peak_start]
        WORST_G = gq[u_peak_start]
        WORST_DAY = path.name
        G_SAT = g["saturated"]

check("one of the two days contains a quarter over the soft limit, or there is "
      "nothing here to grade", worst_start is not None)
print()
print(f"    THE BILLED QUARTER. {WORST_DAY}, block starting "
      f"{worst_start.isoformat()}:")
print(f"      uncontrolled {WORST_U:.1f} W   ->   guarded {WORST_G:.1f} W "
      f"({WORST_U - WORST_G:.1f} W removed, "
      f"{100.0 * (WORST_U - WORST_G) / WORST_U:.1f} %)")
check(f"the guard brings 2026-08-20's billed quarter under the "
      f"{CAP_W:.0f} W soft limit", WORST_G <= CAP_W)
check("...and it does it without ever asking for more than the inverter has - "
      "no saturation notification would have fired on this day",
      G_SAT == 0)
check(f"...and that quarter really was the one specs.md:583's argument is "
      f"about - a billed quarter over the live install's own "
      f"{LIVE_MAX_W:.0f} W rating", WORST_U > LIVE_MAX_W)

U_PCT = 100.0 * totals["u_under"] / totals["u_q"]
G_PCT = 100.0 * totals["g_under"] / totals["g_q"]
print()
print(f"    BASELINE  {totals['u_under']}/{totals['u_q']} = {U_PCT:.1f} % of "
      f"quarters under {CAP_W:.0f} W, uncontrolled")
print(f"    GUARDED   {totals['g_under']}/{totals['g_q']} = {G_PCT:.1f} %")
print(f"    Two days, not thirty. AC 3 is TARIFF-01's and is still not met - "
      f"see test_tariff.py part 9. What this compares is the same quarters with "
      f"and without the guard, which is what two days can support.")
check("the guard improves on the uncontrolled baseline rather than merely "
      "matching it", G_PCT > U_PCT)
check("...and it does not degrade a single quarter anywhere in the two days",
      totals["g_under"] >= totals["u_under"])

healthy = [p for p in REPLAY if "2026-08-23" in p.name]
if healthy:
    h = replay(healthy[0], guarded=True)
    print(f"    HEALTHY DAY FALSE-ENGAGEMENT RATE: {h['graded']} engagements on "
          f"the measurement and {h['blind']} blind, over {h['cycles']} guard "
          f"cycles ({h['wh']:.1f} Wh of modelled discharge)")
    check("zero engagements on the healthy reference day - measured, not "
          "asserted from principle. This is the number that decides trip vs "
          "notify: SAFETY-18 shipped notification-first because a false trip "
          "parks the battery, and this guard's false-trip rate on the only "
          "healthy day that exists is 0",
          h["graded"] == 0 and h["blind"] == 0 and h["wh"] == 0.0)


# ==============================================================================
print()
print("7. the wiring: the real Controller, the real cycle, the real status page")


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:
    def __init__(self):
        self.writes = []
        self.notifications = []

    async def limits(self, entity):
        return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)

    async def number(self, entity, invert=False):
        return None

    async def set_number(self, entity, value):
        self.writes.append(value)
        return True

    async def call(self, domain, service, data):
        self.notifications.append(data.get("notification_id"))
        return True


def built(db_path, opts=None):
    hass = _Hass()
    c = Controller(dict(
        {"setpoint_entity": "number.sp", "meter_source": "ha_signed",
         "p1_net_entity": "sensor.net", "meter_phases": 1,
         "max_w": LIVE_MAX_W, "gain": 0.6, "slew_w": 1000, "deadband_w": 15,
         "target_grid_w": 0, "step_w": 10, "heartbeat_s": 10,
         "stale_input_s": 15}, **(opts or {})),
        hass, _Store(), _Pub(), stop=asyncio.Event(), peak_db_path=str(db_path))
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    return c, hass


c, hass = built(TMP / "wired.db")
check(f"a Controller built with peak_cap_w ABSENT takes the shipped "
      f"{CAP_W:.0f} W - config.yaml and main.py's fallback cannot drift apart",
      c.peak.cap_w == CAP_W)
check("...and before the first cycle the guard has NO decision, which the "
      "status page must not spell as 'idle'", c.peak_guard is None)
pre = [r for r in c.checks() if "peak guard" in r["text"]]
print("    status row before the first cycle: "
      + (pre[0]["text"] if pre else "(none)"))
check("the status page says so in its own row",
      len(pre) == 1 and "no decision yet" in pre[0]["text"])

# ⚠️ THE LEGACY PATH, AND IT IS A REGRESSION THIS CARD ACTUALLY CAUSED.
# `meter_source: off` is the single-entity path an install starts on. There is no
# telegram stream, so `p1.averager` never opens a block, so a guard that ran
# there would be BLIND on every cycle forever - and blind means NO CHARGING. The
# first cut of this card did exactly that, silently, on every install that has
# not opted in to P1; test_clamp.py's export case is what caught it.
c_off, hass_off = built(TMP / "legacy.db",
                        {"meter_source": "off", "meter_entity": "sensor.grid"})
c_off.auto, c_off.soc, c_off.batt, c_off.grid = True, 55.0, 0.0, -6000.0
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c_off.cycle())
off_rows = [r for r in c_off.checks() if "peak guard" in r["text"]]
print(f"    meter_source off: guard={c_off.peak_guard}, wrote "
      f"{hass_off.writes[-1]:+.0f} W")
print("    status row: " + (off_rows[0]["text"] if off_rows else "(none)"))
check("with no P1 the guard does not run at all - it files nothing, and the "
      "legacy path can still charge", c_off.peak_guard is None
      and hass_off.writes and hass_off.writes[-1] < 0)
check("...and the status page says NOT COVERED rather than 'idle' or a green "
      "row, which is the same answer SAFETY-18 already gives on a transport "
      "that cannot see it", len(off_rows) == 1
      and "NOT COVERED" in off_rows[0]["text"] and off_rows[0]["warn"])

# ⚠️ THE FIXTURE MUST LAND IN THE CURRENT QUARTER, AND THAT IS NOT PEDANTRY.
# `cycle()` reads the real clock, and `project()` measures the open block against
# it. Telegrams stamped at a fixed date make `elapsed_wall` the AGE OF THE FIXTURE
# - tens of thousands of uncovered seconds - and the guard correctly answers
# "meter gap". Which is a true answer to a question this part is not asking: the
# first version of part 7 graded its own fixture and not the wiring.
#
# So the telegrams are laid from the START OF THE QUARTER THAT IS OPEN NOW up to
# now. There is one narrow race left - the quarter rolling between the last
# telegram and `cycle()`'s own `datetime.now()`, which leaves the averager on a
# block with no seconds remaining - so the whole setup is retried on a fresh
# Controller rather than left to flake. It is bounded at three attempts: a fourth
# would mean something is wrong that is not a boundary.
def wired_over_limit(db, want_floor_w, opts=None):
    """A real Controller driven one real cycle over a real over-limit quarter.

    ⚠️ THE LOAD IS DERIVED FROM WHERE IN THE QUARTER THE WALL CLOCK LANDS, and
    it has to be. `excess = 900 × (projected − ceiling) ÷ remaining`, so a fixed
    5200 W fixture asks for 1700 W with ten minutes left and 300 kW with five
    seconds left - the second of which saturates, and a check that saturated or
    not depending on the minute it ran would be a coin toss wearing a verdict.
    So the caller names the FLOOR it wants and the load that produces it is
    computed backwards from `remaining` at the moment of the run.
    """
    for attempt in range(4):
        ctl, hs = built(db if attempt == 0 else Path(f"{db}.{attempt}"), opts)
        ctl.auto, ctl.soc, ctl.batt = True, 55.0, 0.0
        ctl.dev_min, ctl.dev_max = -LIVE_MAX_W, LIVE_MAX_W
        now = datetime.now(timezone.utc)
        blk = datetime.fromtimestamp(
            int(now.timestamp() // QUARTER_S) * QUARTER_S, timezone.utc)
        span = int((now - blk).total_seconds())
        remaining = max(1.0, QUARTER_S - (now - blk).total_seconds())
        net_w = CAP_W + want_floor_w * remaining / QUARTER_S
        for i in range(0, span + 1, 5):
            ctl.p1.submit(sample(blk + timedelta(seconds=i), net_w))
        # ⚠️ ASSIGNED, not read back from `p1.net_w`. This helper stamps
        # `ingest_mono` with an epoch second so the averager (which keys on
        # `ingest_ts`) replays correctly, while `net_w` keys on a MONOTONIC age
        # that the same stamp makes meaningless - so `net_w` is None here and the
        # cycle would take the "inputs missing" early return before it ever
        # reached the guard.
        ctl.grid = net_w
        asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
            ctl.cycle())
        g = ctl.peak_guard
        if g is not None and g.engaged and not g.blind:
            return ctl, hs, net_w, remaining
        # Only a quarter boundary crossed between the last telegram and
        # `cycle()`'s own clock gets here. Bounded at four: a fifth would mean
        # something is wrong that is not a boundary.
    return ctl, hs, net_w, remaining


# 30 % of what the inverter can give: comfortably engaged, comfortably short of
# saturation, at any point in any quarter.
c, hass, NET_W, REMAIN = wired_over_limit(TMP / "wired2.db", LIVE_MAX_W * 0.3)
print(f"    after one real cycle over a {NET_W:.0f} W quarter "
      f"({REMAIN:.0f} s left in it): {c.reason}")
print(f"    {c.peak_guard.explain()}")
check("the real cycle files the peak claim and the real arbiter binds the real "
      "loop with it", c.peak_guard is not None and c.peak_guard.engaged
      and not c.peak_guard.saturated and "peak" in (c.reason or ""))
check("...and the number written to the inverter is the guard's floor",
      hass.writes and near(hass.writes[-1], c.peak_guard.floor_w, 1.5))
rows = [r for r in c.checks() if "peak guard" in r["text"].lower()]
print("    status row: " + (rows[0]["text"] if rows else "(none)"))
check("the status page carries exactly one peak-guard row, flagged",
      len(rows) == 1 and rows[0]["warn"])
check("an engagement that is not saturated raises NO notification - §10.3's "
      "'clamp AND alarm' is SAFETY-03's rule about a BUG SIGNAL, and a house "
      "load bigger than the inverter is not a bug",
      "goodwe_peak_saturated" not in hass.notifications)
check("...but it does reach the action log, which is where a control action "
      "that moves energy belongs",
      any("PEAK GUARD" in e for e in c.events))
n_events = sum(1 for e in c.events if "PEAK GUARD" in e)
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c.cycle())
check("...and it is edge-logged, not level-logged: a second engaged cycle adds "
      "no second line, so a quarter of engagement cannot push the cause of the "
      "event off a 40-line log",
      sum(1 for e in c.events if "PEAK GUARD" in e) == n_events)

# The saturated case, through the same real cycle.
# A floor of 1.5x what the inverter can give: it saturates wherever in the
# quarter the clock happens to be, and the load it implies stays inside
# p1.PLAUSIBLE_MAX_W - a fixture the ingest itself rejects would grade nothing.
c2, hass2, NET2, _ = wired_over_limit(TMP / "wired3.db", LIVE_MAX_W * 1.5)
check(f"the saturating fixture ({NET2:.0f} W) is a load the real ingest "
      f"accepts, not one it rejects as implausible", NET2 < PLAUSIBLE_MAX_W)
print(f"    saturated case: {c2.peak_guard.explain()}")
check("a load larger than the inverter saturates the guard", c2.peak_guard.saturated)
check("...and THAT raises exactly one notification, with its own id - the one "
      "state where the money will be spent and a human is the only remaining "
      "actuator", hass2.notifications.count("goodwe_peak_saturated") == 1)
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c2.cycle())
check("...once per episode, not once per cycle",
      hass2.notifications.count("goodwe_peak_saturated") == 1)
check("...and the command is still bounded by the rating, so a saturated guard "
      "never asks the hardware for more than it has",
      abs(hass2.writes[-1]) <= LIVE_MAX_W + 1e-6)

# CFG-03: no fourth defaults site.
cfg = (HERE / "config.yaml").read_text(encoding="utf-8")
val = (HERE / "app" / "validate.py").read_text(encoding="utf-8")
check("no new option was added: config.yaml names no peak-guard key, and "
      "§14's kW/W trap `capacity_peak_limit_kw` stays declined",
      "peak_guard" not in cfg and "capacity_peak_limit_kw" not in cfg
      and "peak_shave" not in cfg)
# Graded on the parsed call syntax, not on the text: the module docstring names
# `peak_cap_w` while arguing that it does not read it, so a grep would be
# satisfied by deleting the explanation and would fail on writing one.
_decide = next(n for n in ast.walk(ast.parse(src))
               if isinstance(n, ast.FunctionDef) and n.name == "decide")
_DECIDE_ARGS = [a.arg for a in _decide.args.args + _decide.args.kwonlyargs]
_getters = [n for n in ast.walk(ast.parse(src))
            if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
            and n.func.attr == "get"]
print(f"    peakguard.py makes {len(_getters)} options-style `.get()` calls")
check("peakguard.py reads no option of its own: it makes no `.get()` call at "
      "all and takes `cap_w` as an argument. CFG-03's three defaults sites are "
      "config.yaml's `options:`, config.yaml's `schema:` and main.py's literal "
      "fallbacks - this card adds no fourth",
      not _getters and "cap_w" in _DECIDE_ARGS)
# And the fallback in main.py must equal the shipped default, for the reason
# test_validate.py's CFG-03 block spells out at length: the two diverging is a
# live defect in this repo for `maintenance_charge_w`, and it is not to be
# repeated on the key this guard's ceiling comes from.
_mainsrc = (HERE / "app" / "main.py").read_text(encoding="utf-8")
_fallbacks = {float(m) for m in
              re.findall(r'peak_cap_w"\s*,\s*([0-9]+(?:\.[0-9]+)?)', _mainsrc)}
print(f"    peak_cap_w: config.yaml ships {CAP_W:.0f}, main.py falls back to "
      f"{sorted(_fallbacks)}")
check("...and every literal fallback for peak_cap_w in main.py equals the "
      "shipped default, so the guard's ceiling cannot differ between a "
      "Supervisor install and a test harness",
      _fallbacks and all(v == CAP_W for v in _fallbacks))


# ==============================================================================
print()
print("8. mutation: what defeats this guard, and which check bites")

# Each entry re-implements ONE defeat and reports whether a check above would
# have gone red. The last row is the honest one: it does NOT bite, and saying so
# is worth more than a green line that is not true.
MUT = []

# 1. blindness() disabled - the fresh averager's 0 W projection graded as a
#    number. Part 4 asserts d_fresh.engaged, so the defeat is visible there.
_ex_fresh = excess_w(p_fresh, CEIL)
MUT.append(("blindness() returns None - the degenerate 0 W projection is graded",
            (0.0 + _ex_fresh) <= 0.0 and d_fresh.engaged,
            "part 4: 'a fresh averager ... reports BLIND'"))

# 2. the frozen term dropped from the guard's inputs.
MUT.append(("frozen=False - SAFETY-18's term dropped",
            not decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
                       achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                       frozen=False).engaged and d_frozen.engaged,
            "part 4: 'THE BACKSTOP: SAFETY-18's runaway ... cut to 0 W'"))

# 3. `is_observed` dropped from quarter_ceiling_w, i.e. any reading's peak
#    trusted. A NOT_OBSERVED reading carries peak_w=None, so the defeat is LOUD.
def _ceiling_without_the_gate(cap, rd):
    return max(float(cap), float(rd.peak_w))


_loud = False
try:
    _ceiling_without_the_gate(CAP_W, unread)
except TypeError:
    _loud = True
MUT.append(("the `is_observed` gate dropped from quarter_ceiling_w",
            _loud, "TypeError on PeakReading.peak_w=None - it fails loud"))

# 4. the 900/remaining gain flattened to a plain difference.
_p800 = PeakTracker.project(a800, B + timedelta(seconds=800))
MUT.append(("the 900/remaining gain replaced by a plain difference",
            not near(excess_w(_p800, CEIL), _p800.projected_w - CEIL, 1.0),
            "part 1: 'the law is the difference times 9.00x'"))

# 5. the achievable_w clamp removed, so the floor reaches the arbiter unbounded.
MUT.append(("the achievable_w clamp removed",
            resolve([Claim.limit("device", P_SAFETY, -DEV_MAX, DEV_MAX, "r"),
                     Claim.set("loop", P_LOOP, -200.0, "grid"),
                     Claim.limit("peak", P_PEAK,
                                 lo=d_absurd.wanted_w)]).contradiction,
            "part 2: 'the claim can NEVER produce lo > hi'"))

# 6. the claim filed as a SET rather than a LIMIT - it would then outrank
#    nothing it should, but it WOULD survive a safety stop losing the tie.
_as_set = resolve(SAFE + [Claim.set("safety", P_SAFETY, 0.0, "stopped"),
                          Claim.set("peak", P_PEAK, pc.lo, "x")])
MUT.append(("filed as a SET at P_PEAK instead of a LIMIT",
            near(_as_set.target_w, 0.0) and pc.kind == "limit",
            "part 2: 'it is a LIMIT at P_PEAK, not a SET'"))

# 7. THE ONE THAT DOES NOT BITE, and it is stated rather than hidden.
_zero_peak = PeakReading(OBSERVED, "2026-08", peak_w=0.0, quarter_start=B)
_same = quarter_ceiling_w(CAP_W, _zero_peak) == quarter_ceiling_w(CAP_W, unread)
MUT.append(("NOT_OBSERVED silently replaced by a peak of 0.0 W",
            not _same,
            "NOTHING - and by construction: the ceiling is max(cap, peak), so a "
            "zero peak is the TIGHTEST possible input to it"))

for label, caught, where in MUT:
    print(f"    {'caught' if caught else 'NOT CAUGHT':>10}  {label}")
    print(f"                {where}")
check("the six mutations that CAN disarm this guard are each caught by a check "
      "above - a guard whose checks survive its own defeat is not a guard",
      all(m[1] for m in MUT[:6]))
check("...and the seventh is honestly reported as not caught, because it cannot "
      "disarm anything: `max(cap, peak)` makes a zero peak the strictest "
      "ceiling there is, so the disarming direction is closed by the shape of "
      "the expression rather than by a check. THAT is why the ceiling is a max "
      "and not a subtraction - `headroom = cap - peak` would have turned the "
      "same zero into FULL freedom, which part 3 shows is why headroom_w() is "
      "not consumed here", not MUT[6][1]
      and quarter_ceiling_w(CAP_W, _zero_peak)[0] == CAP_W)


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