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.

Six parts, and parts 3 and 4 are the ones that carry SAFETY-03's own
acceptance criteria; part 6 is TEST-11'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().

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


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.
    await c.write_setpoint(9999.0, force=True)
    await c.write_setpoint(-9999.0, force=True)
    await c.write_setpoint(float("nan"), force=True)
    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))

# ==============================================================================
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
    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()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")