goodwe-addon/goodwe_controller/test_delivery.py

"""Runnable check for SAFETY-42 - a unit that stops delivering its share.
`python3 test_delivery.py`

⚠️ THE FIXTURE IS THE CARD, AND IT IS PART 4. What this guard has to do is tell
a WEDGED unit from a COLD PACK AT ITS FLOOR, and those two are IDENTICAL in the
power domain - the same shares, the same measured watts, the same meter, the
same `saturated-freeze`. Measured on admin/goodwe's two-master rig over a
120-cycle run: the two arms agree on every one of 120 cycles in (grid, site
target, per-unit share, per-unit measured power), to the last emitted watt. If
this file's two fixtures ever stop differing, the check has stopped being
graded on the only thing that separates the two cases.

⚠️ AND PART 4b IS THE OTHER HALF OF THE SAME LESSON, ONE LEVEL UP. Part 4 shows
that no threshold on the power domain can separate the two cases; part 4b shows
that a check which only ever runs at the SHIPPED value of a threshold cannot
separate "this call site reads the operator's config" from "this call site has a
number typed into it". Both are the same defect - a fixture whose values cannot
reach the failure it names - and the second one hid two literals from all 23
suites in this repository.

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

from app.control import Tuning
from app.delivery import (
    DELIVERY_OPTION_DEFAULTS, REASON_CODE, DeliveryConfig, DeliveryWatchdog,
    UnitDelivery, account, delinquent, short_by,
)
from app.inverter import LINK_STALE_S
from app.main import Controller
from app.peakguard import discharge_permitted
from app.store import unit_key

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


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


def U(name, share, meas, why=None):
    return UnitDelivery(name, share, meas, why)


# ===========================================================================
print("=" * 72)
print("1. the quantity, and the reduction that makes it readable")
print("=" * 72)

check("short_by is the watts of a DISCHARGE share that did not arrive",
      short_by(2000, 0) == 2000.0 and short_by(2000, 1400) == 600.0)
check("...delivering the share, or more than it, is not a shortfall - both "
      "answer None rather than a number a caller would have to sign-test",
      short_by(2000, 2000) is None and short_by(2000, 2600) is None)
check("...a CHARGE share answers None. There is no configured charge ceiling "
      "in this add-on, so a pack in constant-voltage taper cannot be explained "
      "and arming it would fire on every maintenance charge",
      short_by(-2000, 0) is None and short_by(-2000, -100) is None)
# ⚠️ THE CASE THAT ACTUALLY GRADES THE `share <= 0` GUARD, AND THE MUTATION
# BATTERY IS WHAT FOUND IT MISSING. Relaxing that guard to `share == 0` left the
# two rows above GREEN - for a charge share that is UNDER-delivered,
# `share - measured` is negative and the `gap > 0` test answers None anyway, so
# the mutant is masked. The one input where the two differ is a unit moving MORE
# than its charge share: -2000 W told, -3000 W measured, `share - measured` =
# +1000 W, which the mutant reports as a SHORTFALL. It is not one - it is
# something else driving that pack, which is SAFETY-30's and SAFETY-31's
# question - and reading it as "short of its share" would name the wrong fault.
check("...and a unit moving MORE than its charge share is not short of it "
      "either. This is the input that separates `share <= 0` from `share == 0`; "
      "the two rows above do not, because the `gap > 0` test masks them",
      short_by(-2000, -3000) is None and short_by(-2000, -2600) is None)
check("...a missing reading is None, never 'the whole share'. Zero would be "
      "this check inventing a fault out of an absence",
      short_by(2000, None) is None and short_by(None, 0) is None)

print()
print("   `account()` - the ONLY thing that separates a wedged unit from a")
print("   cold pack, because nothing in the power domain does")
check("a pack at its own cut-out accounts for its shortfall",
      account(live=True, parked=False, soc=10.0, bms_floor=10.0) is not None)
check("...and a pack above it does not",
      account(live=True, parked=False, soc=10.1, bms_floor=10.0) is None)
check("...the boundary is `<=`, the same one discharge_permitted() draws, "
      "BECAUSE IT IS discharge_permitted() - not a transcription of it",
      (account(live=True, parked=False, soc=10.0, bms_floor=10.0) is None)
      == (discharge_permitted(10.0, 10.0) is None)
      and (account(live=True, parked=False, soc=10.1, bms_floor=10.0) is None)
      == (discharge_permitted(10.1, 10.0) is None))
check("a unit the allocator has already dropped accounts for itself - firing "
      "inside a state already being handled is the SAFETY-15/19 failure",
      account(live=False, parked=False, soc=50.0, bms_floor=10.0) is not None)
check("a unit parked at 0 W by a sign trip accounts for itself: not "
      "delivering is compliance",
      account(live=True, parked=True, soc=50.0, bms_floor=10.0) is not None)
check("EVERY UNKNOWN FAILS TOWARD SILENCE - an unreadable SoC and an unstated "
      "floor both account for the shortfall, which is the OPPOSITE direction "
      "to discharge_permitted()'s own caller and is the right one here: an "
      "alarm nobody can justify is the crying-wolf failure this file exists "
      "to avoid",
      account(live=True, parked=False, soc=None, bms_floor=10.0) is not None
      and account(live=True, parked=False, soc=50.0, bms_floor=None) is not None)
print(f"       e.g. {account(live=True, parked=False, soc=9.0, bms_floor=10.0)!r}")

print()
print("   the PROPER-SUBSET rule - what keeps a site-wide cause out")
_wedged = delinquent([U("a", 2000, 2000), U("b", 2000, 0)], 500)
check("one unit short and unexplained beside a sibling that is not: NAMED",
      list(_wedged) == ["b"])
check("...and the sentence names the unit, the two watt figures and the peer, "
      "because 'B is not delivering' and 'B is not delivering while A is' are "
      "different facts and only the second one is this check's conclusion",
      REASON_CODE in _wedged["b"] and "+2000 W" in _wedged["b"]
      and "+0 W" in _wedged["b"] and "a" in _wedged["b"])
print(f"       {_wedged['b']}")
_cold = delinquent([U("a", 2000, 2000), U("b", 2000, 0, "its pack is empty")], 500)
check("THE CARD: the same numbers, with the shortfall ACCOUNTED FOR, name "
      "nobody. This pair of assertions is the whole guard",
      _cold == {})
_both = delinquent([U("a", 2000, 0), U("b", 2000, 0)], 500)
check("EVERY asked unit short and unexplained: NOBODY is named. That is a "
      "site condition - a derate, a curtailment, a plant that cannot - and "
      "this check has no instrument that tells it from every box being wedged "
      "at once. Reporting it would be the saturation-shaped alarm again",
      _both == {})
_three = delinquent([U("a", 2000, 2000), U("b", 2000, 0), U("c", 2000, 0)], 500)
check("...but a proper subset of THREE names both of its members",
      sorted(_three) == ["b", "c"])
_mixed = delinquent([U("a", 2000, 0, "at its floor"), U("b", 2000, 0)], 500)
check("a wedged unit beside a pack that is legitimately empty still names the "
      "wedged one: `asked` is the denominator, not `short`",
      list(_mixed) == ["b"])
check("a shortfall AT the threshold is not one - the bound is strict, so a "
      "unit exactly on the gap the control law calls ordinary tracking is not "
      "reported",
      delinquent([U("a", 2000, 2000), U("b", 2000, 1500)], 500) == {}
      and list(delinquent([U("a", 2000, 2000), U("b", 2000, 1499)], 500)) == ["b"])
check("a unit given no share is not asked and cannot be named, and it is not "
      "a peer either - with only one unit asked there is nothing to compare",
      delinquent([U("a", 0, 0), U("b", 2000, 0)], 500) == {})

print()
print("   AC 5: AT ONE INVERTER IT IS INERT, AND BY ALGEBRA RATHER THAN BY A")
print("   ROSTER GUARD - `len(asked) < 2` is unreachable-by-construction there")
_n1 = [delinquent([U("a", s, m)], 500)
       for s in (0, 500, 2000, 5000, -2000, None)
       for m in (0, 250, 2000, -2000, None)]
check(f"every one of {len(_n1)} single-unit rosters over the whole "
      f"(share, measured) grid answers {{}}", all(r == {} for r in _n1))
check("...including a single unit that IS short and unexplained, which is the "
      "case a caller might expect to fire. It must not: with no sibling, "
      "'this box cannot' and 'this plant cannot' are the same observation",
      delinquent([U("a", 2000, 0)], 500) == {})


# ===========================================================================
print()
print("=" * 72)
print("2. CFG-03: config.yaml is the source of truth and there is ONE copy")
print("=" * 72)
_CFG_TEXT = io.open("config.yaml", encoding="utf-8").read()


def read_option(key, block="options"):
    """The value of `key` inside one top-level block of config.yaml, as text.

    ⚠️ READ OUT OF THE FILE, NOT TRANSCRIBED. CFG-03's whole complaint is that a
    default lives in more than one place; a check that hard-coded the number
    here would be a THIRD copy, and the one that never goes red.

    ⚠️ ANCHORED AT EXACTLY TWO SPACES, WHICH IS CFG-04's LESSON AND NOT A STYLE
    CHOICE. a leading-whitespace pattern matches a PER-INVERTER `max_w` nested under
    `inverters:` as readily as the site one, first hit wins, and a reader that
    quietly answers inverter #1's number is the defect `test_inverter.py`'s
    config.yaml read-closure audit exists to hunt. This function is in that
    audit's table by signature - first argument `key` - so it is fed a
    two-inverter config and graded on what it answers.
    """
    body = _CFG_TEXT.split(f"\n{block}:", 1)[1]
    if block == "options":
        body = body.split("\nschema:", 1)[0]
    m = re.search(rf"^  {re.escape(key)}:\s*(\S+)\s*$", body, re.M)
    return m.group(1) if m else None


check("`delivery_hold_s` is in config.yaml's options: block",
      read_option("delivery_hold_s") is not None)
check("...and DELIVERY_OPTION_DEFAULTS agrees with it, so the running add-on - "
      "which cannot read config.yaml, the Dockerfile does not ship it - uses "
      "the number an operator sees on the Configuration page",
      int(read_option("delivery_hold_s"))
      == int(DELIVERY_OPTION_DEFAULTS["delivery_hold_s"]))
check("...and it has a schema: entry, so the Supervisor refuses a value "
      "outside the range the derivation answers for",
      read_option("delivery_hold_s", "schema") is not None)
_schema = read_option("delivery_hold_s", "schema")
_lo, _hi = (int(x) for x in re.match(r"int\((\d+),(\d+)\)", _schema).groups())
check(f"the schema floor is inverter.LINK_STALE_S = 30 s ({_schema}): below "
      f"the firmware's own veto window a unit not executing its share is a "
      f"LINK fault the add-on already answers by dropping it from the split",
      _lo == 30)
check("the schema ceiling is the 900 s quarter that bills the capacity tariff "
      "- a site running short of one unit's share must be reported well "
      "inside one", _hi == 900)
check("the shipped default is inside its own schema",
      _lo <= int(DELIVERY_OPTION_DEFAULTS["delivery_hold_s"]) <= _hi)
check("THERE IS NO SECOND KEY. The watts half is `Tuning.saturation_w`, read "
      "at runtime, so it cannot drift from the number the control law itself "
      "calls 'not following' - CFG-03 already tracks three places a default "
      "has to be kept in step and this card does not earn a fourth",
      list(DELIVERY_OPTION_DEFAULTS) == ["delivery_hold_s"]
      and not re.search(r"^\s+delivery_w:", _CFG_TEXT, re.M))
print(f"       delivery_hold_s: {DELIVERY_OPTION_DEFAULTS['delivery_hold_s']} s, "
      f"schema {_schema}; watts = Tuning.saturation_w = {Tuning().saturation_w:.0f} W")


# ===========================================================================
print()
print("=" * 72)
print("3. the window and the latch")
print("=" * 72)


def drive(seconds, cfg=None, sentence="short", step=5):
    wd = DeliveryWatchdog(cfg or DeliveryConfig(hold_s=60.0))
    at = None
    for t in range(0, int(seconds) + 1, step):
        if wd.evaluate(sentence, T0 + timedelta(seconds=t)) and at is None:
            at = t
    return wd, at


_wd, _at = drive(300)
check("a shortfall held past the window reports", _wd.tripped)
check("...at the window, not before", _at == 60)
check("...and the reason carries the sentence AND how long it stood",
      "short" in _wd.reason and "held for" in _wd.reason and "60 s" in _wd.reason)
print(f"       {_wd.reason!r}")
_wd2, _at2 = drive(55)
check("a shortfall shorter than the window reports nothing",
      not _wd2.tripped and _at2 is None)

_wd3 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0))
for _t in range(0, 300, 5):
    _wd3.evaluate("short" if (_t // 5) % 2 == 0 else None,
                  T0 + timedelta(seconds=_t))
check("an INTERMITTENT shortfall never accumulates: any clean cycle clears the "
      "window, so this needs a CONTINUOUS episode the way SAFETY-02 and "
      "SAFETY-31 do", not _wd3.tripped and _wd3.since is None)

_wd4, _ = drive(300)
_before = _wd4.reason
_wd4.evaluate(None, T0 + timedelta(seconds=999))
check("it LATCHES - and not for SAFETY-02's reason. Nothing here removes the "
      "evidence, because nothing here changes what is commanded; it latches so "
      "that a unit which stopped delivering for minutes and then resumed "
      "cannot erase its own record",
      _wd4.tripped and _wd4.reason == _before)
_wd4.reset()
check("...and reset() is the operator's press, which clears everything",
      not _wd4.tripped and _wd4.reason == "" and _wd4.since is None)

_wd5 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0))
_wd5.evaluate("short", T0)
_wd5.evaluate("short", T0 + timedelta(seconds=5))
_since = _wd5.since
_wd5.arm(DeliveryConfig(hold_s=120.0))
check("arm() is CONFIGURATION, NOT STATE: it does not touch the window, the "
      "latch or the reason. A detector that cleared its own window every cycle "
      "when it was told what it was measuring would never accumulate a hold - "
      "that is SignWatchdog.arm()'s lesson, found the expensive way",
      _wd5.since == _since and _wd5.cfg.hold_s == 120.0 and not _wd5.tripped)
check("held_s takes the caller's clock rather than reading one",
      _wd5.held_s(T0 + timedelta(seconds=65)) == 65.0)
_wd6 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0), tripped=True, reason="from disk")
check("a latch RESTORED from the store comes up tripped, with its reason",
      _wd6.tripped and _wd6.reason == "from disk")
check("...and a reason with no latch behind it is dropped, not carried",
      DeliveryWatchdog(tripped=False, reason="stale").reason == "")


# ===========================================================================
print()
print("=" * 72)
print("4. END TO END, AND THIS IS THE CARD: a wedged unit and a cold pack at")
print("   its floor are IDENTICAL in the power domain. Only the SoC differs.")
print("=" * 72)
# ⚠️ THE PLANT IS MODELLED AND SAYS SO. `_DeliveryHass` is two inverters that
# obey INSTANTLY - no ramp, no lag - and one meter that is `load - sum(measured)`.
# That is deliberately not a claim about transients: the worst legitimate
# episode a real ramp and lag can produce is measured on admin/goodwe's
# two-master rig and recorded in DECISIONS/SAFETY-42-unit-delivery.md. What this
# fixture is for is the DISCRIMINATION - which of two identical power traces is
# reported - and for that an instant plant is the honest model, because it
# removes the one mechanism that could make an obedient unit look absent.
#
# ⚠️ AND INVERTER B DOES NOT PHYSICALLY EXIST (NOTES.md §1.1b). Every
# two-inverter figure in this file, and every one in the decision record, is
# TWO COPIES OF INVERTER A - `plantsim.INVERTER_B`, `measured=False`.

_LOAD_W = 3000.0
_FLOOR_PCT = 10


class _DeliveryHass:
    """Two inverters, one meter, and unit B that produces nothing.

    `b_soc` is the ONLY thing that differs between the two arms of the
    experiment: 50 % is a wedged unit with a healthy pack, `_FLOOR_PCT` is a
    pack sitting exactly on its own cut-out. Every watt on every wire is the
    same in both.
    """

    def __init__(self, load_w=_LOAD_W, b_soc=50.0):
        self.load_w = float(load_w)
        self.b_soc = float(b_soc)
        self.writes = []
        self.calls = []
        self.ctrl = None

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

    async def limits(self, entity):
        return None

    def measured(self):
        if self.ctrl is None:
            return [0.0, 0.0]
        a = self.ctrl.inverters[0]
        # A obeys. B produces NOTHING, whatever it is told - and whether that is
        # a wedged power stage or an empty pack is not a question its power
        # register can answer.
        return [float((a.last_write if a.last_write is not None else a.target)
                      or 0.0), 0.0]

    async def number(self, entity, invert=False):
        m = self.measured()
        return {"sensor.grid": self.load_w - sum(m),
                "sensor.soc_a": 50.0, "sensor.soc_b": self.b_soc,
                "sensor.batt_a": m[0], "sensor.batt_b": m[1]}.get(entity)

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


def _roster(floors=(_FLOOR_PCT, _FLOOR_PCT)):
    """The two-unit roster. `floors` is (A's own cut-out, B's own cut-out).

    ⚠️ IT IS AN ARGUMENT RATHER THAN `_FLOOR_PCT` TWICE, AND THAT IS HALF OF
    PART 4b. Every fixture in this file used to state 10 %, which is also the
    number config.yaml's own `inverters:` example states, so `bms_floor=10.0`
    written at main.py's call site in place of `inv.bms_soc_floor` answered
    identically in every check here and in the other 22 files. A default that
    happens to agree with the fixture is not a pin.
    """
    return [{"name": n, "setpoint_entity": f"number.sp_{t}",
             "soc_entity": f"sensor.soc_{t}", "batt_entity": f"sensor.batt_{t}",
             "max_w": 2000, "bms_soc_floor": f}
            for (n, t), f in zip((("inverter-a", "a"), ("inverter-b", "b")),
                                 floors)]


def _ctrl(hass, store=None, *, floors=(_FLOOR_PCT, _FLOOR_PCT), **over):
    """A two-inverter Controller, with the NEIGHBOURING guards given long windows.

    ⚠️ THE LONG `circulation_hold_s` AND SIGN HOLDS ARE ISOLATION, NOT A CLAIM
    THAT THOSE GUARDS ARE SILENT HERE - part 7 asserts what each of them
    actually concludes on this same fixture, which is the honest version of the
    claim. What must not happen is the SITE being stopped by another guard while
    this section is trying to grade an attribution, because then every check
    below would be measuring the wrong thing.
    """
    opts = {"meter_entity": "sensor.grid", "meter_source": "off",
            "inverters": _roster(floors), "gain": 0.6, "slew_w": 1000,
            # ⚠️ 1 s, NOT THE SHIPPED 95. This window is WALL CLOCK - the
            # detector takes the caller's `now`, and `_drive()` runs fourteen
            # real cycles 0.2 s apart, so a shipped-length hold would make every
            # check below pass for the wrong reason: nothing would have tripped
            # yet and "no alarm" would read the same as "no fault". The SHIPPED
            # value is graded in part 2 against config.yaml, which is where a
            # number belongs; what is graded here is the attribution.
            "delivery_hold_s": 1,
            "unit_sign_disagree_s": 3600, "site_sign_disagree_s": 3600,
            "circulation_hold_s": 3600, "auto_start": True}
    opts.update(over)
    c = Controller(opts, hass, store if store is not None else _MemStore(),
                   _P(), stop=asyncio.Event())
    for inv in c.inverters:
        inv.dev_min, inv.dev_max = -inv.max_w, inv.max_w
    c.auto = True
    hass.ctrl = c
    return c


class _MemStore:
    """A Store that keeps its dict across "restarts", like the real file does.

    Seeded from `store.DEFAULTS` for test_watchdog.py's reason: a fake that
    hand-listed keys answers None where the real store answers a default, and a
    check reading "this latch was never written" then cannot tell an absent key
    from a False one.
    """

    def __init__(self):
        from app.store import DEFAULTS as _SDEF
        self.data = dict(_SDEF)

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, *a):
        return None


class _P:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


async def _drive(c, n=14, dt=0.2, watch=None):
    seen = []
    for _ in range(n):
        await c.read_inputs()
        await c.cycle()
        if watch is not None:
            seen.append(watch(c))
        await asyncio.sleep(dt)
    return seen


def _sample(c):
    return dict(
        grid=c.grid, target=round(c.target, 6),
        shares=tuple(round(float(i.last_write if i.last_write is not None
                                 else i.target), 6) for i in c.inverters),
        meas=tuple(None if i.batt is None else round(float(i.batt), 6)
                   for i in c.inverters),
        soc=tuple(i.soc for i in c.inverters),
        named=tuple(sorted(i.name for i in c.inverters
                           if i.delivery_wd.tripped)),
        reason=c.reason)


async def _arm(b_soc):
    hass = _DeliveryHass(b_soc=b_soc)
    c = _ctrl(hass)
    return c, hass, await _drive(c, watch=_sample)


_cw, _hw, _sw = asyncio.run(_arm(50.0))          # WEDGED: healthy pack, no output
_cf, _hf, _sf = asyncio.run(_arm(float(_FLOOR_PCT)))   # COLD: pack at its floor


def _power_domain(rows):
    return [(r["grid"], r["target"], r["shares"], r["meas"]) for r in rows]


_pw, _pf = _power_domain(_sw), _power_domain(_sf)
check(f"THE TWO ARMS ARE BYTE-IDENTICAL IN THE POWER DOMAIN on all "
      f"{len(_pw)} cycles - grid, site target, every share, every measured "
      f"watt. No threshold on any of those numbers, at any N and any X, can "
      f"separate a wedged unit from a pack at its floor, and the card's own "
      f"proposed discriminator reads only those numbers",
      len(_pw) == len(_pf) and _pw == _pf)
check("...and the ONLY thing that differs is the state of charge",
      [r["soc"] for r in _sw] != [r["soc"] for r in _sf])
print(f"       both arms, last cycle: grid {_sw[-1]['grid']:.0f} W, shares "
      f"{tuple(round(v) for v in _sw[-1]['shares'])}, measured "
      f"{tuple(round(v) for v in _sw[-1]['meas'])}, reason {_sw[-1]['reason']!r}")
print(f"       SoC: wedged arm {_sw[-1]['soc']}  cold arm {_sf[-1]['soc']}")

check("AC 1: the WEDGED unit is NAMED, and only it",
      _sw[-1]["named"] == ("inverter-b",))
check("AC 2: the COLD PACK AT ITS FLOOR IS NOT - on the identical trace",
      _sf[-1]["named"] == ())
print(f"       {_cw.inverters[1].delivery_wd.reason}")

# ⚠️ THE VACUOUS-PASS GUARD, AND IT IS NOT DECORATION. MUTATION.md's third
# failure mode is a check whose data cannot reach the failure it names: "the
# cold arm named nobody" reads exactly the same against a fixture where nothing
# was short, nothing was asked, or the loop never ran. So the cold arm is
# asserted to have REACHED the branch and been turned away by the account, not
# to have missed it.
_cold_b = _cf.inverters[1]
check("...and the cold arm REACHED that decision rather than falling short of "
      "it: B was asked for a discharge share, was short of it by more than "
      "saturation_w, and was turned away by `account()` alone",
      float(_cold_b.standing_w()) > 0
      and short_by(_cold_b.standing_w(), _cold_b.batt) > _cf.tuning.saturation_w
      and account(live=True, parked=False, soc=_cold_b.soc,
                  bms_floor=_cold_b.bms_soc_floor) is not None)
check("...and the WEDGED arm's B differs from it in exactly that one place",
      account(live=True, parked=False, soc=_cw.inverters[1].soc,
              bms_floor=_cw.inverters[1].bms_soc_floor) is None)

# ⚠️ WHICH FLOOR, AND THIS IS A WIRING CHECK THAT THE PURE ONES ABOVE CANNOT
# MAKE. `peakguard.pack_floor_pct()` conjoins a pack's own cut-out with the
# SITE's `peak_soc_floor`, and handing THAT to `account()` would look correct
# and would be wrong: `peak_soc_floor` bounds a money-saving discharge only
# (SAFETY-25), ordinary self-consumption discharges straight past it, and a
# wedged unit sitting anywhere between the two floors would be EXPLAINED AWAY.
# So this arm puts B exactly there - 15 %, above its own 10 % cut-out and below
# a 20 % `peak_soc_floor` - and it must still be named.


async def _between_floors():
    hass = _DeliveryHass(b_soc=15.0)
    c = _ctrl(hass, peak_soc_floor=20)
    await _drive(c)
    return c


_cbf = asyncio.run(_between_floors())
check("a wedged unit BETWEEN its own cut-out and the site's peak_soc_floor is "
      "still named: the floor this check asks about is the pack's own, not the "
      "peak guard's, and swapping them would hide the fault in a band an "
      "ordinary install has 10 points of",
      _cbf.inverters[1].delivery_wd.tripped
      and _cbf.inverters[1].soc == 15.0
      and float(_cbf.o.get("peak_soc_floor")) == 20.0)


# ===========================================================================
print()
print("=" * 72)
print("4b. THE NUMBERS THIS CHECK READS ARE THE CONFIGURED ONES, AND THAT IS")
print("    GRADED AT A CONFIG THAT IS NOT THE SHIPPED ONE")
print("=" * 72)
# ⚠️ WHY THIS PART EXISTS, AND IT IS NOT A MUTATION SCORE. Two of the values
# main.py hands this check are OPTIONS, and the paragraph above
# `delivery.MIN_W_IS_SATURATION_W` and the `delivery_hold_s` block in
# config.yaml both make a PROMISE about them in shipped, operator-readable
# prose: there is no fourth copy of a default to keep in step, and no way for
# the two to drift, because an operator who raises `saturation_w` has told the
# loop that a bigger gap is ordinary tracking and this check moves with them BY
# CONSTRUCTION. A literal written at the call site breaks exactly that promise,
# silently, and the promise is why the design was chosen over a `delivery_w`
# key.
#
# ⚠️ AND EVERY FIXTURE ABOVE IS BLIND TO IT, WHICH IS THIS REPOSITORY'S
# RECURRING DEFECT (MUTATION.md's third failure mode): a fixture whose values
# cannot reach the failure it names. Part 4's gaps are all far above BOTH the
# shipped `saturation_w` and any smaller literal, and every `bms_soc_floor` in
# this file was 10 - the number config.yaml's own example states. So
#
#     self.tuning.saturation_w      ->  250.0        (-19 B)
#     bms_floor=inv.bms_soc_floor   ->  10.0         (-13 B)
#
# were BOTH survivors of the whole 23-file suite, not of this file alone. The
# two blocks below are what kills them, and they kill not just those two
# literals but EVERY literal: each pair of arms is answered one way at one
# configured value and the other way at another, on the same fixture, so no
# single constant can satisfy both rows.

print()
print("   (a) THE WATTS - `saturation_w`, raised to a value inside its own")
print("       schema. Same load, same wedged unit, two configured thresholds.")

# 1800 is inside the shipped schema `int(100,2000)` and is not the shipped
# default - both asserted below rather than asserted here in a comment. The
# load is what puts B's shortfall BETWEEN the two: above the shipped threshold,
# below the raised one. The gap itself is printed by the run that produces it
# and is never written down here; what is asserted is the ORDERING.
_SAT_RAISED = 1800
_SAT_LOAD_W = 1600.0
_SAT_SHIPPED = int(read_option("saturation_w"))
_sat_lo, _sat_hi = (int(x) for x in re.match(
    r"int\((\d+),(\d+)\)", read_option("saturation_w", "schema")).groups())


async def _at_saturation_w(sat):
    hass = _DeliveryHass(load_w=_SAT_LOAD_W)
    c = _ctrl(hass, saturation_w=sat)
    return c, await _drive(c, watch=_sample)


_c_ship, _r_ship = asyncio.run(_at_saturation_w(_SAT_SHIPPED))
_c_up, _r_up = asyncio.run(_at_saturation_w(_SAT_RAISED))
_b_ship, _b_up = _c_ship.inverters[1], _c_up.inverters[1]
_gap_ship = short_by(_b_ship.standing_w(), _b_ship.batt)
_gap_up = short_by(_b_up.standing_w(), _b_up.batt)

check(f"the raised threshold is a config an operator can actually enter: "
      f"{_SAT_RAISED} W is inside the shipped schema "
      f"{read_option('saturation_w', 'schema')}, and it is NOT the shipped "
      f"default ({_SAT_SHIPPED} W) - both read out of config.yaml, not typed "
      f"here",
      _sat_lo <= _SAT_RAISED <= _sat_hi and _SAT_RAISED != _SAT_SHIPPED)
check("...and the controller really is running on it - `Tuning.saturation_w` "
      "carries the option through, so this is a changed CONFIG and not a "
      "changed test",
      _c_up.tuning.saturation_w == float(_SAT_RAISED)
      and _c_ship.tuning.saturation_w == float(_SAT_SHIPPED))

# ⚠️ THE VACUOUS-PASS GUARD AGAIN, and here it is the whole load. "the raised
# arm named nobody" reads identically against a fixture where B was never
# asked, was not short, or was explained away. So the raised arm is asserted to
# have reached the threshold comparison and to have been turned away by NOTHING
# BUT the number.
check("the raised arm REACHED the comparison: B was asked for a discharge "
      "share, is short of it, is live, is not parked, and NOTHING accounts for "
      "it - the only thing standing between this unit and an alarm is the "
      "configured watts",
      float(_b_up.standing_w()) > 0 and _gap_up is not None
      and not _b_up.sign_wd.tripped
      and account(live=True, parked=False, soc=_b_up.soc,
                  bms_floor=_b_up.bms_soc_floor) is None)
check("...and its shortfall lands STRICTLY BETWEEN the two configured "
      "thresholds, which is the one place the two answers can differ. Part 4's "
      "fixtures are all above both, which is exactly why they cannot see this",
      _gap_up is not None and float(_SAT_SHIPPED) < _gap_up < float(_SAT_RAISED))

check("AT THE SHIPPED THRESHOLD THE UNIT IS NAMED",
      _r_ship[-1]["named"] == ("inverter-b",))
check("AT THE RAISED THRESHOLD, ON THE SAME FIXTURE, IT IS NOT - the operator "
      "has told the loop that a gap this size is ordinary tracking, and this "
      "check moved with them. THIS PAIR IS THE PIN: a literal at the call site "
      "is one number, and no one number answers both of these rows",
      _r_up[-1]["named"] == ())
print(f"       shipped saturation_w {_SAT_SHIPPED} W: B short by "
      f"{_gap_ship:.0f} W -> named {_r_ship[-1]['named']}")
print(f"       raised  saturation_w {_SAT_RAISED} W: B short by "
      f"{_gap_up:.0f} W -> named {_r_up[-1]['named']}")

# And the same two answers out of the pure function, on the raised arm's OWN
# two watt figures - so the flip is a property of `delinquent()`'s threshold
# and not of anything the controller did differently between the two runs.
_pair = [U("inverter-a", float(_c_up.inverters[0].standing_w()),
           _c_up.inverters[0].batt),
         U("inverter-b", float(_b_up.standing_w()), _b_up.batt)]
check("...and `delinquent()` says the same thing about that one pair of watt "
      "figures: named at the shipped number, silent at the raised one. The "
      "threshold is the only input that moved",
      list(delinquent(_pair, _SAT_SHIPPED)) == ["inverter-b"]
      and delinquent(_pair, _SAT_RAISED) == {})

print()
print("   (b) THE PERCENT - each unit's OWN `bms_soc_floor`, at two cut-outs")
print("       that BRACKET the shipped example from both sides.")

# ⚠️ WHY TWO ARMS AND NOT ONE. A single arm at a deeper cut-out kills `10.0`
# but not `50.0`; a single arm at a shallower one kills `10.0` but not `0.0`.
# One arm on each side of the shipped example leaves no constant at all: the
# deep arm must stay SILENT (its pack is genuinely below its own cut-out) and
# the shallow arm must SPEAK (its pack is genuinely above its own cut-out), and
# a literal that satisfies the first cannot satisfy the second.
_FLOOR_DEEP, _DEEP_SOC = 25, 15.0        # pack BELOW its own cut-out: cold
_FLOOR_SHALLOW, _SHAL_SOC = 5, 8.0       # pack ABOVE its own cut-out: wedged
_floor_lo, _floor_hi = (int(x) for x in re.match(
    r"int\((\d+),(\d+)\)", read_option("bms_soc_floor", "schema")).groups())


async def _at_floor(b_floor, b_soc, peak):
    hass = _DeliveryHass(b_soc=b_soc)
    # ⚠️ `peak_soc_floor` IS RAISED WITH THE DEEP ARM SO THE FIXTURE STAYS A
    # CONFIG THE ADD-ON WOULD START ON: validate.py rule 8 refuses a
    # `peak_soc_floor` at or below ANY unit's `bms_soc_floor`. A fixture that
    # only a refused config can produce would grade a state no install reaches.
    c = _ctrl(hass, floors=(_FLOOR_PCT, b_floor), peak_soc_floor=peak)
    return c, await _drive(c, watch=_sample)


_c_deep, _r_deep = asyncio.run(_at_floor(_FLOOR_DEEP, _DEEP_SOC, 30))
_c_shal, _r_shal = asyncio.run(_at_floor(_FLOOR_SHALLOW, _SHAL_SOC, 20))
_bd, _bs = _c_deep.inverters[1], _c_shal.inverters[1]

check(f"both cut-outs are configs an operator can enter - inside the shipped "
      f"schema {read_option('bms_soc_floor', 'schema')} - and NEITHER is the "
      f"{read_option('bms_soc_floor')} % this file and config.yaml's own "
      f"example both state, which is the number a literal would have agreed "
      f"with everywhere",
      _floor_lo <= _FLOOR_SHALLOW < int(read_option("bms_soc_floor"))
      < _FLOOR_DEEP <= _floor_hi)
check("...and each arm is a config validate.py rule 8 accepts: its "
      "peak_soc_floor is above EVERY unit's own cut-out",
      all(float(c.o.get("peak_soc_floor")) > float(i.bms_soc_floor)
          for c in (_c_deep, _c_shal) for i in c.inverters))
check("...and the two arms are IDENTICAL IN THE POWER DOMAIN on every cycle - "
      "grid, site target, every share, every measured watt. The configured "
      "cut-out is the only thing that differs, which is part 4's claim made "
      "again at two cut-outs that are not the shipped one",
      _power_domain(_r_deep) == _power_domain(_r_shal))

check(f"a pack at {_DEEP_SOC:.0f} % against its OWN {_FLOOR_DEEP} % cut-out is "
      f"cold, and is NOT named - even though it is far above the "
      f"{read_option('bms_soc_floor')} % a literal would have graded it "
      f"against, which would have called an empty pack a fault",
      _r_deep[-1]["named"] == () and float(_bd.bms_soc_floor) == _FLOOR_DEEP
      and _bd.soc == _DEEP_SOC)
check("...and it REACHED that decision: it was asked for a discharge share, "
      "was short of it by more than saturation_w, and was turned away by "
      "`account()` alone",
      float(_bd.standing_w()) > 0
      and short_by(_bd.standing_w(), _bd.batt) > _c_deep.tuning.saturation_w
      and account(live=True, parked=False, soc=_bd.soc,
                  bms_floor=_bd.bms_soc_floor) is not None)
check(f"a pack at {_SHAL_SOC:.0f} % against its OWN {_FLOOR_SHALLOW} % cut-out "
      f"is WEDGED, and IS named - even though it is below the "
      f"{read_option('bms_soc_floor')} % a literal would have explained it "
      f"away with, which would have lost the fault this card exists to find. "
      f"THIS ROW AND THE ONE ABOVE ARE THE PIN: no constant answers both",
      _r_shal[-1]["named"] == ("inverter-b",)
      and float(_bs.bms_soc_floor) == _FLOOR_SHALLOW and _bs.soc == _SHAL_SOC)
print(f"       deep    cut-out {_FLOOR_DEEP} %, SoC {_DEEP_SOC:.0f} % -> named "
      f"{_r_deep[-1]['named']}")
print(f"       shallow cut-out {_FLOOR_SHALLOW} %, SoC {_SHAL_SOC:.0f} % -> "
      f"named {_r_shal[-1]['named']}")


# ===========================================================================
print()
print("=" * 72)
print("5. IT REPORTS AND DOES NOT ACT - asserted by counterfactual, not by")
print("   reading the code")
print("=" * 72)


async def _never_trips():
    hass = _DeliveryHass(b_soc=50.0)
    c = _ctrl(hass, delivery_hold_s=900)     # the schema ceiling: never, here
    return c, await _drive(c, watch=_sample)


_cn, _sn = asyncio.run(_never_trips())
check("the guard TRIPPED in one run and not in the other, so the comparison "
      "below is between two different states of this detector",
      _sw[-1]["named"] == ("inverter-b",) and _sn[-1]["named"] == ()
      and _cw.inverters[1].delivery_wd.tripped
      and not _cn.inverters[1].delivery_wd.tripped)
check("...and EVERY WATT IS THE SAME. Grid, site target, both shares and both "
      "measured powers are identical cycle for cycle whether this guard "
      "reported or not - which is what 'it does not act' means, stated as a "
      "measurement instead of as an intention",
      _power_domain(_sw) == _power_domain(_sn))
check("...the wedged unit is still being commanded its share after the trip, "
      "not parked: this is NOT SAFETY-02's remedy and must not become it",
      abs(float(_cw.inverters[1].target)) > 0)
check("...and nothing named `delivery` reaches the arbiter - the resolution's "
      "own sentence is unchanged by the trip",
      "deliver" not in _cw.reason.lower() and _cw.reason == _cn.reason)
_pay = _cw.status_payload()
check("...and control is still running, which is the whole difference from "
      "every other latched guard in this add-on",
      _cw.auto and _pay["level"] != "bad")
check("...but the banner does NOT go on saying 'Running - holding grid at "
      "zero'. That sentence is a claim the site cannot make while a unit is "
      "absent, and 'the page reads healthy while the fault runs' is the shape "
      "this whole card exists to remove. It names the unit and stays a WARNING",
      _pay["level"] == "warn" and "NOT DELIVERING" in _pay["banner"]
      and "inverter-b" in _pay["banner"]
      and "holding grid at zero" not in _pay["banner"])
print(f"       banner: {_pay['level']!r} / {_pay['banner']!r}")
_row = [r for r in _cw.checks() if "NOT DELIVERING" in r["text"]]
check("the status page carries it as a CAUTION, not a failure: the site is "
      "still running and short, which a red cross would misdescribe",
      len(_row) == 1 and _row[0]["ok"] is False and _row[0]["warn"] is True)


# ===========================================================================
print()
print("=" * 72)
print("6. what the NEIGHBOURING guards conclude on this same fixture - the")
print("   honest version of 'every existing guard is silent'")
print("=" * 72)
_b = _cw.inverters[1]
check("SAFETY-30 per unit: SILENT. It needs OPPOSITION on both sides of "
      f"{_cw.unit_sign_cfg.min_w:.0f} W, and a unit at 0 W is not opposing - "
      "it is absent, so the gate never opens",
      not _b.sign_wd.tripped and _b.sign_wd.disagree_since is None)
check("SAFETY-30 site: SILENT. The sum of the standing commands still agrees "
      "with the meter in DIRECTION; only the magnitude is short",
      not _cw.site_sign_wd.tripped and _cw.site_sign_wd.disagree_since is None)
check("SAFETY-31: SILENT, and identically so. sum(|Bi|) - |sum(Bi)| is ZERO "
      "when one unit does nothing - circulation needs two units pulling "
      "AGAINST each other",
      _cw.site_circulating_w == 0.0 and not _cw.circ_wd.tripped)
check("the saturation freeze: FIRES - and that is the point. It is the one "
      "signal that does, and clamp.py:93-96 documents it as the alarm an "
      "operator learns to ignore, which is why this card needed a different "
      "signal rather than a louder one",
      _cw.sat_count >= _cw.tuning.saturation_cycles)
print(f"       saturation counter {_cw.sat_count} (freezes at "
      f"{_cw.tuning.saturation_cycles}); site commanded "
      f"{_cw.target:.0f} W against {_cw.site_batt_w:.0f} W measured")
check("...and it fires IDENTICALLY on the benign arm, on the identical trace, "
      "at the same counter. A saturation-shaped alarm cannot tell these two "
      "apart at any threshold and any duration",
      _cf.sat_count == _cw.sat_count
      and _cf.site_batt_w == _cw.site_batt_w
      and _cf.target == _cw.target)

print()
print("   ...and a unit the ALLOCATOR HAS ALREADY DROPPED is not reported twice")
# ⚠️ A WIRING CHECK THE PURE ONE IN PART 1 CANNOT MAKE, AND THE MUTATION BATTERY
# IS WHAT FOUND IT MISSING. `account(live=False, ...)` is graded up there; what
# is graded HERE is that `cycle()` actually ASKS `inv.is_live(now)`. Replacing
# that call with a literal `True` left every check green, because no fixture had
# a unit whose writes were being rejected. A dropped unit still has its last
# ACCEPTED command standing on the wire (`write_setpoint` leaves `last_write`
# alone on a rejected write) and still measures nothing, so it looks exactly
# like a wedged one - and reporting it would be a second, louder alarm about a
# state the allocator has already handled and logged, which is the
# SAFETY-15/SAFETY-19 failure this module takes its 30 s floor from.


class _LinkDeadHass(_DeliveryHass):
    """B is healthy, then its LINK dies: writes rejected and output gone.

    ⚠️ `rejected_since` IS BACK-DATED RATHER THAN WAITED OUT. `is_live()`'s
    window is `LINK_STALE_S` = 30 s of wall clock and this fixture runs in about
    two; waiting it out would make the check's runtime the thing it grades. What
    is graded is the CALL, not the timer - `test_inverter.py` part 3 owns the
    timer.
    """

    def __init__(self, dead_from=5, **kw):
        super().__init__(**kw)
        self.dead_from = int(dead_from)
        self.cycle_n = 0

    def measured(self):
        a, b = self.ctrl.inverters
        w = lambda i: float((i.last_write if i.last_write is not None
                             else i.target) or 0.0)
        return [w(a), 0.0 if self.cycle_n >= self.dead_from else w(b)]

    async def set_number(self, entity, value):
        self.writes.append((entity, float(value)))
        if entity == "number.sp_b" and self.cycle_n >= self.dead_from:
            return False
        return True


async def _link_dead():
    hass = _LinkDeadHass(dead_from=5)
    c = _ctrl(hass)
    b = c.inverters[1]
    seen = []
    for n in range(18):
        hass.cycle_n = n
        if n == 5:
            # the moment the link dies, back-dated past the firmware's window
            b.rejected_since = (datetime.now(timezone.utc)
                                - timedelta(seconds=LINK_STALE_S + 10))
        await c.read_inputs()
        await c.cycle()
        seen.append(dict(live=b.is_live(datetime.now(timezone.utc))[0],
                         stood=float(b.standing_w()), meas=b.batt,
                         named=b.delivery_wd.tripped))
        await asyncio.sleep(0.15)
    return c, seen


_cld, _sld = asyncio.run(_link_dead())
_bld = _cld.inverters[1]
check("the fixture REACHES the case: B is not live, is still standing at a "
      "discharge command, and is measuring nothing - i.e. it looks exactly like "
      "a wedged unit to every number this check reads",
      not _sld[-1]["live"] and _sld[-1]["stood"] > _cld.tuning.saturation_w
      and _sld[-1]["meas"] == 0.0)
check("...and it is NOT reported, because `account()` was asked and answered. "
      "The allocator has already dropped it and said so in the action log; a "
      "second alarm about that is the SAFETY-15/19 failure",
      not _bld.delivery_wd.tripped and _bld.delivery_wd.since is None)
check("...and the answer is the LINK one specifically, not the SoC one - so a "
      "reader can tell which branch withheld the alarm",
      "link" in (account(live=False, parked=False, soc=_bld.soc,
                         bms_floor=_bld.bms_soc_floor) or ""))
print(f"       B: live={_sld[-1]['live']} standing={_sld[-1]['stood']:.0f} W "
      f"measured={_sld[-1]['meas']:.0f} W -> not reported")

print()
print("   ...and the one case where BOTH have something to say: a unit that is")
print("   REVERSED is short too, and SAFETY-30 gets there first")
# ⚠️ THE OVERLAP IS REAL AND IT RESOLVES ITSELF, WHICH IS WORTH ASSERTING
# RATHER THAN ASSUMING. A unit measured at -2000 W against a +2000 W share is
# OPPOSING (SAFETY-30's fault) and is also short by 4000 W with nothing to
# explain it (this one). SAFETY-30's hold is shorter, it parks the unit, the
# share goes to 0 W - and this check's gate closes on `share > 0` while
# `account()` answers "parked". Two alarms for one fault is how an operator
# learns to skim, and this is the arrangement that prevents it.


class _ReversedHass(_DeliveryHass):
    """Unit B does the OPPOSITE of what it is told - SAFETY-30's fault."""

    def measured(self):
        a, b = self.ctrl.inverters
        w = lambda i: float((i.last_write if i.last_write is not None
                             else i.target) or 0.0)
        return [w(a), -w(b)]


async def _reversed_run():
    hass = _ReversedHass(b_soc=50.0)
    # SAFETY-30's per-unit hold SHORTER than this check's, which is the shipped
    # ordering too: 85 s against 185 s.
    c = _ctrl(hass, unit_sign_disagree_s=1, delivery_hold_s=2)
    seen = await _drive(c, n=24, dt=0.15, watch=lambda c: dict(
        sign=c.inverters[1].sign_wd.tripped,
        deliv=c.inverters[1].delivery_wd.tripped,
        share=float(c.inverters[1].target),
        acct=account(live=True, parked=c.inverters[1].sign_wd.tripped,
                     soc=c.inverters[1].soc,
                     bms_floor=c.inverters[1].bms_soc_floor)))
    return c, seen


_crev, _srev = asyncio.run(_reversed_run())
check("a REVERSED unit is short by more than saturation_w too, so both guards "
      "have evidence - the fixture reaches the overlap rather than dodging it",
      any(r["deliv"] or r["sign"] for r in _srev)
      and any(abs(r["share"]) > _crev.tuning.saturation_w for r in _srev))
check("SAFETY-30 gets there first and PARKS it - its hold is shorter, which is "
      "also the shipped ordering (85 s against 185 s)",
      _crev.inverters[1].sign_wd.tripped)
check("...and once parked, this check goes quiet on its own: the share is 0 W "
      "so it is no longer ASKED, and `account()` answers 'parked' anyway. ONE "
      "alarm for one fault",
      not _crev.inverters[1].delivery_wd.tripped
      and _crev.inverters[1].delivery_wd.since is None
      and float(_crev.inverters[1].target) == 0.0
      and _srev[-1]["acct"] is not None)
print(f"       shipped ordering: unit sign hold "
      f"{_crev.unit_sign_cfg.hold_s:.0f} s in this fixture, delivery hold "
      f"{_crev.delivery_cfg.hold_s:.0f} s; account now {_srev[-1]['acct']!r}")


# ===========================================================================
print()
print("=" * 72)
print("7. AC 5: AT ONE INVERTER NOTHING CHANGES - asserted, not assumed")
print("=" * 72)


class _OneHass(_DeliveryHass):
    """One inverter that produces nothing against a non-zero share."""

    def measured(self):
        return [0.0]

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


async def _one():
    hass = _OneHass()
    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
            "meter_source": "off", "max_w": 2000, "gain": 0.6,
            # ⚠️ THE SAME 1 s AS THE TWO-INVERTER FIXTURE, AND THAT IS WHAT
            # MAKES THIS SECTION NON-VACUOUS. The run below is 2 s of wall clock
            # against a 1 s window, so a hold is not what stops this firing -
            # the reduction is. With a shipped-length hold "nothing was
            # reported" would be true of a fixture that simply ran out of time.
            "bms_soc_floor": _FLOOR_PCT, "delivery_hold_s": 1,
            "sign_disagree_s": 3600, "auto_start": True}
    store = _MemStore()
    c = Controller(opts, hass, store, _P(), stop=asyncio.Event())
    c.inverters[0].dev_min, c.inverters[0].dev_max = -2000.0, 2000.0
    c.auto = True
    hass.ctrl = c
    await _drive(c, n=20, dt=0.1)
    return c, hass, store


_c1, _h1, _st1 = asyncio.run(_one())
_only = _c1.inverters[0]
check("a roster of ONE, whose only unit is standing at a discharge share and "
      "delivering nothing for the whole run - the exact fault, at N=1",
      len(_c1.inverters) == 1 and float(_only.standing_w()) > 0
      and _only.batt == 0.0
      and short_by(_only.standing_w(), _only.batt) > _c1.tuning.saturation_w)
check("...and NOTHING is reported. With no sibling, 'this box cannot' and "
      "'this plant cannot' are the same observation, and the saturation freeze "
      "is still the whole answer there", not _only.delivery_wd.tripped)
check("...the window was never even opened", _only.delivery_wd.since is None)
check("...and no store key was written, so a one-inverter install's state file "
      "is byte-identical to what it was before this card",
      unit_key(_only.name, "delivery_tripped") not in _st1.data
      and unit_key(_only.name, "delivery_reason") not in _st1.data)
_n1row = [r for r in _c1.checks() if "delivery watchdog" in r["text"]]
check("...and the status page says n/a rather than 'armed', which would be a "
      "claim to be watching something that cannot happen",
      len(_n1row) == 1 and _n1row[0]["ok"] and "n/a at one inverter" in _n1row[0]["text"])
check("...no persistent notification of this kind was raised either",
      not [c for c in _h1.calls
           if "not_delivering" in str(c[2].get("notification_id", ""))])


# ===========================================================================
print()
print("=" * 72)
print("8. SAFETY-34: the report survives a Home Assistant restart, which is")
print("   how the 2026-09-02 outage went unseen for 6 h 25 m")
print("=" * 72)


async def _restart():
    shared = _MemStore()
    h1 = _DeliveryHass(b_soc=50.0)
    c1 = _ctrl(h1, store=shared)
    await _drive(c1)
    # A NEW process, a NEW HomeAssistant object (so the in-memory notification
    # is gone with it), the SAME durable file. That is what a Home Assistant
    # restart is from this add-on's point of view: the add-on goes with it.
    h2 = _DeliveryHass(b_soc=50.0)
    c2 = _ctrl(h2, store=shared)
    await c2.check_not_delivering()
    return c1, h1, c2, h2, shared


_r1, _rh1, _r2, _rh2, _rs = asyncio.run(_restart())
_key = unit_key("inverter-b", "delivery_tripped")
check("the latch reached the durable file, namespaced under the unit it is "
      "about", _rs.data.get(_key) is True)
check("...with its reason, and the reason was written FIRST - a crash between "
      "the two must not persist a latch nobody can explain",
      bool(_rs.data.get(unit_key("inverter-b", "delivery_reason"))))
check("...and there is NO flat mirror of it, where SAFETY-02 has one: the "
      "version a rollback would land on drives exactly one actuator, a roster "
      "at which this check is inert by construction",
      "delivery_tripped" not in _rs.data and "delivery_reason" not in _rs.data)
check("a fresh process comes up with the report STILL STANDING",
      _r2.inverters[1].delivery_wd.tripped)
check("...and RE-RAISES the notification, because a persistent_notification is "
      "in memory and did not survive - that is the whole of SAFETY-34",
      any(c[0] == "persistent_notification"
          and "not_delivering" in str(c[2].get("notification_id", ""))
          for c in _rh2.calls))
_note = [c for c in _rh2.calls
         if "not_delivering" in str(c[2].get("notification_id", ""))][0]
check("the notification NAMES THE UNIT - in its id, its title and its body - "
      "which is card requirement 2 and the whole reason SAFETY-30 split "
      "per-unit from site",
      "inverter-b" in _note[2]["notification_id"].replace("_", "-")
      and "inverter-b" in _note[2]["title"]
      and "inverter-b" in _note[2]["message"])
check("...and it says, in the operator's own sentence, that nothing has been "
      "commanded differently. An alarm that let somebody believe the add-on "
      "had handled it would be worse than no alarm",
      "NOTHING HAS BEEN COMMANDED DIFFERENTLY" in _note[2]["message"])
print(f"       id={_note[2]['notification_id']!r} title={_note[2]['title']!r}")
_before_calls = len(_rh2.calls)
await_again = asyncio.run(_r2.check_not_delivering())
check("...and it is raised ONCE per process, not once per cycle: the alarm "
      "storm this add-on has argued against twice already",
      len(_rh2.calls) == _before_calls)

_r2.auto = False
asyncio.run(_r2.handle_action("auto_toggle"))     # off
asyncio.run(_r2.handle_action("auto_toggle"))     # and on again: the press
check("the operator's START press is the only way out, and it clears the "
      "latch, the reason and the 'a human has seen this' flag together",
      not _r2.inverters[1].delivery_wd.tripped
      and _rs.data.get(_key) is False
      and _rs.data.get(unit_key("inverter-b", "delivery_reason")) == ""
      and _r2._delivery_announced == set())


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