goodwe-addon/goodwe_controller/test_clamp.py

"""Runnable check for SAFETY-03, the output clamp and its winding-loop alarm.

    python3 test_clamp.py

No framework, asserts only, same house style as test_freeze.py.

Seven parts, and parts 3 and 4 are the ones that carry SAFETY-03's own
acceptance criteria; part 6 is TEST-11's and part 7 is SAFETY-28's:

  1. THE CLAMP ARITHMETIC, on the shipped `clamp_to_rating()` - boundaries,
     symmetry, the device second stage, and the non-finite cases. Table driven.
  2. THE ALARM'S CADENCE - rising edge, re-assert, hysteresis - on the shipped
     `ClampWatchdog`. Every engagement counted, notifications not flooded.
  3. AC 1 + AC 3, WIRED. A computed setpoint that exceeds the rating, driven
     through the REAL `Controller.cycle()`: nothing over the rating reaches the
     device, and the alarm fires. Includes the historical runaway from
     specs.md 10.3 replayed against an unclamped transcription, so the check
     says what the clamp is actually worth in watts rather than asserting a
     constant.
  4. AC 2, AND IT IS THE LOAD-BEARING HALF. A loop that SATURATES WITHOUT
     CLAMPING - the inverter refusing to follow a perfectly legal command - must
     raise NO clamp alarm. If it does, the bug signal is inside the noise again
     and the whole ticket has bought nothing.
  5. NO PATH BYPASSES IT. write_setpoint() is the only route to the wire, so it
     is called directly with an out-of-range value the way the heartbeat and the
     operator's buttons call it.
  6. TEST-11. THE GUARD READS THE WRITE THAT JUST HAPPENED. `self.clamp_last`
     is assigned by `write_setpoint()` and by nothing else, so a
     `check_output_clamp()` moved above the write keeps its loop-side evidence
     and silently loses its write-side half. Graded as a CONSEQUENCE - what the
     watchdog could see when it was consulted - so that a rename or a helper
     does not trip it, and with the write-path half exercised on a stimulus of
     its own, because parts 3-5 never produce one through cycle().
  7. SAFETY-28. THE WRITE PATH IS GATED ON THE ALLOCATION RECORD - a value the
     single allocator did not place on THIS tick does not reach the wire, a
     record from a previous tick cannot authorise one, and 0 W is authorised
     unconditionally so the failsafe can never deadlock behind the gate. This is
     the check that replaced a config-shape proxy: `ACTUATOR_WRITE_KEYS` graded
     how many option KEYS exist, which stays 1 while a per-inverter config adds
     a second real writer.

------------------------------------------------------------------------------
WHY PART 4 ALSO ASSERTS ON `Decision`'s BOOLEANS AND NOT ON `reason`

`Decision.reason` is one string and the stages after the clamp overwrite it. The
check below drives a cycle that clamps the integrator AND is then slew limited,
and asserts that `reason` does NOT say so while `i_clamped` does. That is not a
style preference - an alarm keyed on the string is silent on exactly the cycles a
winding loop produces, and this is the check that would go red if someone
"simplified" main.py back onto it.

------------------------------------------------------------------------------
WHAT THIS CHECK DOES NOT COVER

- The firmware-side magnitude clamp (the card's "two clamps, not one"). This
  add-on's `write_setpoint()` is the add-on half; the ESP32 half is DRIVER-07's
  and is not in this repo's Python.
- A rig run. Nothing here needs one: the clamp is arithmetic on the write path
  and part 3 drives the shipped `Controller.cycle()` against a stub Home
  Assistant, which is the same surface test_freeze.py part 1-2 use. The wire-
  level invariant the card asks for (`|emitted| <= rating` on every frame) is
  asserted here over every value that reached `set_number()`, which is the last
  thing this process controls; proving it at the inverter needs the witness log
  and belongs with DRIVER-07.

⚠️ 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 io
import math
import sys
from datetime import datetime, timedelta, timezone

from app.clamp import (
    CLEAR_CYCLES, REASON_CODE, REASSERT_S, ClampWatchdog, clamp_to_rating,
    is_commandable,
)
from app.allocator import Allocation, UnitPlan, plan
from app.arbiter import P_LOOP, Claim, UnitBounds, allocate, resolve
from app.control import Tuning, compute
from app.main import Controller

fails = []
T0 = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)

# The live install's numbers. NOTES.md and freeze.py both say max_w: 5000 there,
# which is also the device rating, so this is the bound the alarm is written at.
LIVE_MAX_W = 5000.0
CLAMP_ID = "goodwe_output_clamp"


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


# ==============================================================================
print("1. the clamp arithmetic, on the shipped expression")

# (requested, limit, expected applied, expected engaged, expected bound)
TABLE = [
    (5001.0, 5000.0, 5000.0, True, "max_w"),      # +limit+1
    (-5001.0, 5000.0, -5000.0, True, "max_w"),    # -limit-1, the export side
    (5000.0, 5000.0, 5000.0, False, ""),          # exactly +limit: NOT engaged
    (-5000.0, 5000.0, -5000.0, False, ""),        # exactly -limit: NOT engaged
    (0.0, 5000.0, 0.0, False, ""),                # the failsafe value
    (1234.0, 5000.0, 1234.0, False, ""),
    (14768.0, 5000.0, 5000.0, True, "max_w"),     # 10.3's reported runaway
]
for requested, limit, want_applied, want_engaged, want_bound in TABLE:
    r = clamp_to_rating(requested, limit)
    check(f"{requested:+.0f} W against a {limit:.0f} W rating -> "
          f"{r.applied_w:+.0f} W, engaged={r.engaged}, bound={r.bound!r}",
          r.applied_w == want_applied and r.engaged is want_engaged
          and r.bound == want_bound)

# It is a MAGNITUDE clamp (3.2), so import and export must be bound identically.
# Derived over a sweep rather than asserted on two hand-picked values.
asym = [m for m in range(0, 12001, 37)
        if clamp_to_rating(m, LIVE_MAX_W).applied_w
        != -clamp_to_rating(-m, LIVE_MAX_W).applied_w]
check(f"symmetric over {len(range(0, 12001, 37))} magnitudes "
      f"({len(asym)} asymmetric)", not asym)

# The engagement flag has to mean "the RAW value differed", which is the words
# AC 2 uses. Re-derived from the result's own fields, not from the table.
mismatch = [m for m in range(0, 12001, 37)
            for r in [clamp_to_rating(m, LIVE_MAX_W)]
            if r.engaged != (r.requested_w != r.applied_w)]
check("engaged is exactly 'raw differs from applied', over the same sweep",
      not mismatch)

# The device second stage. Only reachable when max_w is ABOVE the device range,
# i.e. when validate_startup_config() did not run or no longer holds.
r = clamp_to_rating(3000.0, 5000.0, dev_min=-2000.0, dev_max=2000.0)
check(f"a device narrower than max_w still bites, and says which bound "
      f"({r.applied_w:+.0f} W, {r.bound!r})",
      r.applied_w == 2000.0 and r.engaged and r.bound == "device")
r = clamp_to_rating(3000.0, 5000.0, dev_min=-5000.0, dev_max=5000.0)
check("...and is a no-op once the startup validation holds",
      r.applied_w == 3000.0 and not r.engaged)

# ⚠️ REGRESSION, REVIEW-FOUND. THE FIRST STAGE USED TO `return` AND THE DEVICE
# STAGE WAS THEN NEVER REACHED, so the harder the overshoot the less the guard
# did: at max_w 5000 against a +-4000 device, +9999 emitted +5000 - OUTSIDE THE
# DEVICE RANGE - while a mere +4500 was correctly held to +4000. That inverts
# the guard and it was a regression against the pre-PR `write_setpoint()`, whose
# comment read "never send a value the device will reject outright".
#
# The property is asserted over a sweep, not on the two witness values, because
# the defect was invisible to every single-value check that happened to sit
# inside one stage.
DEV_LO, DEV_HI = -4000.0, 4000.0
escapes = [(m, clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w)
           for m in range(-15000, 15001, 61)
           for a in [clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w]
           if not (DEV_LO <= a <= DEV_HI)]
big = clamp_to_rating(9999.0, 5000.0, DEV_LO, DEV_HI)
small = clamp_to_rating(4500.0, 5000.0, DEV_LO, DEV_HI)
print(f"    max_w 5000 vs a device of {DEV_LO:+.0f}..{DEV_HI:+.0f}: "
      f"+9999 -> {big.applied_w:+.0f} W ({big.bound}), "
      f"+4500 -> {small.applied_w:+.0f} W ({small.bound})")
check(f"BOTH stages always run - {big.requested_w:+.0f} W emits "
      f"{big.applied_w:+.0f} W, inside the device's {DEV_HI:+.0f} W, and names "
      f"both bounds ({big.bound!r})",
      big.applied_w == DEV_HI and big.bound == "max_w+device")
check(f"...and it holds over the whole range, so no overshoot escapes the "
      f"device ({len(escapes)} escapes in {len(range(-15000, 15001, 61))})",
      not escapes)
check("...a big overshoot is not clamped LESS than a small one",
      abs(big.applied_w) <= abs(small.applied_w) + 0.0)

print("   non-finite input is loss of input, not an out-of-range value")
for bad in (None, float("nan"), float("inf"), float("-inf"), "abc", object()):
    check(f"is_commandable({bad!r:.30}) is False", is_commandable(bad) is False)
for good in (0, -5000.0, 5000, 12.5):
    check(f"is_commandable({good!r}) is True", is_commandable(good) is True)

# ⚠️ WHY THAT GUARD EXISTS, AND THE ANSWER IS WORSE THAN "it passes a nan
# through". MEASURED HERE, not recalled: every comparison against a nan is
# False, so `min(limit, nan)` keeps its FIRST argument and a naive clamp turns a
# nan into a command at the FULL RATING - and then reports engaged=True, because
# limit != nan, so the winding-loop counter gets a transport bug filed in it.
# The number below is computed by the same expression the guard replaces.
nan = float("nan")
naive = max(-LIVE_MAX_W, min(LIVE_MAX_W, nan))
check(f"a naive clamp turns a nan into {naive:+,.0f} W and calls it "
      f"engaged={naive != nan} - hence the guard",
      naive == LIVE_MAX_W and (naive != nan) is True)


# ==============================================================================
print()
print("2. the alarm's cadence: every engagement counted, no flood")

wd = ClampWatchdog()
t = T0
fired = [wd.note(True, "winding", t + timedelta(seconds=i)) for i in range(10)]
check(f"the rising edge fires ({fired[0]})", fired[0] is True)
check(f"the next 9 one-second cycles fire {sum(fired[1:])} more times "
      f"(re-assert is {REASSERT_S:.0f} s)", sum(fired[1:]) == 0)
check(f"but every one of them is counted: engagements={wd.engagements}",
      wd.engagements == 10)
check(f"and it is one episode, not ten: episodes={wd.episodes}",
      wd.episodes == 1)

# --- THE RE-ASSERT CADENCE, PINNED IN BOTH DIRECTIONS ------------------------
# ⚠️ REVIEW-FOUND. This used to read `wd.note(True, ..., T0 + REASSERT_S + 9)`,
# which is green for ANY value of the constant - REASSERT_S = 3600 passed, and an
# hour of silence is precisely what the module promises not to do. A check whose
# stimulus is written in terms of the constant it is grading cannot grade it.
#
# The requirements are stated in wall-clock seconds and derived from what the
# alarm is FOR, so they do not move when the constant does:
#
#   floor - a loop pinned on the rail must speak at least MIN_PER_QUARTER times
#           inside the 900 s quarter that bills the capacity tariff. That window
#           is the same one freeze.py and DOCS.md reason about.
#   ceiling - it must not speak on more than MAX_FRACTION of cycles, or it is
#           the per-cycle flood the split exists to remove.
QUARTER_S = 900.0
MIN_PER_QUARTER = 5
MAX_FRACTION = 0.20

pinned = ClampWatchdog()
alarms = sum(1 for i in range(int(QUARTER_S))
             if pinned.note(True, "pinned", T0 + timedelta(seconds=i)))
print(f"    pinned on the rail for {QUARTER_S:.0f} s at 1 s cycles: "
      f"{pinned.engagements} engagements, {alarms} alarm(s) "
      f"({alarms / QUARTER_S:.1%} of cycles)")
check(f"a loop pinned for a whole billing quarter speaks at least "
      f"{MIN_PER_QUARTER} times (it spoke {alarms})",
      alarms >= MIN_PER_QUARTER)
check(f"...and on no more than {MAX_FRACTION:.0%} of cycles (it was "
      f"{alarms / QUARTER_S:.1%})", alarms <= MAX_FRACTION * QUARTER_S)
check("...while every single cycle is still counted",
      pinned.engagements == int(QUARTER_S))

# --- THE HYSTERESIS, PINNED AGAINST THE CHATTER DIRECTION --------------------
# ⚠️ REVIEW-FOUND. This used to issue `range(CLEAR_CYCLES - 1)` clean cycles,
# which at CLEAR_CYCLES = 1 issues ZERO of them and passes while printing
# "0 clean cycles do NOT clear it". Green for every value >= 1 - including 1,
# which IS the chatter the constant exists to prevent. The stimulus is a literal
# now: ONE clean cycle must not clear the alarm, whatever the constant says.
wd2 = ClampWatchdog()
wd2.note(True, "winding", T0)
wd2.note(False, "", T0 + timedelta(seconds=1))
check(f"ONE clean cycle does NOT clear the alarm (active={wd2.active}) - "
      f"a command dithering across the rail must not chatter",
      wd2.active is True)
survived = 1
while wd2.active and survived < 50:
    wd2.note(False, "", T0 + timedelta(seconds=1 + survived))
    survived += 1
print(f"    it took {survived} consecutive clean cycles to de-assert "
      f"(CLEAR_CYCLES = {CLEAR_CYCLES})")
check(f"...and it does de-assert, after {survived} of them, rather than "
      f"latching", wd2.active is False and survived == CLEAR_CYCLES)
check("...and the engagement count survives the clear, so the evidence does",
      wd2.engagements == 1)
check("a clean cycle on an idle watchdog says nothing",
      wd2.note(False, "", T0 + timedelta(seconds=300)) is False)


# ==============================================================================
print()
print("3. AC 1 + AC 3, wired: over the rating both clamps AND alarms")


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:
    """Stub HA. `track` False models an inverter that is switched off entirely,
    which is the 10.3 runaway's own setup."""

    def __init__(self, grid, batt=0.0, track=True):
        self.grid = grid
        self.batt = batt
        self.track = track
        self.calls = []
        self.writes = []

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

    async def number(self, entity, invert=False):
        return {"sensor.grid": self.grid, "sensor.soc": 50.0,
                "sensor.batt": self.batt}.get(entity)

    async def set_number(self, entity, value):
        self.writes.append(float(value))
        if self.track:
            self.batt = float(value)
        return True

    async def call(self, domain, service, data):
        self.calls.append((domain, service, data))
        return True


def wired(grid, cycles=40, batt=0.0, track=True, **over):
    async def go():
        hass = _Hass(grid, batt=batt, track=track)
        c = Controller(dict(
            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
             "meter_source": "off", "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},
            **over), hass, _Store(), _Pub(), stop=asyncio.Event())
        c.auto = True
        c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
        # The red status row exists only WHILE the clamp is engaged, and the
        # hysteresis clears it a few cycles later - so it is captured live
        # rather than looked for after the run, when it is correctly gone.
        c.red_snapshot = []
        for _ in range(cycles):
            await c.read_inputs()
            await c.cycle()
            if c.clamp_wd.active and not c.red_snapshot:
                c.red_snapshot = c.checks()
        return hass, c
    return asyncio.run(go())


def clamp_notes(hass):
    return [d for dom, svc, d in hass.calls
            if (dom, svc) == ("persistent_notification", "create")
            and d.get("notification_id") == CLAMP_ID]


# THE 10.3 RUNAWAY, replayed. "With the inverter switched off entirely, an
# unclamped integrator climbed ~130 W per 4 s to over 10 kW and, pinned at the
# 5 kW rail, reported 14 768 W." `track=False` is the inverter switched off: it
# reports 0 W however hard it is commanded, so the error never closes.
hass_r, ctl_r = wired(6000.0, cycles=60, track=False)
emitted = [abs(w) for w in hass_r.writes]

# What the SAME error would have done to an accumulator with no bound - the
# transcription is deliberately the unclamped velocity form 10.3 describes, so
# the figure below is computed here and not quoted from the spec.
runaway, err = 0.0, 6000.0
for _ in range(60):
    runaway += 0.6 * err
print(f"    unclamped, the same 60 cycles reach {runaway:,.0f} W; "
      f"the largest value that reached the device is {max(emitted):,.0f} W")
check(f"an unbounded accumulator really does run away here "
      f"({runaway:,.0f} W >> {LIVE_MAX_W:.0f} W)", runaway > 3 * LIVE_MAX_W)
check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
      f"exceeds the {LIVE_MAX_W:.0f} W rating",
      emitted and max(emitted) <= LIVE_MAX_W)
# ⚠️ AND THE CLAMP IS NOT THE ONLY THING HOLDING THAT LINE HERE, which is worth
# saying out loud rather than letting the number above imply otherwise. Against
# an inverter reporting 0 W the saturation freeze engages after
# `saturation_cycles` and pins the command well BELOW the rail - that is why the
# largest emitted value is what it is and not 5000 W. The clamp is what makes it
# bounded; the freeze is what makes it small. A check that read
# `max(emitted) == max_w` would have been asserting the freeze, not the clamp.
check(f"...and the saturation freeze is what pins it at {max(emitted):,.0f} W "
      f"rather than the rail", ctl_r.sat_count >= ctl_r.tuning.saturation_cycles)

notes_r = clamp_notes(hass_r)
print(f"    the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
      f"in {ctl_r.clamp_wd.episodes} episode(s), raising {len(notes_r)} "
      f"notification(s)")
check("AC 3: and it ALARMED - silently clipping this is the defect",
      ctl_r.clamp_wd.engagements > 0 and len(notes_r) >= 1)
check("...with its own notification_id, not shared with any other guard",
      all(d["notification_id"] == CLAMP_ID for d in notes_r))
check("...one episode, re-asserted rather than one per cycle",
      ctl_r.clamp_wd.episodes == 1 and len(notes_r) < ctl_r.clamp_wd.engagements)
check("...and it names itself a bug signal, not a limit in use",
      bool(notes_r) and "out of range" in notes_r[0]["message"])
rows_r = [e for e in ctl_r.events if REASON_CODE in e]
check(f"the action log records every engagement, coalesced into one row per "
      f"episode ({len(rows_r)} row(s), {ctl_r.clamp_wd.engagements} "
      f"engagement(s))",
      len(rows_r) == ctl_r.clamp_wd.episodes
      and f"{ctl_r.clamp_wd.engagements} engagement(s)" in rows_r[0])

# ⚠️ REGRESSION, AND IT WAS A REAL BUG BEFORE IT WAS A CHECK. `log_event()` keeps
# 40 rows. A row per engagement flushes that ring inside 40 cycles of a winding
# loop, and the first thing it evicted was SAFETY-18's own "METER FROZEN?" row -
# i.e. this guard silencing another guard's log. Driven here with a plant that
# TRACKS the command, so the saturation freeze never engages and the integrator
# clamps on essentially every cycle.
hass_l, ctl_l = wired(9000.0, cycles=80, track=True)
rows_l = [e for e in ctl_l.events if REASON_CODE in e]
other_l = [e for e in ctl_l.events if REASON_CODE not in e]
print(f"    80 winding cycles: {ctl_l.clamp_wd.engagements} engagement(s) -> "
      f"{len(rows_l)} log row(s), leaving room for {len(other_l)} other event(s)")
check(f"a long winding episode does NOT flood the 40-row event log "
      f"({len(rows_l)} rows for {ctl_l.clamp_wd.engagements} engagements)",
      ctl_l.clamp_wd.engagements > 40 and len(rows_l) <= ctl_l.clamp_wd.episodes)
check("...and the count in the row is the counter's, so nothing is lost",
      f"{ctl_l.clamp_wd.engagements} engagement(s)" in rows_l[-1])
# The RED row exists only while engaged, and the episode above is short, so it
# is captured mid-run rather than asserted after the hysteresis has cleared it.
check("the status page goes red WHILE it is engaged",
      any(not c["ok"] and not c["warn"] and "OUTPUT CLAMP ENGAGED" in c["text"]
          for c in ctl_r.red_snapshot))
check("...and once clear, a warn row still says how many times it engaged",
      any(c["warn"] and f"{ctl_r.clamp_wd.engagements} time(s)" in c["text"]
          for c in ctl_r.checks()))

# ⚠️ NOT A TRIP. 11.2's action is "clamp and alarm"; the loop keeps running and
# keeps writing the clamped value. If a later cut changes that it must come here.
check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W, not 0 W",
      ctl_r.target != 0.0 and ctl_r.target == hass_r.writes[-1])
check("...and the controller still reports itself running, not failsafe",
      ctl_r.auto and not ctl_r.sign_wd.tripped)

# The export side, because the clamp is a magnitude clamp and a sign bug in it
# would be invisible on the import case alone.
hass_x, ctl_x = wired(-6000.0, cycles=60, track=False)
check(f"the export side clamps and alarms too "
      f"({min(hass_x.writes):+,.0f} W, {len(clamp_notes(hass_x))} note(s), "
      f"{ctl_x.clamp_wd.engagements} engagement(s))",
      min(hass_x.writes) < 0 and abs(min(hass_x.writes)) <= LIVE_MAX_W
      and clamp_notes(hass_x) and ctl_x.clamp_wd.engagements > 0)


# ==============================================================================
print()
print("4. AC 2: saturation WITHOUT clamping must NOT raise a clamp alarm")

# A legal command the inverter will not follow. The meter sits just outside the
# deadband so the loop moves a little and then rests; `batt` is pinned far from
# whatever is commanded, which is the saturation condition (|cmd - actual| >
# saturation_w) and nothing else. The integrator never approaches the rating.
hass_s, ctl_s = wired(60.0, cycles=40, batt=2500.0, track=False)
peak = max(abs(w) for w in hass_s.writes)
print(f"    saturated for {ctl_s.sat_count} consecutive cycles; the command "
      f"peaked at {peak:.0f} W, {peak / LIVE_MAX_W:.1%} of the rating")
# ⚠️ THE POSITIVE HALF FIRST. A negative check that saw nothing happen is not an
# observation - if the saturation freeze never engaged, "no clamp alarm" is
# vacuous and this check is worthless.
check(f"the saturation freeze REALLY DID engage (sat_count="
      f"{ctl_s.sat_count} >= {ctl_s.tuning.saturation_cycles})",
      ctl_s.sat_count >= ctl_s.tuning.saturation_cycles)
check("...and the event log says so, in the saturation's own words",
      any("saturation freeze" in e for e in ctl_s.events))
check(f"...while the command stayed well inside the rating "
      f"({peak:.0f} W < {LIVE_MAX_W:.0f} W)", peak < LIVE_MAX_W)
# Now the negative, which is only worth anything because of the three above.
check("AC 2: NO clamp notification was raised", not clamp_notes(hass_s))
check("AC 2: the clamp counted ZERO engagements", ctl_s.clamp_wd.engagements == 0)
check("AC 2: no OUTPUT_CLAMP row in the action log",
      not [e for e in ctl_s.events if REASON_CODE in e])
check("AC 2: and no clamp row on the status page",
      not [c for c in ctl_s.checks() if "output clamp" in c["text"].lower()])

print("   ...and the two alarms do not share a notification_id")
ids_clamp = {d["notification_id"] for d in clamp_notes(hass_r)}
ids_other = {d.get("notification_id") for dom, svc, d in hass_r.calls
             if svc == "create"} - ids_clamp
check(f"clamp ids {sorted(ids_clamp)} are disjoint from every other alarm's "
      f"{sorted(x for x in ids_other if x)}", not (ids_clamp & ids_other))

print("   ...and the alarm cannot be keyed on Decision.reason")
# A cycle that clamps the integrator AND is then slew limited. `reason` reports
# the LAST stage, so a string-keyed alarm is silent on it; the boolean is not.
t = Tuning(max_w=2000.0, slew_w=100.0, gain=0.6)
d = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t, i_w=1900.0)
print(f"    reason={d.reason!r}  i_clamped={d.i_clamped}  clamped={d.clamped}")
check("the integrator DID clamp on that cycle", d.i_clamped is True)
check("...but `reason` says 'slew-limited' - the clamp was overwritten "
      "downstream, so a string-keyed alarm is SILENT here",
      d.reason == "slew-limited" and "clamp" not in d.reason)
# And the same for the OUTPUT clamp, on a config that lets the integrator run
# above the rail - then both flags fire and `reason` still reports neither.
t2 = Tuning(max_w=2000.0, integrator_max_w=3000.0, slew_w=100.0, gain=0.6)
d2 = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t2, i_w=2900.0)
print(f"    reason={d2.reason!r}  i_clamped={d2.i_clamped}  clamped={d2.clamped}")
check("both clamps engaged, and `reason` reports neither",
      d2.i_clamped and d2.clamped and "clamp" not in d2.reason)
# ⚠️ HOW THE FREEZE AND THE CLAMP INTERACT, MEASURED RATHER THAN ASSUMED, AND
# THE MEASUREMENT REFUSED TWO GUESSES BEFORE THIS ONE. Written first as "a
# frozen cycle never clamps" (false), then as "only the i_w == 0 reseed
# exception clamps" (also false - at i_w == 0 with a matching actual the cycle
# is not saturated at all, so it is never frozen). What the sweep actually
# shows: a frozen cycle clamps exactly when the integrator REVERSES. control.py
# permits that on purpose - "it may fall, cross zero, or reverse outright" -
# because the alternative deadlocks the loop, and a reversal is a full swing
# that can clear the bound on the far side in one step.
#
# That matters for this ticket rather than being trivia: it is the one shape
# where the saturation freeze and a clamp engagement coincide, so it is exactly
# where an alarm that shared a channel with saturation would lose the clamp.
grid_sweep = (-9000.0, -3000.0, 3000.0, 9000.0)
grid = [(pp, g) for pp in (-1900.0, -500.0, 0.0, 500.0, 1900.0)
        for g in grid_sweep]
runs = {(pp, g): compute(pp, g, 0.0, t, sat_count=5, i_w=pp) for pp, g in grid}
frozen = [k for k, d in runs.items() if d.frozen]
frozen_clamps = [k for k in frozen if runs[k].i_clamped]
# The property, stated independently of the control law's own branches: the
# correction pushes AGAINST the standing integrator, and the resulting swing
# clears the bound.
reversals = [(pp, g) for pp, g in frozen
             if pp * g < 0 and abs(pp + t.gain * g) > t.max_w]
print(f"    {len(frozen)} of {len(grid)} sweep cells are frozen; "
      f"those that also clamp: {frozen_clamps}")
check(f"a frozen cycle CAN clamp, and only by reversing "
      f"({len(frozen_clamps)} of {len(frozen)} frozen cells)",
      frozen_clamps and frozen_clamps == reversals)
check("...and a frozen cycle pushing the way it already leans never clamps - "
      "that is the freeze doing its job",
      not [k for k in frozen if runs[k].i_clamped and k[0] * k[1] > 0])
check("a healthy cycle sets neither flag",
      not any([compute(0.0, 100.0, 0.0, t).i_clamped,
               compute(0.0, 100.0, 0.0, t).clamped]))


# ==============================================================================
print()
print("5. no path reaches the wire around the clamp")


def arm(c, value):
    """Give `c` a real allocation record authorising `value` on the CURRENT tick.

    ⚠️ SAFETY-28 MADE THIS NECESSARY, AND THE FACT THAT IT IS NECESSARY IS THE
    TICKET. `write_setpoint()` now refuses any value the single allocator did
    not place on this tick, so the probes below - which call it directly, the
    way the heartbeat and the operator's buttons do - have to say where their
    number came from. Before this they did not, and that was precisely the hole:
    a second writer's number reached the wire on nothing but the fact that
    somebody called the method.

    ⚠️ AND IT IS BUILT THROUGH THE SHIPPED `resolve()`, NOT BY HAND. A helper
    that assembled an `Allocation` itself would be a second allocator living in
    the test file, and every check below would then be grading the fixture's
    idea of a record rather than the one main.py produces. The tick is bumped
    the same way `cycle()` bumps it, for the same reason.
    """
    c.alloc_tick += 1
    res = resolve([Claim.set("loop", P_LOOP, value, "test stimulus")],
                  units=c.units, live=c.units, tick=c.alloc_tick)
    c.allocation = res.allocation
    return res.allocation


async def _direct():
    hass = _Hass(0.0)
    c = Controller({"setpoint_entity": "number.sp", "max_w": 2000,
                    "meter_source": "off"},
                   hass, _Store(), _Pub(), stop=asyncio.Event())
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    # The shape the heartbeat, shutdown() and handle_action() all use - each one
    # authorised by this tick's allocation, which is what those callers have
    # too: the heartbeat re-affirms `self.target`, and the buttons write 0 W,
    # which needs no allocator at all (see part 7).
    for wanted in (9999.0, -9999.0):
        arm(c, wanted)
        await c.write_setpoint(wanted, force=True)
    # A nan is loss of input, not an unauthorised write: rule 1 turns it into
    # 0 W ABOVE the gate, and 0 W is always authorised. No arming, on purpose -
    # if the gate ever moved above the coercion this line would go red.
    await c.write_setpoint(float("nan"), force=True)
    arm(c, 1500.0)
    await c.write_setpoint(1500.0, force=True)
    return hass, c


hass_d, ctl_d = asyncio.run(_direct())
print(f"    asked for 9999 / -9999 / nan / 1500 W; the device saw "
      f"{hass_d.writes}")
check("a direct out-of-range write is clamped, not passed through",
      hass_d.writes[:2] == [2000.0, -2000.0])
check("a nan is refused and becomes 0 W, per 11.1 rule 1",
      hass_d.writes[2] == 0.0)
check("...and is NOT counted as a clamp engagement - it is a transport bug",
      sum(1 for e in ctl_d.events if REASON_CODE in e) == 2)
check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
check("every emitted frame satisfies |emitted| <= the rating",
      all(abs(w) <= 2000.0 for w in hass_d.writes))

# --- 5b. ALLOC-01: AND THE ALLOCATOR DID NOT ADD ONE ------------------------
# ⚠️ "NO PATH REACHES THE WIRE AROUND THE CLAMP" IS A CLAIM ABOUT THE WHOLE
# ADD-ON, so it has to be re-made every time a module learns to produce a number
# that looks like a setpoint. ALLOC-01 added `app/allocator.py`, whose output is
# a dict of per-inverter watts - numbers that are exactly the shape of a thing
# somebody would write - and the temptation it creates is real: a second writer
# for a second inverter is DEV-02's card, and the wrong way to build it is a
# `set_number` call next to the allocation.
#
# Two checks, and neither is about the allocator's arithmetic:
#   * it CANNOT write, and that is established from its imports rather than
#     from its behaviour, because a behavioural check only sees the paths a
#     fixture happens to take;
#   * at the shipped roster the number that reaches the wire is still the
#     arbiter's `target_w`, unclamped by the allocator, so SAFETY-03's clamp is
#     still the binding one and not a second opinion after a first clamp.
import app.allocator as _alloc_mod                                # noqa: E402

_alloc_src = io.open(_alloc_mod.__file__, encoding="utf-8").read()
_alloc_imports = sorted(
    n.split(".")[0]
    for line in _alloc_src.splitlines() if line.startswith(("import ", "from "))
    for n in [line.split()[1]])
print(f"    app/allocator.py imports: {_alloc_imports}")
check("the allocator imports NOTHING that can reach the wire - no `hass`, no "
      "`aiohttp`, not even `main`. It is a pure function of intervals and a "
      "number, so 'it cannot write' is a property of the file rather than of "
      "the paths a fixture happened to take",
      _alloc_imports == ["dataclasses", "math"])
check("...and it never mentions the setpoint entity, `set_number`, or the "
      "clamp - the three names any bypass would have to spell",
      not any(n in _alloc_src for n in ("set_number", "setpoint_entity",
                                        "clamp_to_rating")))
check("...and it must not: it has no measurement either. The allocator reading "
      "P1 would be a SECOND integrator on the site measurement, which is the "
      "same two-writers failure one level up",
      not any(n in _alloc_src for n in ("p1", "P1Ingest", "grid_w")))

check("at the shipped roster of ONE the allocator is a pass-through, so the "
      "number handed to write_setpoint() is the arbiter's own target and "
      "SAFETY-03's clamp is still the FIRST and only clamp it meets. An "
      "allocator that pre-clamped here would leave this file grading a value "
      "that was already in range, and the clamp untested on the live path",
      allocate(9999.0, {"inverter-a": UnitBounds()}) == ({"inverter-a": 9999.0},
                                                        True))
check("...and that unclamped 9999 W is exactly what part 5 above just proved "
      "the write path stops at the rating. The two halves meet here: the "
      "allocator does not clamp, the wire does",
      hass_d.writes[0] == 2000.0)

# ==============================================================================
print()
print("6. TEST-11: the guard's evidence is THIS cycle's write, not the last one")

# WHY THIS PART EXISTS. `check_output_clamp()` sits one line below
# `await self.write_setpoint(self.target)` in cycle(), and moving it above the
# write left this file, test_freeze.py and test_loop.py ALL GREEN. That is not a
# missing assertion about line numbers - it is a missing assertion about
# EVIDENCE. The guard has two sources:
#
#   * `self.loop_clamped`, set from the Decision well ABOVE the write. Still
#     seen wherever the call is put, which is why the move is nearly invisible:
#     on the §10.3 replay above, every engagement the alarm reports is this one.
#   * `self.clamp_last`, which ONLY `write_setpoint()` assigns (main.py) and
#     which is None until it has run. This half is what loses its evidence.
#
# So the check below does not pin the call's POSITION IN THE SOURCE, and must
# not: the comment at that line warns against tidying SAFETY-03's ordering and
# SAFETY-18's into one rule, and a check a rename breaks is worse than no check.
# It pins the CONSEQUENCE - that when the guard is consulted, `self.clamp_last`
# is the result of the write of THIS cycle - and part 6b runs the identical
# assertions through a deliberately refactored controller to prove it.


class _SpyWatchdog(ClampWatchdog):
    """The shipped watchdog, plus a record of what the guard could SEE.

    ⚠️ IT HOOKS THE CONSUMER, NOT THE GUARD'S NAME. Wrapping
    `check_output_clamp` by name would make a rename - AC 2's legitimate
    refactor - look identical to the defect, and this ticket is explicitly not
    allowed to buy its check with a nuisance. `clamp_wd.note()` is the one call
    the guard exists to make, so it is hooked instead: it is reached from the
    guard under any name, from any helper, as long as that helper runs after the
    write.
    """

    def __init__(self):
        super().__init__()
        self.owner = None
        self.seen = []          # clamp_last.requested_w at each consultation
        self.seen_write = []    # was the WRITE-path half engaged, per cycle

    def note(self, engaged, reason, now):
        cl = self.owner.clamp_last
        self.seen.append(None if cl is None else cl.requested_w)
        self.seen_write.append(bool(cl and cl.engaged))
        return super().note(engaged, reason, now)


def probed(cls=Controller, grid=9000.0, cycles=5, **over):
    """`wired()`, with the watchdog replaced by the spy and the per-cycle
    command returned. A RAMP is used on purpose: the slew limiter makes every
    cycle's command a different number, which is what makes "this cycle's
    evidence" and "last cycle's evidence" distinguishable at all."""
    async def go():
        # `track=True` - the plant follows, so the saturation freeze never
        # engages and the slew limiter keeps the command MOVING; against a
        # dead plant the freeze pins it flat after three cycles and the ramp
        # stops being a ramp.
        hass = _Hass(grid, track=True)
        c = cls(dict(
            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
             "meter_source": "off", "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},
            **over), hass, _Store(), _Pub(), stop=asyncio.Event())
        c.auto = True
        c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
        c.clamp_wd = _SpyWatchdog()
        c.clamp_wd.owner = c
        asked = []
        for _ in range(cycles):
            await c.read_inputs()
            await c.cycle()
            asked.append(c.target)
        return hass, c, asked
    return asyncio.run(go())


def evidence_checks(label, asked, wd):
    """The same four claims, run against the shipped controller (6a) and against
    a refactored one (6b). Sharing them is the point: if a rename or a helper
    changed the answer, this function would say so in the same words."""
    seen = wd.seen
    print(f"    {label}: asked "
          f"{[f'{w:+.0f}' for w in asked]}")
    print(f"    {label}: the guard saw "
          f"{['None' if s is None else f'{s:+.0f}' for s in seen]}")
    check(f"{label}: PRECONDITION - the command MOVES on every cycle, so this "
          f"cycle's evidence and last cycle's are different numbers",
          len(set(asked)) == len(asked) and len(asked) >= 4)
    check(f"{label}: the guard was consulted exactly once per cycle",
          len(seen) == len(asked))
    check(f"{label}: AC 1 - on EVERY cycle self.clamp_last held the result of "
          f"the write THAT CYCLE. write_setpoint() is the only line that "
          f"assigns self.clamp_last, so a check_output_clamp() moved above the "
          f"write grades the PREVIOUS cycle's write instead",
          seen == asked)
    check(f"{label}: ...and on the FIRST cycle self.clamp_last was already set "
          f"- before the run's first write it is None, i.e. the write-path half "
          f"of the guard has no evidence at all",
          bool(seen) and seen[0] is not None)


hass_e, ctl_e, asked_e = probed()
evidence_checks("shipped", asked_e, ctl_e.clamp_wd)

# ⚠️ AND HERE IS WHY AC 3 NEEDS A STIMULUS OF ITS OWN, MEASURED RATHER THAN
# ASSERTED. Through the real cycle() the write-path half never engages: the
# arbiter files the device range and max_w as P_SAFETY LIMIT claims, so
# `resolution.target_w` is already inside both bounds before write_setpoint()
# re-applies them (main.py calls that duplication defence in depth, and it is).
# Every engagement the alarm reports on this path is therefore the loop-side
# one - which is exactly what makes the move nearly invisible, and exactly why
# 6c below drives the write path directly instead of hoping cycle() will.
check(f"MEASURED: the write-path half engaged 0 times in {len(asked_e)} real "
      f"cycles ({sum(ctl_e.clamp_wd.seen_write)}), so no cycle()-driven check "
      f"can be watching it", not any(ctl_e.clamp_wd.seen_write))


class _RefactoredController(Controller):
    """AC 2, as a fixture rather than as a promise: the two refactors a P3 tidy
    would actually make, applied at once.

      * the guard is RENAMED to `_clamp_guard()`, the old name left as the thin
        delegate such a rename leaves behind;
      * the write MOVES INTO A HELPER, `_emit_setpoint()`.

    Neither changes when the guard runs relative to the write, so every check in
    `evidence_checks()` must stay green. If one of them goes red here, the check
    is grading the text of main.py and has to be thrown away.
    """

    async def _emit_setpoint(self, value, *, force=False):
        await Controller.write_setpoint(self, value, force=force)

    async def write_setpoint(self, value, *, force=False):
        await self._emit_setpoint(value, force=force)

    async def _clamp_guard(self):
        await Controller.check_output_clamp(self)

    async def check_output_clamp(self):
        await self._clamp_guard()


# ⚠️ VERIFIED AGAINST main.py ITSELF, NOT ONLY AGAINST THIS SUBCLASS. Renaming
# `check_output_clamp` throughout main.py AND moving both the write and the guard
# into a new `_emit_and_guard()` helper called from cycle() leaves every check in
# this part green; moving the guard above the write turns the two AC 1 lines
# below red. The one thing a repo-wide rename does cost is the direct call in 6c,
# which is a mechanical find/replace and fails as an AttributeError naming the
# new method - not as an assertion claiming the ordering is wrong, which is the
# false signal this part was written to avoid.
hass_f, ctl_f, asked_f = probed(cls=_RefactoredController)
evidence_checks("renamed + helper", asked_f, ctl_f.clamp_wd)
check("AC 2: the refactored controller behaves identically to the shipped one "
      "- same commands, same evidence, so the check above is not pinning "
      "either name",
      asked_f == asked_e and ctl_f.clamp_wd.seen == ctl_e.clamp_wd.seen)


# --------------------------------------------------------------------------- #
# 6c. AC 3 - the WRITE-PATH half, exercised
# --------------------------------------------------------------------------- #
# The stimulus is SAFETY-20's shape and that is not a coincidence: `max_w` 5000
# on an inverter that reports +-2500 is the one configuration where the value
# put on the wire DIFFERS from the value the loop asked for while every loop-side
# flag stays clean. That difference is the whole content of `clamp_last`, and a
# DEVICE-bound engagement is the only kind `loop_clamped` cannot also see.
#
# It is driven through `write_setpoint()` and the guard directly rather than
# through cycle(), because the measurement above says cycle() cannot produce it
# today. That is stated as a limitation, not hidden: what is being graded here is
# the guard's own contract - given a write that clamped, alarm - and the ordering
# in cycle() is what 6a grades.


async def _write_path(moved=False):
    """`moved` runs the two calls in the order the ticket asks about."""
    hass = _Hass(0.0)
    c = Controller({"setpoint_entity": "number.sp", "meter_source": "off",
                    "max_w": LIVE_MAX_W},
                   hass, _Store(), _Pub(), stop=asyncio.Event())
    c.auto = True
    # The inverter is SMALLER than the configured ceiling - SAFETY-20's install.
    c.dev_min, c.dev_max = -2500.0, 2500.0
    c.target = 3000.0
    # SAFETY-28: the write path only carries numbers the allocator placed, so
    # this stimulus is authorised the same way cycle()'s is. It has to be: the
    # clamp is what this part grades, and a gate refusal would substitute 0 W
    # and there would be no clamp to see.
    arm(c, c.target)
    if moved:
        await c.check_output_clamp()
        await c.write_setpoint(c.target)
    else:
        await c.write_setpoint(c.target)
        await c.check_output_clamp()
    return hass, c


hass_w, ctl_w = asyncio.run(_write_path())
notes_w = clamp_notes(hass_w)
print(f"    max_w {LIVE_MAX_W:.0f} on a +-2500 W device, asked for "
      f"{ctl_w.target:+.0f} W: the device saw {hass_w.writes[-1]:+.0f} W, "
      f"bound={ctl_w.clamp_last.bound!r}, loop_clamped={ctl_w.loop_clamped[0]}")
check("PRECONDITION: the write really did clamp, on the DEVICE bound",
      ctl_w.clamp_last is not None and ctl_w.clamp_last.engaged
      and ctl_w.clamp_last.bound == "device"
      and hass_w.writes[-1] == 2500.0)
check("...and the loop-side half is SILENT here, so self.clamp_last is the "
      "only evidence that this happened at all",
      ctl_w.loop_clamped == (False, ""))
check("AC 3: the guard alarmed on write-path evidence alone",
      ctl_w.clamp_wd.engagements == 1 and len(notes_w) == 1)
check("...and the operator is told WHICH bound bit, so a device-rating clamp "
      "is not reported as a max_w one",
      bool(notes_w) and "device bound" in notes_w[0]["message"])
check("...and the action log carries the row, written by write_setpoint()",
      any(REASON_CODE in e and "device bound" in e for e in ctl_w.events))

# The same stimulus, the same clamp, the two calls swapped. This is the ticket
# in one pair of runs.
hass_m, ctl_m = asyncio.run(_write_path(moved=True))
print(f"    the same stimulus with the guard consulted BEFORE the write: "
      f"{ctl_m.clamp_wd.engagements} engagement(s), "
      f"{len(clamp_notes(hass_m))} notification(s), and the clamp still "
      f"happened (engaged={ctl_m.clamp_last.engaged})")
check("THE DEFECT, DEMONSTRATED: consulted before the write the guard is "
      "SILENT - self.clamp_last is still None at that point - while the wire "
      "carries a clamped value either way",
      not clamp_notes(hass_m) and ctl_m.clamp_wd.engagements == 0
      and ctl_m.clamp_last.engaged and hass_m.writes[-1] == 2500.0)
check("...and main.py:856's consumer is the line that goes quietly false",
      bool(ctl_w.clamp_last and ctl_w.clamp_last.engaged))


# ==============================================================================
print()
print("7. SAFETY-28: the write path is gated on the allocation record")

# WHY THIS PART IS IN THIS FILE. Part 5 already claims "no path reaches the wire
# around the clamp", and the clamp is the last thing between a bug and the
# hardware. SAFETY-28 adds a second thing on the same path and asks a different
# question of it: not "is this number in range" but "WHO CHOSE THIS NUMBER".
#
# THE DEFECT IT REPLACES. specs §1.5 makes two writers on one actuator a
# non-goal, and until this ticket that was enforced by
# `len(validate.ACTUATOR_WRITE_KEYS) == 1` - one option key, so a second writer
# is not expressible. True, and a statement about the OPTIONS SCHEMA rather than
# about the writes: add CFG-04's per-inverter config and there are two real
# writers with that length still 1. test_validate.py part 2 grades the config
# half; this part grades the write half, which is the one that does not care
# what the options look like.

A_UNIT, B_HYPO = "inverter-a", "inverter-b"


def fresh(**over):
    """A Controller with a stub HA, no cycle run, and therefore NO record."""
    hass = _Hass(0.0)
    c = Controller(dict({"setpoint_entity": "number.sp", "max_w": 2000,
                         "meter_source": "off"}, **over),
                   hass, _Store(), _Pub(), stop=asyncio.Event())
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    return hass, c


# --------------------------------------------------------------------------- #
# 7a. AC 1 - an unauthorised value is REFUSED AT THE WRITE PATH
# --------------------------------------------------------------------------- #
# The second writer, modelled as what it would actually look like in code: some
# other strategy with arithmetic of its own calling the one method that reaches
# the wire. It has a perfectly reasonable number - in range, finite, nothing the
# clamp would object to - and the only thing wrong with it is that nothing
# allocated it.
async def _second_writer():
    hass, c = fresh()
    arm(c, 1500.0)                      # the allocator placed 1500 W this tick
    await c.write_setpoint(1500.0, force=True)      # the loop's own write
    await c.write_setpoint(1200.0, force=True)      # ...and the second writer's
    return hass, c


hass_2w, ctl_2w = asyncio.run(_second_writer())
print(f"    allocator placed +1500 W; a second writer asked for +1200 W; "
      f"the device saw {hass_2w.writes}")
check("AC 1: a value the allocation record does not contain never reaches the "
      "device. 1200 W is in range and finite - the clamp has no opinion about "
      "it - so the ONLY thing that stopped it is that no allocator placed it",
      1200.0 not in hass_2w.writes)
check("...and it is REFUSED, not merely logged: 0 W goes on the wire in its "
      "place. Returning instead would leave the inverter latched on the last "
      "command with the ESP32 heartbeat still being fed - holding an "
      "unauthorised setpoint quietly, which is the outcome this gate exists "
      "to prevent",
      hass_2w.writes == [1500.0, 0.0])
check("...and the action log says so in words an operator can act on",
      any("WRITE REFUSED" in e for e in ctl_2w.events))
check("...while the authorised write of the SAME shape went through "
      "untouched, so the gate is not simply refusing everything",
      hass_2w.writes[0] == 1500.0)

# --------------------------------------------------------------------------- #
# 7b. AC 2 - a record from a previous tick cannot authorise a write in this one
# --------------------------------------------------------------------------- #
# ⚠️ THIS IS THE CACHED-ENVELOPE FAULT WEARING A DIFFERENT HAT. allocator.py's
# rule is that a unit's interval is read per tick and never cached, because a
# pack at its floor or a stale link moves it between one cycle and the next.
# A record is that envelope written down - so code that keeps one and replays it
# is caching the envelope through the record, and the allocator's own purity
# cannot stop it. The tick stamp is what stops it.
async def _replay():
    hass, c = fresh()
    stale = arm(c, 1500.0)              # tick N
    await c.write_setpoint(1500.0, force=True)
    arm(c, 800.0)                       # tick N+1: the envelopes have moved on
    c.allocation = stale                # ...and somebody puts the old one back
    await c.write_setpoint(1500.0, force=True)
    return hass, c, stale


hass_rp, ctl_rp, stale_rec = asyncio.run(_replay())
print(f"    a tick-{stale_rec.tick} record replayed on tick "
      f"{ctl_rp.alloc_tick}: the device saw {hass_rp.writes}")
check("AC 2: replaying the previous tick's record does NOT authorise its "
      "value again - the write is refused and 0 W goes out instead",
      hass_rp.writes == [1500.0, 0.0])
check("...and the refusal names both ticks, so the reason is legible rather "
      "than 'computer says no'",
      "tick" in (ctl_rp.write_refusal(1500.0) or "")
      and str(stale_rec.tick) in (ctl_rp.write_refusal(1500.0) or "")
      and str(ctl_rp.alloc_tick) in (ctl_rp.write_refusal(1500.0) or ""))
check("...and an UNSTAMPED record - one built by a caller that passed no tick "
      "at all - is refused too, because 'I cannot say when I was made' must "
      "fail closed and not read as tick zero",
      ctl_rp.allocation is not None
      and (lambda c: (setattr(c, "allocation",
                              plan(1500.0, {A_UNIT: UnitBounds()}))
                      or c.write_refusal(1500.0) is not None))(ctl_rp))

# ⚠️ AND THE TICK HAS TO ADVANCE ON THE LIVE PATH, WHICH IS A DIFFERENT CLAIM
# FROM ANY OF THE ABOVE. Everything so far drives `arm()`, which bumps the
# counter itself - so all of it would stay green with `self.alloc_tick += 1`
# deleted from cycle(), and the replay protection would be gone on the only path
# that matters. This drives REAL cycles and asks the same question of them.
async def _live_staleness():
    hass = _Hass(6000.0)
    c = Controller({"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
                    "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
                    "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
                    "slew_w": 1000, "step_w": 10, "stale_input_s": 15},
                   hass, _Store(), _Pub(), stop=asyncio.Event())
    c.auto = True
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    await c.read_inputs()
    await c.cycle()
    first, first_v = c.allocation, c.target
    await c.read_inputs()
    await c.cycle()
    # ...and now the HOLD: one missed poll, which must NOT age the record, or a
    # single dropped meter reading would drop the site to 0 W.
    held_before = c.alloc_tick
    c.grid = None
    c.inputs_bad_since = datetime.now(timezone.utc)
    await c.cycle()
    hass.writes.clear()
    await c.write_setpoint(c.target, force=True)     # the heartbeat's own call
    return c, first, first_v, held_before, hass


ctl_ls, rec1, val1, held_before, hass_ls = asyncio.run(_live_staleness())
print(f"    two real cycles: tick {rec1.tick} -> {ctl_ls.allocation.tick}; "
      f"a held cycle left it at {ctl_ls.alloc_tick}")
check("PRECONDITION: two real cycles produced two records with two different "
      "ticks and a value worth replaying",
      rec1.tick != ctl_ls.allocation.tick and val1 != 0.0)
check("the tick advances on the LIVE path, so cycle 1's record does not "
      "authorise a write on cycle 2. Everything above this line drives the "
      "test's own `arm()` and would stay green with the increment deleted from "
      "cycle() - this is the check that would not",
      (lambda c: (setattr(c, "allocation", rec1)
                  or c.write_refusal(val1) is not None))(ctl_ls))
check("⚠️ AND A CYCLE THAT PRODUCED NO RESOLUTION DOES NOT AGE IT. The "
      "'inputs missing, not yet stale' hold returns before the allocator runs, "
      "precisely so one dropped meter reading does not zero the inverter, and "
      "the heartbeat re-affirms self.target through this same gate while it "
      "does. A tick incremented per LOOP ITERATION rather than per RESOLUTION "
      "would refuse that heartbeat and turn the hold into the outage it exists "
      "to prevent",
      ctl_ls.alloc_tick == held_before and hass_ls.writes == [ctl_ls.target]
      and ctl_ls.target != 0.0)

# --------------------------------------------------------------------------- #
# 7c. THE FAILSAFE CANNOT DEADLOCK BEHIND THE GATE
# --------------------------------------------------------------------------- #
# ⚠️ THE HAZARD THIS PART EXISTS FOR. `write_setpoint()` is also how layer 2
# commands 0 W: cycle()'s two refusal branches substitute a 0 W failsafe and
# write it like any other cycle, and shutdown() and the operator's stop button
# do the same. Every one of those runs while the allocator has REFUSED - `ok` is
# False and there are no shares - so a gate that asked the allocator's
# permission to command zero would block the safety write behind the very
# component whose failure the safety write answers. A gate that can do that is
# worse than the hole it closes.
#
# The resolution: 0 W is authorised unconditionally, and it is authorised on the
# VALUE rather than on a `failsafe=True` flag the caller asserts. The argument is
# at Controller.write_refusal(): the hazard §1.5 names is N-times loop gain from
# broadcasting one target to N inverters, and at S = 0 the broadcast and the
# correct allocation COINCIDE - a_i = 1 for every unit gives every unit 0 W,
# which is exactly its share. There is no over-command to make at zero.
_hostile = []
_h, _c = fresh()
_hostile.append(("no record at all", _c))
_h, _c = fresh()
_c.alloc_tick = 1
_c.allocation = plan(1500.0, {A_UNIT: UnitBounds()}, scope_errors=("mis-routed",),
                     tick=1)
_hostile.append(("the allocator REFUSED this tick", _c))
_h, _c = fresh()
_c.allocation = arm(_c, 1500.0)
_c.alloc_tick += 7                      # ...and then seven ticks went by
_hostile.append(("only a stale record", _c))
for _why, _c in _hostile:
    check(f"0 W is authorised with {_why}", _c.write_refusal(0.0) is None)
    check(f"...and a NON-zero write is still refused with {_why}, so the zero "
          f"exit is a floor and not an amnesty",
          _c.write_refusal(1500.0) is not None)


async def _failsafe_wire():
    """The refusal path END TO END, through the real cycle().

    ⚠️ THE ROSTER HERE IS A HYPOTHESIS AND NOTHING IS MEASURED FROM IT.
    INVERTER B DOES NOT EXIST. It is named only so `live_units()` can return
    nothing and drive the allocator into REFUSE_EMPTY, which is the state
    main.py's first refusal branch is written for and the one this check needs:
    the allocator itself has failed, and layer 2 must still be able to command
    zero.
    """
    hass = _Hass(0.0)

    class _Blind(Controller):
        def live_units(self):
            return ()

    c = _Blind({"setpoint_entity": "number.sp", "meter_source": "off",
                "max_w": LIVE_MAX_W, "soc_entity": "sensor.soc",
                "batt_entity": "sensor.batt", "meter_entity": "sensor.grid"},
               hass, _Store(), _Pub(), stop=asyncio.Event())
    c.units = (A_UNIT, B_HYPO)
    c.auto = True
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    for _ in range(3):
        await c.read_inputs()
        await c.cycle()
    return hass, c


hass_fs, ctl_fs = asyncio.run(_failsafe_wire())
print(f"    allocator refused every tick "
      f"({ctl_fs.allocation.refused!r}); the device saw {hass_fs.writes}")
check("the allocator having failed, the 0 W failsafe still REACHES THE WIRE - "
      "this is the deadlock the gate must not have, driven through the real "
      "cycle() and its real refusal branch rather than asserted",
      not ctl_fs.allocation.ok and hass_fs.writes == [0.0]
      and ctl_fs.target == 0.0)
check("...and it was the ALLOCATOR's refusal that produced it, not the gate's "
      "- if the gate were the thing zeroing this cycle the action log would "
      "say WRITE REFUSED and the distinction between the two failures would "
      "be lost to whoever has to debug it",
      any("ARBITER: REFUSED" in e for e in ctl_fs.events)
      and not any("WRITE REFUSED" in e for e in ctl_fs.events))

# --------------------------------------------------------------------------- #
# 7d. `unmet_w` IS NOT `ok`, AND THE GATE MUST NOT FUSE THEM
# --------------------------------------------------------------------------- #
# allocator.py says it plainly: an allocation that could not place every watt is
# still SAFE TO WRITE and strictly better than commanding zero. Fusing the two
# would let ONE SATURATED INVERTER PARK THE WHOLE SITE - the site target goes
# unmet by 3800 W, every share is legal, and a gate reading `unmet_w` would
# answer that by commanding 0 W and meeting even less of it.
_tight = plan(4000.0, {A_UNIT: UnitBounds(lo=-100.0, hi=100.0),
                       B_HYPO: UnitBounds(lo=-100.0, hi=100.0)}, tick=1)
_h, _c = fresh()
_c.alloc_tick, _c.allocation = 1, _tight
print(f"    saturated site: {_tight.explain()}")
check("PRECONDITION: this record is ok AND has watts it could not place",
      _tight.ok and _tight.unmet_w == 3800 and _tight.shares[A_UNIT] == 100)
check("the gate authorises this unit's share anyway. `unmet_w` is not read - "
      "it is the site target that went unmet, not a broken allocation",
      _c.write_refusal(100.0) is None)
check("...and it still refuses a value that is not the share, so 'do not read "
      "unmet_w' has not been implemented as 'read nothing'",
      _c.write_refusal(4000.0) is not None)

# --------------------------------------------------------------------------- #
# 7e. A RECORD THAT CONTRADICTS ITSELF AUTHORISES NOTHING
# --------------------------------------------------------------------------- #
# `UnitPlan.explain()` renders a share outside its own interval as
# "ENVELOPE VIOLATED, DO NOT WRITE". This is the DO NOT WRITE. The shape is the
# half-open clamp defect ALLOC-01 fixed: a plan stating the floor and a share
# below it in the same breath, and agreeing with itself.
#
# ⚠️ UNREACHABLE FROM THE ALLOCATOR AS IT STANDS, AND GRADED ANYWAY. At the
# shipped roster of one the pass-through's interval is (-inf, +inf) and every
# number is inside it. PR #40's reviewer proved what an unreachable ungraded
# branch is worth by replacing one with `if False:` and watching the whole suite
# stay green, so this one is exercised on a constructed record rather than
# trusted.
_liar = Allocation(target_w=500.0, sum_w=500, shares={A_UNIT: 500},
                   plans=(UnitPlan(A_UNIT, 500, 2000.0, math.inf, 1.0,
                                   "weighted share"),), tick=1)
_h, _c = fresh()
_c.alloc_tick, _c.allocation = 1, _liar
print(f"    {_liar.plans[0].explain()}")
check("PRECONDITION: the record is `ok`, and its share IS what the gate would "
      "otherwise match on - so nothing but the envelope question stops it",
      _liar.ok and _liar.shares[A_UNIT] == 500)
check("a share outside its own plan's interval is refused, and the refusal "
      "quotes the plan's own DO NOT WRITE line",
      "DO NOT WRITE" in (_c.write_refusal(500.0) or ""))

# --------------------------------------------------------------------------- #
# 7f. AC 5 - AT N=1 NOTHING CHANGES
# --------------------------------------------------------------------------- #
# The strongest form of this is not a new assertion, it is the 17 other files in
# this suite passing unedited on real cycles. What is added here is the
# statement that the live path is not being quietly zeroed: over part 3's
# sixty-cycle replay - the 10.3 runaway, an inverter that never follows - the
# gate refused nothing at all.
check("AC 5: across part 3's 60 real cycles the gate refused ZERO writes, so "
      "the shipped one-inverter path reaches the wire exactly as it did "
      "before. A gate that silently substituted 0 W here would leave every "
      "clamp check above green and the install dead",
      not [e for e in ctl_r.events if "WRITE REFUSED" in e])
check("...and the numbers those cycles put on the wire are the arbiter's own "
      "target, unrounded and unaltered by the record it now has to appear in",
      hass_r.writes and all(abs(w) <= LIVE_MAX_W for w in hass_r.writes))


print()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")