"""Runnable check for DEV-02 - per-instance Inverter objects.  `python3 test_inverter.py`

THE DEFECT, in one sentence: every piece of per-device state in the Controller
was a SCALAR, and several of those scalars are safety inputs - so a second
inverter would have been graded against the first one's numbers.

Twelve parts, and what each one is deliberately NOT allowed to be. Parts 1-8
and 12 are DEV-02's; parts 9-11 are CFG-04's, which is the card that gave an
OPERATOR a way to declare the second inverter DEV-02 made the code ready for.

  1. AC 1. THE GRADER, over `app/main.py` itself. Read, not run: the claim is
     "no per-device value remains a Controller scalar", which is a statement
     about EVERY configuration and not about the ones a fixture reaches. It is
     an AST scan plus a live `vars()` comparison, and the two halves catch
     different re-introductions - a field assigned in `__init__` and a field
     grown at runtime.
  2. AC 2. THE STORE MIGRATION, and it is graded rather than performed. A latch
     that silently fails to migrate is a DISARMED SAFETY GUARD on an install
     whose state file says it is protected, so the migration is BROKEN here and
     the guard is watched to fire.
  3. AC 3. A TRIP ON B DOES NOT LATCH A, driven through real cycles with real
     store keys, then through a real RESTART - because the whole reason
     SAFETY-02's latch is persisted is that a restart is the first thing an
     operator tries.
  4. AC 4. EACH RATING FROM ITS OWN ENTITY, and the clamp, the rail and
     `achievable_w` that follow it. Inverter B is given a DIFFERENT rating from
     inverter A on purpose: a check where both are 5000 W cannot tell "B's
     number" from "A's number".
  5. AC 5. `actual_w` IS SITE-LEVEL, pinned on main.py's own call site. The
     measurement behind it is test_alloc.py part 8 and the rig; what is pinned
     here is that the loop is fed the SUM.
  6. AC 6. INDEPENDENT HEARTBEATS, with the hang INJECTED. A wall-clock
     experiment against a transport that never answers, and the counterfactual -
     the serialised write this replaces - run beside it so the number means
     something.
  7. AC 7. `live_units()` ANSWERS HONESTLY, on the shipped method.
  8. AC 8. N=1 IS UNCHANGED. The rest of this repository's 17 files are the real
     evidence for that and they run unmodified; what is here is the part they
     cannot see - that the compatibility views REFUSE at N>=2 rather than
     answering for the first inverter.
  9. CFG-04. A TWO-INVERTER CONFIG, through the shipped `build_roster()`, into
     a roster the shipped `live_units()` answers for - and the other half of it,
     that an EMPTY `inverters:` is byte-identical to what shipped before.
 10. CFG-04. RULE 9, and SAFETY-28's two-writers hazard, which is expressible
     for the first time in this add-on's life. Also the two PERMISSIVE READERS
     this card found in shipped code - rule 5 grading against the flat `max_w`,
     and run_control() skipping the whole range re-check when the flat
     `setpoint_entity` is blank - both run BOTH WAYS, because a refusal that
     fires on everything passes a check that only looks for the refusal.
 11. CFG-04. EVERY config.yaml READER IN THE SUITE, FED A TWO-INVERTER CONFIG.
     Not a lint over their source: the functions are extracted and EXECUTED
     against a synthetic file whose per-inverter numbers differ from its flat
     ones, because on the shipped file a first-match reader is wrong and right
     at the same time.
 12. MUTATION. Which check bites for each way of defeating this card.

⚠️ INVERTER B DOES NOT EXIST YET (~2 weeks out; SPIKE-04 measures it). Every
second unit below is built from parameters passed IN, and where a number is
needed it is a HYPOTHESIS marked as one. Nothing here is a measured parameter
for a second device, and no check depends on one.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you.
Read MUTATION.md before you report a survivor.
"""

import ast
import asyncio
import inspect
import io
import sys
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

from named_absence import first, given, nth, present

from app import main as app_main
from app.inverter import LINK_STALE_S, UNIT_A, Inverter, build_roster
from app.main import WRITE_BUDGET_S, Controller
from app.store import (
    DEFAULTS, LEGACY_UNIT_KEYS, SCHEMA_KEY, SCHEMA_VERSION, migrate_unit_keys,
    unit_key,
)
from app.validate import FIRMWARE_INGEST_VETO_S
from app.freeze import rail_w
from app.arbiter import P_LOOP, P_SAFETY, Claim, resolve

fails = []
HERE = Path(__file__).resolve().parent
MAIN_SRC = io.open(HERE / "app" / "main.py", encoding="utf-8").read()
MAIN_AST = ast.parse(MAIN_SRC)
CTRL = [n for n in MAIN_AST.body
        if isinstance(n, ast.ClassDef) and n.name == "Controller"][0]


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


def read_schema_max(key):
    """The schema's upper bound for `key` out of the SHIPPED config.yaml.

    ⚠️ ANCHORED AT EXACTLY TWO SPACES, WHICH IS THE WHOLE OF CFG-04's FIX HERE.
    This used to match `line.strip()`, and CFG-04's `inverters:` put a NESTED
    `max_w: int(100,5000)` six spaces in and earlier in the file - so the
    stripped version answered with the per-inverter bound and stayed green,
    because the two bounds are the same number today. Part 12 below runs this
    against a config where they are NOT the same number.
    """
    body = io.open(HERE / "config.yaml", encoding="utf-8").read()
    for line in body[body.index("\nschema:"):].splitlines():
        if not line.startswith(f"  {key}:") or line[2:3].isspace():
            continue
        st = line.strip()
        if "(" in st and ")" in st:
            return float(st[st.index("(") + 1:st.rindex(")")].split(",")[1])
    return None


# --------------------------------------------------------------------------- #
# fixtures
# --------------------------------------------------------------------------- #
class MemStore:
    """A Store that keeps its dict across "restarts", like the real file does."""

    def __init__(self, data=None):
        self.data = dict(DEFAULTS, **(data or {}))
        self.writes = 0

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

    def get_time(self, *a):
        return None

    def set_time(self, *a):
        pass


class Pub:
    enabled = False

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

    def close(self):
        pass


class Hass:
    """A Home Assistant that answers per ENTITY, which is the whole point here.

    ⚠️ EVERY FIXTURE IN THIS FILE ANSWERS PER ENTITY RATHER THAN PER CALL. A
    stub that returned one number whatever it was asked cannot tell "inverter B
    was read from inverter B's sensor" from "inverter B was handed inverter A's
    reading", which is the defect this card is about.
    """

    def __init__(self, numbers=None, limits=None, reject=(), hang=(),
                 hang_s=30.0):
        self.numbers = dict(numbers or {})
        self.limits_by = dict(limits or {})
        self.reject = set(reject)
        self.hang = set(hang)
        self.hang_s = hang_s
        self.writes = []            # (entity, value)
        self.write_at = []          # (entity, value, monotonic)
        self.calls = []

    async def number(self, entity, invert=False):
        v = self.numbers.get(entity)
        return None if v is None else (-v if invert else v)

    async def limits(self, entity):
        return self.limits_by.get(entity, (None, None, None))

    async def set_number(self, entity, value):
        if entity in self.hang:
            # A transport that accepted the request and never answers. This is
            # the shape AC 6 is about: not an error, not a refusal - silence.
            await asyncio.sleep(self.hang_s)
        self.writes.append((entity, value))
        self.write_at.append((entity, value, time.monotonic()))
        return entity not in self.reject

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


A_ENT = {"setpoint_entity": "number.sp_a", "soc_entity": "sensor.soc_a",
         "batt_entity": "sensor.batt_a"}
# ⚠️ SPIKE-04a (2026-09-04) ESTABLISHED B'S REAL RATING FROM ITS DATASHEET:
# a GoodWe GW5048D-ES, serial `55048ESU22AW0992`, 4600 W on the battery path
# (goodwe/NOTES.md §1.1b). THIS FIXTURE DELIBERATELY DOES NOT USE IT. 3000 W is
# kept because the only property these checks consume is that B's number is
# UNMISTAKABLY not A's, and the real pair is 5000/4600 - a 400 W gap, narrower
# than `slew_w` and inside the range where an off-by-one-unit bug and ordinary
# control arithmetic produce indistinguishable numbers. Collapsing the fixture
# onto the real values would cost these checks their separating power to buy
# realism they do not use. So: ILLUSTRATIVE, NOT B'S RATING, and not a
# placeholder for it either - the placeholder era ended on 2026-09-04.
B_RATING_HYPOTHESIS = 3000.0
# SAFETY-32. Each pack's OWN BMS cut-out percent, now a REQUIRED roster key.
# ⚠️ 10 IS MEASURED (register 0x0560 on the reference GW5000S-BP and the shipped
# `bms_soc_floor` default); 15 IS SIMULATED - there is no second pack. They
# DIFFER on purpose: a fixture whose two per-unit numbers are equal cannot show
# a check that read the wrong unit's, which is MUTATION.md habit 1 and the defect
# that has now bitten four consecutive cards. Both sit below the shipped
# `peak_soc_floor` of 20, so rule 8 accepts this roster.
A_BMS_FLOOR = 10
B_BMS_FLOOR_SIM = 15

A_RATING = 5000.0

# ⚠️ `slew_w` IS 0 HERE, AND IT USED TO BE 1000 - SAFETY-39, AND THE WAY THIS
# FIXTURE FOUND OUT IS THE ARGUMENT FOR THE RULE. This dict pairs `slew_w: 1000`
# with `max_w: 5000` because both were typed when the flat 5000 W install was
# new, and that pair is EXACTLY the configuration measured on the owner's system
# on 2026-09-05 driving the meter from 3.4 kW of export to 2.8 kW of import. The
# fixture was not wrong when it was written and nothing in this file noticed it
# had become so; validate.py rule 10 refused it the day it existed, which is
# what a rule about a relationship is for. 0 means "follow the rail"
# (`control.SLEW_RAIL_FRAC * max_w`) and is what config.yaml ships.
# ⚠️ SAFETY-10: `shadow_mode="off"` EXPLICITLY, BECAUSE NOTHING HERE PRESSES
# START. The shipped default is `auto`, which is SHADOW MODE while control is
# stopped - and in shadow mode every write is substituted with 0 W at the one
# choke point these parts grade. Without this line the per-unit clamp, the
# per-unit write budget and the per-unit heartbeat would all be measured on a
# controller that is correctly refusing to command anything, which is a
# different (and vacuous) claim. The shipped write path is unaffected: a
# stopped controller only ever writes 0 W in production anyway.
BASE_OPTS = dict(A_ENT, meter_source="off", meter_entity="sensor.grid",
                 max_w=A_RATING, gain=0.6, slew_w=0, deadband_w=15,
                 target_grid_w=0, step_w=10, heartbeat_s=10, stale_input_s=15,
                 shadow_mode="off")


def built(opts=None, store=None, hass=None, second=None):
    """A real Controller, optionally with a real second `Inverter`."""
    hass = hass or Hass(
        {"sensor.grid": 2000.0, "sensor.soc_a": 50.0, "sensor.batt_a": 0.0,
         "sensor.soc_b": 50.0, "sensor.batt_b": 0.0},
        {"number.sp_a": (-A_RATING, A_RATING, 1.0),
         "number.sp_b": (-B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS, 1.0)})
    c = Controller(dict(BASE_OPTS, **(opts or {})), hass,
                   store if store is not None else MemStore(), Pub(),
                   stop=asyncio.Event())
    c.inverters[0].dev_min, c.inverters[0].dev_max = -A_RATING, A_RATING
    if second is not None:
        c.inverters.append(second)
    return c, hass


def inverter_b(store=None, max_w=B_RATING_HYPOTHESIS, dev=None,
               setpoint="number.sp_b", sign_cfg=None):
    """A second inverter, built the way `build_roster()` builds the first.

    ⚠️ `sign_cfg` IS PASSED THROUGH RATHER THAN DEFAULTED, and that is not
    housekeeping. `SignWatchdogConfig` carries `sign_disagree_w` and
    `sign_disagree_s`, which are OPTIONS: a roster entry constructed without
    them gets the module defaults instead of the install's, so a site would run
    two watchdogs on two different thresholds and only one of them would be the
    one the operator configured. `build_roster()` hands every unit the same
    config for exactly that reason, and this helper copies the shape so the
    checks below grade what ships.
    """
    inv = Inverter("inverter-b", setpoint_entity=setpoint,
                   soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
                   max_w=max_w, store=store, sign_cfg=sign_cfg)
    inv.dev_min, inv.dev_max = dev or (-max_w, max_w)
    inv.soc, inv.batt = 50.0, 0.0
    return inv


def arm(c, shares):
    """Give `c` a real allocation record authorising exactly `shares`.

    ⚠️ SAFETY-28 MADE THIS NECESSARY AND THE FACT THAT IT IS NECESSARY IS THAT
    TICKET. `write_setpoint()` refuses any number the single allocator did not
    place on this tick, and the probes below call it directly - the way the
    heartbeat and the operator's buttons do - so they have to say where their
    number came from. Before that gate they did not, and that was the hole.

    ⚠️ AND IT IS BUILT THROUGH THE SHIPPED `resolve()`, NOT BY HAND, for the
    reason test_clamp.py's twin says: a helper that assembled an `Allocation`
    itself would be a second allocator living in the test file. The per-unit
    numbers are forced the way the add-on forces them - each unit files its own
    `supervised` envelope and the site target is their sum, so the water-fill
    has exactly one answer - and the assert is what says this fixture got it.
    """
    c.alloc_tick += 1
    claims = [Claim.set("loop", P_LOOP, float(sum(shares.values())),
                        "test stimulus")]
    for unit, watts in shares.items():
        claims.append(Claim.limit("supervised", P_SAFETY, -abs(float(watts)),
                                  abs(float(watts)), "max_w", actuator=unit))
    res = resolve(claims, units=c.units, live=c.units, tick=c.alloc_tick)
    c.allocation = res.allocation
    assert res.allocation.shares == {u: float(w) for u, w in shares.items()}, (
        f"arm() asked for {shares} and the shipped allocator placed "
        f"{res.allocation.shares} - fix the fixture, not the assert")
    return res.allocation


def run(coro):
    return asyncio.run(coro)


print("DEV-02 - per-instance Inverter objects")
print()
# ==============================================================================
print("1. AC 1: no per-device value is a Controller scalar, GRADED not read")

# ⚠️ THE NAMES ARE THE CARD'S OWN LIST, and they are here rather than derived
# because the claim is about THESE values. A derived list - "everything on the
# Inverter" - would go green the moment somebody deleted a field from both
# sides, which is the one refactor this check has to survive.
PER_DEVICE = ("soc", "batt", "dev_min", "dev_max", "last_write", "last_write_at",
              "last_write_ok", "clamp_last", "sign_wd", "last_beat",
              "setpoint_entity", "soc_entity", "batt_entity", "max_w")
# ⚠️ `target` IS NOT IN THAT LIST AND MUST NOT BE, WHICH IS THE MOST IMPORTANT
# LINE IN THIS PART. `Controller.target` is the SITE grid target - one
# integrator on P1, above the split - and `Inverter.target` is that unit's SHARE
# of it. They are two different quantities with the same name, and folding them
# together is the thing ALLOC-01's stability argument does not survive: the gain
# at the meter is a convex combination of the per-unit gains only while ONE
# integrator produces the number that is then divided. So the site keeps its
# scalar, on purpose, and part 5 is where that is graded.

_init = [n for n in CTRL.body
         if isinstance(n, ast.FunctionDef) and n.name == "__init__"][0]
_assigned = set()
for node in ast.walk(_init):
    for tgt in (node.targets if isinstance(node, ast.Assign) else
                [node.target] if isinstance(node, ast.AnnAssign) else []):
        for t in ([tgt] if not isinstance(tgt, ast.Tuple) else tgt.elts):
            if (isinstance(t, ast.Attribute) and isinstance(t.value, ast.Name)
                    and t.value.id == "self"):
                _assigned.add(t.attr)
_leaked = sorted(set(PER_DEVICE) & _assigned)
print(f"    Controller.__init__ assigns {len(_assigned)} attributes; "
      f"per-device ones among them: {_leaked or 'none'}")
check("⚠️ AC 1, STATICALLY. `Controller.__init__` assigns NOT ONE of the "
      "per-device names. This is the scan a re-introduction goes red on at the "
      "line it is written - `self.soc = None` back in the constructor is the "
      "whole defect, and it would pass every behavioural check in this "
      "repository at a roster of one",
      _leaked == [])

# The dynamic half. A field can also be grown at runtime, which the scan above
# cannot see, so the live object is compared against the roster entry.
_c1, _ = built()
_inst = set(vars(_c1))
print(f"    a live Controller carries {len(_inst)} instance attributes; "
      f"per-device ones among them: {sorted(set(PER_DEVICE) & _inst) or 'none'}")
check("...and DYNAMICALLY, on a constructed Controller: none of them is an "
      "instance attribute either. The compatibility views below are PROPERTIES "
      "on the class, which is why they do not appear here - and why an "
      "assignment through one cannot create a second copy to drift",
      not (set(PER_DEVICE) & _inst))
check("...and every one of them IS on the roster entry, which is the other "
      "half of the same claim: they were MOVED, not deleted",
      all(hasattr(_c1.inverters[0], n) for n in PER_DEVICE))
check("...and the roster is a list of Inverter objects rather than a list of "
      "names, so `units` cannot name an actuator the site has no object for",
      all(isinstance(i, Inverter) for i in _c1.inverters)
      and _c1.units == tuple(i.name for i in _c1.inverters))
_roster = build_roster(dict(BASE_OPTS), MemStore())
check("⚠️ AND `build_roster()` STILL RETURNS EXACTLY ONE INVERTER, from the "
      "FLAT options, under the shipped name. INVERTER B DOES NOT EXIST YET "
      "(~2 weeks; SPIKE-04 measures it, CFG-04 shapes the option that carries "
      "it), and this card must not invent it: what it delivers is that the CODE "
      "no longer assumes there is one. Adding the second entry is a change to "
      "that one function and to nothing else, which is the point - and this is "
      "the line that goes red if somebody adds a placeholder inverter B here",
      len(_roster) == 1 and _roster[0].name == UNIT_A
      and _roster[0].setpoint_entity == BASE_OPTS["setpoint_entity"]
      and _roster[0].max_w == BASE_OPTS["max_w"])

# ⚠️ AND MAIN.PY MUST NOT READ ITS OWN COMPATIBILITY VIEWS. A view main.py used
# would be the scalar back with a longer name: it answers for the FIRST
# inverter, so every consumer through it silently grades inverter B against
# inverter A - the defect, restored, behind a property.
VIEWS = ("soc", "batt", "dev_min", "dev_max", "last_write", "last_write_at",
         "last_write_ok", "clamp_last", "sign_wd", "unit")
_view_reads = []
for m in CTRL.body:
    if not isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)):
        continue
    if m.name in VIEWS or m.name == "_only":
        continue
    for n in ast.walk(m):
        if (isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name)
                and n.value.id == "self" and n.attr in VIEWS):
            _view_reads.append(f"{m.name}:{n.lineno} self.{n.attr}")
print(f"    main.py's own reads of the N=1 views: {_view_reads or 'none'}")
check("⚠️ AND MAIN.PY NEVER READS ONE OF ITS OWN N=1 VIEWS. Every path in the "
      "file iterates the roster. A `self.dev_min` anywhere inside Controller "
      "would answer for the first inverter at every roster size, which is "
      "exactly the scalar this card removed - and at N>=2 it would not even "
      "fail, because `_only()` is only reached when the roster is one",
      _view_reads == [])
check("...and the views are PROPERTIES, not attributes, so `c.soc = 4` writes "
      "inverter A's own reading rather than creating a second copy of it",
      all(isinstance(getattr(Controller, n, None), property) for n in VIEWS))


# ==============================================================================
print()
print("2. AC 2: the store migration, and it is GRADED rather than performed")

async def _one_cycle_early(c):
    """One real cycle. `_one_cycle` below is the same three lines; this one is
    needed before it, and duplicating three lines beats moving a definition
    somebody is reading in place."""
    await c.read_inputs()
    await c.cycle()
    return c


# ⚠️ THE RISKY HALF, IN THE CARD'S OWN WORDS. Namespacing `sign_tripped` is
# right - flat, a trip on B latches A - but a latch that silently fails to move
# is a disarmed safety guard on an install whose state file still says
# `sign_tripped: true`. Every check below is about that sentence.
_pre = MemStore({"sign_tripped": True,
                 "sign_reason": "SIGN_DISAGREEMENT: commanded -1500 W against "
                                "a meter reading +2000 W for 181 s",
                 "phase": "drain", "last_completed": "2026-08-01T00:00:00+00:00"})
_c2, _ = built(store=_pre)
print(f"    a pre-DEV-02 store with a LATCHED failsafe: "
      f"{_c2.store_migration.explain()}")
check("the latch lands under the FIRST inverter's own key, which is the only "
      "inverter an install written before this card could have had",
      _pre.data[unit_key(UNIT_A, "sign_tripped")] is True)
check("...and so does the reason, so the operator is still told WHY. A latch "
      "that migrated without its reason is a 0 W hold nobody can act on",
      "SIGN_DISAGREEMENT" in _pre.data[unit_key(UNIT_A, "sign_reason")])
check("⚠️ AND THE WATCHDOG THE CONTROLLER ACTUALLY BUILT COMES UP TRIPPED. "
      "THIS is the check that matters: the keys being right is worth nothing "
      "if `Inverter.__init__` reads somewhere else, and this reads the object "
      "the cycle will consult",
      _c2.inverters[0].sign_wd.tripped
      and "SIGN_DISAGREEMENT" in _c2.inverters[0].sign_wd.reason)
check("...and THE HISTORY IS UNTOUCHED. The maintenance keys are NOT "
      "namespaced - this add-on holds one maintenance state machine, and "
      "re-homing a month-scale history under a unit name is a different card - "
      "so an install keeps its phase and its last completed cycle",
      _pre.data["phase"] == "drain"
      and _pre.data["last_completed"] == "2026-08-01T00:00:00+00:00")
check("...and the store is marked migrated, so the next start does no work",
      _pre.data[SCHEMA_KEY] == SCHEMA_VERSION)
check("...and the keys that moved are exactly the ones store.py names as "
      "belonging to one inverter - read from `LEGACY_UNIT_KEYS` rather than "
      "transcribed here, so a key added to the migration without a check, or "
      "checked here without being migrated, cannot pass both ends",
      set(_c2.store_migration.moved)
      == {unit_key(UNIT_A, k) for k in LEGACY_UNIT_KEYS}
      and "phase" not in LEGACY_UNIT_KEYS)

# IDEMPOTENCE, AND IT IS THE ONE THAT PROTECTS THE OPERATOR'S CLEAR.
#
# ⚠️ THE CLEAR IS PRESSED, NOT SIMULATED, AND THAT CHANGE IS THE TICKET. This
# check used to clear the NAMESPACED key by hand and leave the flat one saying
# `true`, then assert that a restart did not resurrect the latch. It graded a
# state the shipped clear path does not produce: `handle_action("auto_toggle")`
# clears BOTH spellings, precisely so that `flat True + namespaced False` can
# only ever mean a trip taken while the install was rolled back (see below).
# Driving the button is also the only way to find that out - a fixture that
# hand-writes the store is a check on the fixture.
run(_c2.handle_action("auto_toggle"))          # the operator presses START
_writes_before = _pre.writes
_c2b, _ = built(store=_pre)
check("⚠️ RUNNING AGAIN AFTER AN OPERATOR CLEARED THE LATCH DOES NOT RESURRECT "
      "IT. A guard that went on comparing against a flat key nothing cleared "
      "would re-arm a failsafe the operator has already dealt with, once per "
      "restart, forever - which is what the `rep.ran` gate used to buy, and "
      "what clearing both spellings buys instead",
      not _c2b.inverters[0].sign_wd.tripped
      and _pre.data[unit_key(UNIT_A, "sign_tripped")] is False)
check("...and it wrote nothing at all the second time, which is what "
      "`schema_version` buys",
      _pre.writes == _writes_before)
check("⚠️ AND THE OPERATOR'S PRESS CLEARED THE PRE-DEV-02 FLAT KEYS TOO. That "
      "is what makes the two spellings comparable at all: cleared together, "
      "the only disagreement they can reach is `flat says tripped, nothing "
      "namespaced does` - which is a latch written by a version that had no "
      "namespaces, i.e. a trip taken while this install was ROLLED BACK",
      _pre.data["sign_tripped"] is False and _pre.data["sign_reason"] == "")

# ⚠️ THE ROLLBACK ROUND TRIP, RUN END TO END. THIS IS THE STATE THE `rep.ran`
# GATE SWALLOWED, and it is not a thought experiment - a firmware version change
# with a rollback path was performed on this site on 2026-09-02. store.py's
# rule 2 keeps the flat keys "so a rollback still finds its latch"; the operator
# path that rule contemplates is a ROUND TRIP, and the guard used to return
# early on every start that did not migrate, which is every start after the
# first.
#
#   1. install migrates      -> schema_version=1, unit/inverter-a/... = False
#   2. operator DOWNGRADES, and SAFETY-02 trips inside the downgrade window.
#      The old add-on knows only the flat key, so that is all it writes.
#   3. operator upgrades again. `rep.ran` is False - the store is already at
#      schema 1 - so the migration and the guard were both skipped and
#      `Inverter.__init__` read the namespaced key, which is still False.
#
# Observed before this fix: the controller came up with `sign_wd.tripped =
# False` while /data/state.json said `sign_tripped: true`. That is store.py's
# own header sentence - the file still says tripped, the banner is green, and
# the controller resumes driving the wrong way - rebuilt by the version marker.
_trip = MemStore()
_c2r0, _ = built(store=_trip)                              # 1. install, migrate
check("PRECONDITION 1: the install migrated and came up armed and untripped",
      _trip.data[SCHEMA_KEY] == SCHEMA_VERSION
      and _trip.data[unit_key(UNIT_A, "sign_tripped")] is False
      and not _c2r0.inverters[0].sign_wd.tripped)
# 2. THE DOWNGRADE WINDOW. The old add-on is the one that shipped before this
# card: it writes the flat pair and knows nothing about `unit/...`. Written the
# way that version wrote them - two `Store.set()` calls - rather than by
# reaching into `data`, so this is the file an old add-on would leave behind.
_trip.set("sign_reason", "SIGN_DISAGREEMENT: commanded -1800 W against a "
                         "meter reading +2100 W for 181 s")
_trip.set("sign_tripped", True)
check("PRECONDITION 2: the rolled-back add-on latched under the flat key ONLY, "
      "and left the namespaced one exactly as the migration wrote it. This is "
      "the state that made `rep.ran` a trap - the store is already at schema "
      "1, so nothing on the next start has any reason to run",
      _trip.data["sign_tripped"] is True
      and _trip.data[unit_key(UNIT_A, "sign_tripped")] is False
      and _trip.data[SCHEMA_KEY] == SCHEMA_VERSION)
_c2r, _ = built(store=_trip)                               # 3. upgrade again
print(f"    the rollback round trip: {_c2r.store_migration.explain()}; "
      f"inverter-a came up tripped={_c2r.inverters[0].sign_wd.tripped}")
check("PRECONDITION 3: this start did NOT migrate anything, which is exactly "
      "why the guard used to be skipped on it",
      not _c2r.store_migration.ran)
check("⚠️ AND THE LATCH IS CAUGHT ANYWAY. The guard runs on EVERY start now, "
      "so a trip taken while the install was rolled back is re-armed instead "
      "of being silently discarded by the version marker. A latch that a "
      "version change disarms is worse than one that was never persisted, "
      "because the state file is evidence saying it is fine",
      _c2r.inverters[0].sign_wd.tripped
      and "SIGN_DISAGREEMENT" in _c2r.inverters[0].sign_wd.reason
      and "-1800 W" in _c2r.inverters[0].sign_wd.reason)
check("...and it is persisted under the NAMESPACED key as it re-arms, so the "
      "next start reads it from the live spelling rather than re-deriving it",
      _trip.data[unit_key(UNIT_A, "sign_tripped")] is True)
check("...and the operator is told in the action log, because a latch re-armed "
      "on their behalf that only appears in a banner is a 0 W hold with no "
      "explanation",
      any("re-armed" in e for e in _c2r.events))
run(_one_cycle_early(_c2r))
check("...and it commands 0 W on its first cycle, which is what makes the "
      "re-arm a latch rather than a log line",
      _c2r.target == 0.0 and "failsafe" in _c2r.reason)
run(_c2r.handle_action("auto_toggle"))
_c2r2, _ = built(store=_trip)
check("⚠️ AND THE OPERATOR CAN STILL GET OUT OF IT - ONCE. The press clears "
      "both spellings, so the next start does NOT re-arm. This is the "
      "'re-arm forever' objection the `rep.ran` gate existed to answer, "
      "answered where it belongs: by clearing the fossil rather than by "
      "refusing to look at it",
      not _c2r2.inverters[0].sign_wd.tripped
      and _trip.data["sign_tripped"] is False
      and _trip.data[unit_key(UNIT_A, "sign_tripped")] is False)

# A FRESH INSTALL MUST NOT INVENT A LATCH.
_fresh = MemStore()
_c2c, _ = built(store=_fresh)
check("a fresh install migrates the DEFAULTS and comes up ARMED AND UNTRIPPED. "
      "`absent` and `never tripped` are the same statement, which is what "
      "store.py's own header is about",
      not _c2c.inverters[0].sign_wd.tripped
      and _fresh.data[unit_key(UNIT_A, "sign_tripped")] is False)

# ⚠️ AND NOW BREAK IT. THIS IS THE CHECK THE CARD ASKED FOR BY NAME.
class LosesTheLatch(MemStore):
    """A store that accepts the namespaced write and drops it on the floor.

    Not a hypothetical: it is what every partial write, full disk and
    silently-swallowed OSError looks like from here - `Store.save()` catches
    OSError and carries on by design, because losing the maintenance history is
    recoverable and refusing to run is not.
    """

    def set(self, k, v):
        if k.startswith("unit/"):
            return
        super().set(k, v)


_broken = LosesTheLatch({"sign_tripped": True,
                         "sign_reason": "SIGN_DISAGREEMENT: measured"})
_rep = migrate_unit_keys(_broken, UNIT_A)
print(f"    a store that silently drops the migrated keys: {_rep.explain()}")
check("⚠️ THE MIGRATION REPORTS THAT IT FAILED, because it READS BACK what it "
      "wrote instead of assuming the write landed. That read-back is the whole "
      "difference between this and a two-line copy",
      _rep.ran and not _rep.verified and not _rep.ok and _rep.lost)
check("...and it does NOT mark the store migrated, so a later version runs it "
      "again rather than skipping a migration that never happened",
      SCHEMA_KEY not in _broken.data)
check("...and a store that only fails to persist a DEFAULT is not reported as "
      "lost state. `sign_tripped=False` is the same statement as `absent`, and "
      "an alarm that fires on the harmless case is one an operator learns to "
      "dismiss before the real one arrives",
      migrate_unit_keys(LosesTheLatch(), UNIT_A).ok is True)

# ⚠️ AND THE MARKER GATE, ON A STORE WHERE SOMETHING WAS ALREADY NAMESPACED.
# FOUND BY MAINT-05's ROUND-3 BATTERY, AND IT WAS A SURVIVOR OF ALL 19 FILES.
# `migrate_unit_keys()` gates `SCHEMA_KEY` on `rep.ok`, and every check above
# reaches that gate on a store where NOTHING was already under `unit/...` - so
# `rep.kept` is empty in all of them, and the mutation
#
#     if rep.ok:            ->    if rep.ok or rep.kept:
#
# stamped the marker on a migration that DROPPED THE LATCH, with the whole suite
# green. That is a disarmed safety guard plus a store that says the migration
# finished, so the next start skips the one thing that would have re-armed it -
# DEV-02's own worst case with the paper trail filed under "done".
#
# The fixture is the one shape that separates them: `sign_reason` ALREADY
# namespaced (so `kept` is non-empty and the mutant's `or` fires) and the
# `sign_tripped: true` write dropped on the floor (so `lost` is non-empty and
# `ok` is False). `migrate_unit_keys()` is passed no roster, so `misdirected`
# cannot be what refuses this - it is the `lost` guard or nothing.
_kept_reason = "SIGN_DISAGREEMENT: already namespaced by a later version"
_broken_kept = LosesTheLatch({
    "sign_tripped": True,
    "sign_reason": "SIGN_DISAGREEMENT: measured",
    unit_key(UNIT_A, "sign_reason"): _kept_reason})
_rep_kept = migrate_unit_keys(_broken_kept, UNIT_A)
print(f"    a dropped latch write beside an already-namespaced reason: "
      f"{_rep_kept.explain()}")
print(f"    lost={len(_rep_kept.lost)} kept={len(_rep_kept.kept)} "
      f"ok={_rep_kept.ok}; marker "
      f"{_broken_kept.data.get(SCHEMA_KEY, '<<absent>>')!r}")
check("PRECONDITION: this fixture reaches the marker gate through the `lost` "
      "guard with a NON-EMPTY `kept`, which is the combination none of the "
      "checks above has. One key dropped, one key left alone, `ok` False. If a "
      "later edit empties `kept` here the mutation goes invisible again, so the "
      "shape is asserted rather than assumed",
      _rep_kept.ran and len(_rep_kept.lost) == 1 and len(_rep_kept.kept) == 1
      and not _rep_kept.ok
      and _broken_kept.data[unit_key(UNIT_A, "sign_reason")] == _kept_reason)
check("⚠️ AND THE STORE IS NOT MARKED MIGRATED. The key that was ALREADY there "
      "does not buy a marker for the latch that was not: stamping here would "
      "tell the next start that the migration which dropped `sign_tripped` had "
      "finished, so it would never run again and the guard would stay off on an "
      "install whose file still reads `sign_tripped: true`",
      SCHEMA_KEY not in _broken_kept.data)
check("...and the counterfactual, one property different: the SAME store shape "
      "on a store that does not drop writes IS marked migrated. So the refusal "
      "above is about the latch that was lost, not about `kept` being non-empty",
      migrate_unit_keys(
          MemStore({"sign_tripped": True,
                    "sign_reason": "SIGN_DISAGREEMENT: measured",
                    unit_key(UNIT_A, "sign_reason"): _kept_reason}),
          UNIT_A).ok is True)

_broken2 = LosesTheLatch({"sign_tripped": True,
                          "sign_reason": "SIGN_DISAGREEMENT: measured"})
_c2d, _ = built(store=_broken2)
check("⚠️ AND THE CONTROLLER REFUSES TO COME UP DISARMED ON IT. The store says "
      "this install is latched and the namespaced key cannot be made to hold "
      "it, so the latch is RE-ARMED from the legacy value rather than lost. "
      "The fail-safe direction for a latch is ON: a spurious trip costs a 0 W "
      "hold an operator clears with one button; a missing one costs whatever "
      "SAFETY-02 was watching for",
      _c2d.inverters[0].sign_wd.tripped
      and "SIGN_DISAGREEMENT" in _c2d.inverters[0].sign_wd.reason)
check("...and it says so in the action log, so the operator is not left to "
      "infer it from a banner",
      any("re-armed" in e for e in _c2d.events))


async def _one_cycle(c):
    await c.read_inputs()
    await c.cycle()
    return c


run(_one_cycle(_c2d))
check("...and it commands 0 W from its very first cycle, which is what a latch "
      "IS. A latch that reports itself and then commands normally is not a "
      "latch, and the re-arming above would be theatre",
      _c2d.target == 0.0 and "failsafe" in _c2d.reason)


# ==============================================================================
print()
print("3. AC 3: a trip on B does not latch A, and it survives a restart")

# ⚠️ THE STIMULUS IS ASYMMETRIC ON PURPOSE, AND THAT TOOK SOME ARRANGING. Both
# inverters get a share of ONE site target, so an ordinary wrong-way loop trips
# BOTH - correctly, because both really are pushing the wrong way. The state
# this part is about is the one the flat store keys made unreachable: ONE
# inverter standing on a command that opposes what it is graded against while
# the other does not. It is reached the way the field would reach it - inverter
# B's writes are being REJECTED, so `inv.last_write` stays where it was while
# A's tracks the loop - and `standing` is what SAFETY-02 grades, per unit.
#
# ⚠️ SAFETY-30 CHANGED WHAT "OPPOSES" MEANS HERE, AND THE FIXTURE HAD TO MOVE
# WITH IT - WHICH IS THIS CHECK DOING ITS JOB. It used to stand B on a -1500 W
# charge against a +2000 W METER and let the site figure convict it. At a roster
# of two that comparison is exactly the defect SAFETY-30 removes: the meter
# answers for BOTH inverters, so it is evidence about neither. B is now
# convicted by its OWN power sensor - standing on a -1500 W charge while
# `sensor.batt_b` measures it DISCHARGING at +1500 W, which is one inverter
# doing the opposite of what it was told and is a fact about that inverter
# alone. Inverter A's own reading stays at 0 W, so A is never gated in.
#
# ⚠️ AND THE REJECTION WINDOW IS WHY THIS RUNS IN THREE SECONDS AND NOT THIRTY.
# `unit_sign_disagree_s` is 1 s here; `LINK_STALE_S` is 30 s, so B is still LIVE
# for the whole run and the trip is graded before liveness has anything to say.
def _tripping_site():
    hass = Hass(
        {"sensor.grid": 2000.0, "sensor.soc_a": 50.0, "sensor.batt_a": 0.0,
         "sensor.soc_b": 50.0, "sensor.batt_b": 1500.0},
        {"number.sp_a": (-A_RATING, A_RATING, 1.0),
         "number.sp_b": (-B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS, 1.0)},
        reject={"number.sp_b"})
    store = MemStore()
    c, _ = built({"sign_disagree_w": 300, "sign_disagree_s": 1,
                  "unit_sign_disagree_w": 300, "unit_sign_disagree_s": 1,
                  "auto_start": True}, store=store, hass=hass)
    c.inverters.append(inverter_b(store, sign_cfg=c.inverters[0].sign_wd.cfg))
    c.auto = True
    # B is standing on a charge command while its own sensor says it is
    # discharging. Its writes are rejected, so this is what it goes on
    # standing at.
    c.inverters[1].last_write = -1500.0
    return c, hass, store


async def _trip_b():
    c, hass, store = _tripping_site()
    for _ in range(12):
        await c.read_inputs()
        await c.cycle()
        await asyncio.sleep(0.25)
    return c, hass, store


_c3, _h3, _s3 = run(_trip_b())
_a3, _b3 = _c3.inverters
print(f"    after 3 s: {_a3.name} tripped={_a3.sign_wd.tripped}, "
      f"{_b3.name} tripped={_b3.sign_wd.tripped}; "
      f"shares {_c3.allocation.shares}")
check("PRECONDITION: inverter B really did trip - the stimulus is not vacuous",
      _b3.sign_wd.tripped and "SIGN_DISAGREEMENT" in _b3.sign_wd.reason)
check("⚠️ AC 3. AND INVERTER A DID NOT. One SignWatchdog per actuator, which is "
      "what watchdog.py's own docstring has always said and what the Controller "
      "did not do: it held exactly one, so B's evidence latched A",
      not _a3.sign_wd.tripped and _a3.sign_wd.reason == "")
check("...and A IS STILL COMMANDING. B's failsafe is a P_SAFETY SET on B's own "
      "actuator, so stage 1 pins B at 0 W and the allocator hands the site "
      "target to A - which is the site running on the inverter that works "
      "rather than stopping because one of them is in trouble",
      _c3.allocation.shares.get("inverter-b") == 0
      and _c3.allocation.shares.get("inverter-a") == _c3.allocation.sum_w
      and _c3.allocation.sum_w != 0)
check("...and the number A was told reached A's own entity",
      ("number.sp_a", float(_a3.target)) in _h3.writes and _a3.target != 0.0)
check("⚠️ AND THE LATCHES ARE UNDER SEPARATE KEYS, which is the half that "
      "survives a restart. Flat, `sign_tripped` was ONE boolean: B's trip wrote "
      "it, and every Controller built afterwards read it as A's",
      _s3.data.get(unit_key("inverter-b", "sign_tripped")) is True
      and _s3.data.get(unit_key(UNIT_A, "sign_tripped")) is False)
check("⚠️ ...AND THE TRIP IS MIRRORED ONTO THE PRE-DEV-02 FLAT PAIR, WHICH IS "
      "THE ROLLBACK VIEW. store.py's rule 2 keeps the flat keys 'so a rollback "
      "still finds its latch', and that only ever covered the latch present AT "
      "MIGRATION TIME: nothing flat was written for a trip taken afterwards, so "
      "an operator who downgraded found `sign_tripped: false` and came up "
      "DRIVING, into the fault. The site-level spelling is right for it - the "
      "version being rolled back TO drives one actuator",
      _s3.data.get("sign_tripped") is True
      and "SIGN_DISAGREEMENT" in str(_s3.data.get("sign_reason", "")))

# THE RESTART, which is the whole reason the latch is persisted at all.
_h3b = Hass({"sensor.grid": 2000.0, "sensor.soc_a": 50.0,
             "sensor.batt_a": 0.0, "sensor.soc_b": 50.0,
             "sensor.batt_b": 1500.0},
            {"number.sp_a": (-A_RATING, A_RATING, 1.0),
             "number.sp_b": (-B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS, 1.0)})
_c3r, _ = built({"sign_disagree_w": 300, "sign_disagree_s": 1,
                 "unit_sign_disagree_w": 300, "unit_sign_disagree_s": 1,
                 "auto_start": True}, store=_s3, hass=_h3b)
_c3r.inverters.append(inverter_b(_s3,
                                 sign_cfg=_c3r.inverters[0].sign_wd.cfg))
_c3r.auto = True
run(_one_cycle(_c3r))
print(f"    after a restart against the same store: "
      f"A tripped={_c3r.inverters[0].sign_wd.tripped}, "
      f"B tripped={_c3r.inverters[1].sign_wd.tripped}, "
      f"shares {_c3r.allocation.shares}")
check("a brand new Controller against the same store comes back with B "
      "latched and A NOT. 'Restart the add-on' is the first thing an operator "
      "confronted with a FAILSAFE banner tries; before this card that restart "
      "handed them a site with both inverters parked",
      _c3r.inverters[1].sign_wd.tripped
      and not _c3r.inverters[0].sign_wd.tripped)
check("...and A goes on carrying the site target across the restart too",
      _c3r.allocation.shares.get("inverter-a") == _c3r.allocation.sum_w
      and _c3r.allocation.sum_w != 0)
check("...and the status page shows BOTH per-inverter rows - one latched, one "
      "armed. A single row would have to pick one inverter to describe, and "
      "'the sign watchdog is armed' printed over a latched sibling is the page "
      "lying",
      len([r for r in _c3r.checks()
           if r.get("id") == "direction" and r["ok"]
           and "inverter-a" in r["text"]]) == 1
      and len([r for r in _c3r.checks()
               if r.get("id") == "direction" and not r["ok"]
               and not r["warn"] and "inverter-b" in r["text"]]) == 1)
check("⚠️ AND SAFETY-30's SITE ROW IS THERE BESIDE THEM, EXACTLY ONCE. At two "
      "inverters there are three sign questions on this page and they are three "
      "rows: is A doing what it was told, is B, and is the SITE'S TOTAL "
      "pointing the right way. The last one is not derivable from the other "
      "two - both units can execute their shares perfectly while the number "
      "they were given is itself wrong - so a page that showed only the "
      "per-unit rows would be advertising two guards where the install has "
      "three",
      len([r for r in _c3r.checks()
           if r.get("id") == "site_direction"]) == 1)
check("...and the per-inverter rows now say WHAT the unit was compared "
      "against, because at a roster of two it is no longer the meter. A row "
      "still reading 'the command opposed the meter' would send an operator to "
      "look at P1 for a fault that is visible only on one inverter's own power "
      "sensor",
      len([r for r in _c3r.checks() if r.get("id") == "direction"]) == 2
      and all("own power" in r["text"] for r in _c3r.checks()
              if r.get("id") == "direction"))

# The operator's clear, which is one button and has to reach every latch.
run(_c3r.handle_action("auto_toggle"))          # -> stopped
check("pressing stop clears nothing, exactly as before this card",
      _c3r.inverters[1].sign_wd.tripped
      and _s3.data.get(unit_key("inverter-b", "sign_tripped")) is True)
run(_c3r.handle_action("auto_toggle"))          # -> started
check("pressing start clears EVERY latched inverter, under its own key. One "
      "button is what the UI has and what the banner tells the operator to "
      "press; a press that cleared A while B stayed latched, with nothing on "
      "the page changing, is worse than no button",
      not _c3r.inverters[1].sign_wd.tripped
      and _s3.data[unit_key("inverter-b", "sign_tripped")] is False
      and _s3.data[unit_key("inverter-b", "sign_reason")] == "")
check("...and the same press cleared the flat rollback view, so a downgrade "
      "after a clear does not find a trip the operator has already dealt with "
      "- and neither does the next start, which now reads that key on EVERY "
      "boot rather than only on the one that migrates",
      _s3.data["sign_tripped"] is False and _s3.data["sign_reason"] == "")
_c3r2, _ = built({"sign_disagree_w": 300, "sign_disagree_s": 1,
                  "unit_sign_disagree_w": 300, "unit_sign_disagree_s": 1},
                 store=_s3, hass=_h3b)
check("⚠️ AND THE RESTART AFTER THAT CLEAR COMES UP UNLATCHED. This is the "
      "'re-arm forever' hazard the every-start guard would have if the clear "
      "did not reach both spellings, driven through the real button and a real "
      "reconstruction rather than argued",
      not _c3r2.inverters[0].sign_wd.tripped
      and not any("re-armed" in e for e in _c3r2.events))


# ==============================================================================
print()
print("4. AC 4: each rating from its OWN entity, and what follows from it")

# ⚠️ THE TWO RATINGS DIFFER, WHICH IS THE ONLY REASON THIS PART CAN SEE
# ANYTHING. Give both inverters 5000 W and every check below passes against a
# controller that reads one entity twice.
async def _read_ratings():
    store = MemStore()
    c, hass = built(store=store, second=inverter_b(store))
    c.inverters[0].dev_min = c.inverters[0].dev_max = None
    c.inverters[1].dev_min = c.inverters[1].dev_max = None
    # main.py's own startup read, on the real method.
    c.stop_event = asyncio.Event()

    async def _limits(inv):
        if inv.setpoint_entity:
            inv.dev_min, inv.dev_max, _ = await hass.limits(inv.setpoint_entity)
    await asyncio.gather(*(_limits(i) for i in c.inverters))
    return c, hass


_c4, _h4 = run(_read_ratings())
_a4, _b4 = _c4.inverters
print(f"    A reports {_a4.dev_min:.0f}...{_a4.dev_max:.0f} W from "
      f"{_a4.setpoint_entity}; B reports {_b4.dev_min:.0f}...{_b4.dev_max:.0f} W "
      f"from {_b4.setpoint_entity}")
check("⚠️ EACH INVERTER'S DEVICE RANGE COMES FROM ITS OWN SETPOINT ENTITY. "
      "There was ONE `hass.limits()` call on ONE entity, so inverter B's "
      "rating was never read at all - its clamp, its freeze rail and the peak "
      "guard's achievable_w were all inverter A's numbers",
      (_a4.dev_min, _a4.dev_max) == (-A_RATING, A_RATING)
      and (_b4.dev_min, _b4.dev_max)
      == (-B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS))
check("...and the per-unit rating is the conjunction of its supervised cap and "
      "its own device range, written once so three consumers cannot transcribe "
      "it three different ways",
      _a4.rating_w() == A_RATING
      and _b4.rating_w() == B_RATING_HYPOTHESIS)
check("...and a bound that could NOT be read is skipped rather than read as "
      "zero. (None, None) means ESPHome was not up; answering 0 W there would "
      "say this inverter can do nothing, which is the disarming direction",
      inverter_b(dev=(None, None)).rating_w() == B_RATING_HYPOTHESIS)

# THE CLAMP. Same requested value, two inverters, two answers.
async def _clamp_both():
    c, hass = built(store=MemStore(), second=inverter_b())
    # SAFETY-28's gate is upstream of the clamp, so the out-of-range request
    # has to be one the allocator actually placed or it is substituted with
    # 0 W and this part grades the gate instead of the clamp.
    arm(c, {"inverter-a": 9999.0, "inverter-b": 9999.0})
    await c.write_setpoint(9999.0, force=True, unit=c.inverters[0])
    await c.write_setpoint(9999.0, force=True, unit=c.inverters[1])
    return c, hass


_c4c, _h4c = run(_clamp_both())
print(f"    both asked for +9999 W: {_h4c.writes}")
check("⚠️ THE CLAMP FOR B USES B'S NUMBERS. The same out-of-range request "
      "reaches A's wire at A's rating and B's wire at B's - which it could not, "
      "while there was one `max_w` and one device pair on the Controller",
      _h4c.writes == [("number.sp_a", A_RATING),
                      ("number.sp_b", B_RATING_HYPOTHESIS)])
check("...and each unit's own clamp evidence records ITS OWN engagement, so a "
      "clamp on B is not reported as a clamp on A",
      _c4c.inverters[0].clamp_last.limit_w == A_RATING
      and _c4c.inverters[1].clamp_last.limit_w == B_RATING_HYPOTHESIS
      and all(i.clamp_last.engaged for i in _c4c.inverters))
check("...and the action log NAMES the inverter, because 'the computed "
      "setpoint was clamped' over a site with two of them is a row an operator "
      "cannot act on",
      any("inverter-b" in e and "OUTPUT_CLAMP" in e for e in _c4c.events)
      and any("inverter-a" in e and "OUTPUT_CLAMP" in e for e in _c4c.events))

# THE SITE AGGREGATES, and each one's reduction is the check.
_c4s, _ = built(store=MemStore(), second=inverter_b())
print(f"    site: max_w {_c4s.site_max_w:.0f} W, rating {_c4s.site_rating_w:.0f} W, "
      f"rail {rail_w(_c4s.site_max_w, dev_min=_c4s.site_dev_bound('dev_min'), dev_max=_c4s.site_dev_bound('dev_max')):.0f} W")
check("`achievable_w` and the freeze rail are the SUM of the per-unit "
      "conjunctions, not a conjunction of the sums. Summing first would let one "
      "unit's small device range cap the whole site; conjoining first is what "
      "makes inverter B's rating bind inverter B",
      _c4s.site_rating_w == A_RATING + B_RATING_HYPOTHESIS
      and _c4s.site_max_w == A_RATING + B_RATING_HYPOTHESIS)
check("...and a device bound that ANY unit has not answered makes the SITE "
      "bound unknown rather than smaller - which is the direction the shipped "
      "single-inverter path already took with (None, None)",
      _c4s.site_dev_bound("dev_max") == A_RATING + B_RATING_HYPOTHESIS
      and built(store=MemStore(),
                second=inverter_b(dev=(None, None)))[0]
      .site_dev_bound("dev_max") is None)
check("...and at a roster of ONE every one of them is the shipped number, "
      "exactly. That is AC 8 stated on the aggregates rather than on the suite",
      _c1.site_max_w == A_RATING and _c1.site_rating_w == A_RATING
      and _c1.site_dev_bound("dev_max") == A_RATING)

# The two SoC reductions, which are the ones a wrong answer costs a pack.
_c4s.inverters[0].soc, _c4s.inverters[1].soc = 80.0, 21.0
check("⚠️ THE PACK FLOOR IS THE EMPTIEST PACK, NOT THE MEAN. SAFETY-24's floor "
      "asks whether the site may spend the pack for money; an average of 80 % "
      "and 21 % is 50 %, which is permission to discharge a pack that is one "
      "point above its floor",
      _c4s.pack_floor_soc == 21.0)
_c4s.inverters[1].soc = None
check("⚠️ ...AND IT IS None IF ANY UNIT DID NOT ANSWER, not the min of the ones "
      "that did. `decide()` treats None as floor-limited, which costs a shave "
      "rather than a pack; a min() over the units that answered would be a "
      "measurement asserted about a pack nobody read - which is NOTES.md 4.4b's "
      "defect with a second inverter in it",
      _c4s.pack_floor_soc is None)
_c4s.inverters[1].soc = 21.0
check("...and back to the emptiest pack the moment it answers again",
      _c4s.pack_floor_soc == 21.0)


# ==============================================================================
print()
print("5. AC 5: `actual_w` is SITE level while the readings are per-instance")

# ⚠️ THE TWO HALVES OF §6.2's IDENTITY BREAK IN OPPOSITE DIRECTIONS, AND GETTING
# THAT BACKWARDS REINTRODUCES A MEASURED DIVERGENCE. ALLOC-01's review isolated
# it on admin/goodwe's two-master rig: with the loop fed ONE inverter's power
# the aggregate trace at the meter becomes a function of the split - up to
# 510 W apart at gain 0.2 and 250 W at gain 0.4 - while the allocator's shares
# summed to S in every one of those runs. The counterfactual is the proof, and
# it is `tools/alloc_sim.py --splits --feedback site`: every split bit-identical
# at every gain. test_alloc.py part 8 runs the same comparison against a plant
# it models itself; what is pinned HERE is main.py's own call site.
_compute_calls = [n for n in ast.walk(MAIN_AST)
                  if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
                  and n.func.id == "compute"]
check("main.py has exactly ONE `compute(` call site, so the argument graded "
      "below is unambiguously the shipped one", len(_compute_calls) == 1)
_actual_kw = first((k.value for k in nth(_compute_calls, 0).keywords if k.arg == "actual_w"),
                   "no actual_w= keyword")
_actual_src = ast.unparse(_actual_kw) if present(_actual_kw) else repr(_actual_kw)
_site_batt_assign = [ast.unparse(n.value) for n in ast.walk(CTRL)
                     if isinstance(n, ast.Assign) and len(n.targets) == 1
                     and isinstance(n.targets[0], ast.Name)
                     and n.targets[0].id == _actual_src]
print(f"    main.py's compute(actual_w=...) is `{_actual_src}`, assigned from "
      f"{_site_batt_assign}")
check("⚠️ AND IT IS THE SITE SUM. `compute()`'s anti-windup test is "
      "`abs(prev_w - actual_w) > saturation_w` where `prev_w` is the SITE "
      "target, so one unit's power there compares a site number against a unit "
      "number, trips the saturation freeze permanently and leaves a standing "
      "grid error. `self.batt` here - the shipped expression before this card - "
      "is the defect",
      _site_batt_assign == ["self.site_batt_w"])
_c5, _ = built(store=MemStore(), second=inverter_b())
_c5.grid = 1000.0
_c5.inverters[0].soc = 50.0
_c5.inverters[0].batt, _c5.inverters[1].batt = 1000.0, 2000.0
check("...and the property is a plain SUM of the per-unit readings: a healthy "
      "pair at 1000 and 2000 W is a site delivering 3000 W, which is what the "
      "meter sees and what the site target is compared against",
      _c5.site_batt_w == 3000.0)
check("...and None if any unit did not answer, which falls into the existing "
      "`inputs missing -> command 0 W` path rather than into a sum that is "
      "quietly short by one inverter",
      _c5.inputs_ok is not None
      and (setattr(_c5.inverters[1], "batt", None) or _c5.site_batt_w is None))
check("...and `inputs_ok` requires EVERY pack's own two readings, so a silent "
      "unit stops the site at P_SAFETY instead of being dropped from the split. "
      "A missing measurement and a dead link are different faults with "
      "different remedies", not _c5.inputs_ok)
_c5.inverters[1].batt = 2000.0
check("...and it is true again the moment the pack answers", _c5.inputs_ok)
check("...and each unit's own reading is still ITS OWN, read from ITS OWN "
      "entity - which is the half that had to become per-instance for the sum "
      "above to be a sum of anything",
      _c5.inverters[0].batt_entity == "sensor.batt_a"
      and _c5.inverters[1].batt_entity == "sensor.batt_b")


async def _reads_are_per_entity():
    store = MemStore()
    hass = Hass({"sensor.grid": 1000.0, "sensor.soc_a": 80.0,
                 "sensor.batt_a": 1000.0, "sensor.soc_b": 21.0,
                 "sensor.batt_b": 2000.0},
                {"number.sp_a": (-A_RATING, A_RATING, 1.0)})
    c, _ = built(store=store, hass=hass, second=inverter_b(store))
    await c.read_inputs()
    return c


_c5r = run(_reads_are_per_entity())
print(f"    after read_inputs(): A soc={_c5r.inverters[0].soc} "
      f"batt={_c5r.inverters[0].batt}; B soc={_c5r.inverters[1].soc} "
      f"batt={_c5r.inverters[1].batt}; site batt={_c5r.site_batt_w}")
check("⚠️ AND `read_inputs()` FILLS THEM FROM THE RIGHT SENSORS. This is the "
      "line a roster loop that read `o['soc_entity']` for every unit goes red "
      "on - it would give both packs A's state of charge, which is the pack "
      "floor grading B against A",
      (_c5r.inverters[0].soc, _c5r.inverters[0].batt) == (80.0, 1000.0)
      and (_c5r.inverters[1].soc, _c5r.inverters[1].batt) == (21.0, 2000.0)
      and _c5r.site_batt_w == 3000.0 and _c5r.pack_floor_soc == 21.0)


# ==============================================================================
print()
print("6. AC 6: independent heartbeats, with the hang INJECTED")

# ⚠️ WHAT THE DEFECT ACTUALLY WAS, said precisely, because the fix only makes
# sense against it. There was ONE `last_beat` and ONE serialised
# `await self.write_setpoint(...)`. Layer 1 - the ESP32's own watchdog -
# commands 0 W and KEEPS COMMANDING IT after ~30 s without a refresh, and that
# is the failsafe this whole add-on is built around. So a transport that
# accepted a request for inverter A and never answered would hold the single
# write path, and the refresh inverter B was waiting for would arrive late or
# not at all: ONE INVERTER'S TRANSPORT FAULT TAKING A HEALTHY SIBLING TO ZERO.
#
# Two things fix it and both are graded: the writes go out CONCURRENTLY, and
# each one is bounded by `WRITE_BUDGET_S`.
_HB_MAX = given(read_schema_max("heartbeat_s"), "heartbeat_s bound")
print(f"    WRITE_BUDGET_S {WRITE_BUDGET_S:.0f} s + the schema's largest "
      f"heartbeat_s {_HB_MAX:.0f} s = {WRITE_BUDGET_S + _HB_MAX:.0f} s, against "
      f"layer 1's {FIRMWARE_INGEST_VETO_S:.0f} s veto")
check("⚠️ THE BUDGET IS DERIVED, AND THE INEQUALITY IS WHAT IS GRADED. The "
      "worst a hung write can do to a sibling is delay its NEXT beat by the "
      "budget, so budget + the largest heartbeat the schema accepts has to fit "
      "inside layer 1's veto. Raising `heartbeat_s`'s ceiling takes THIS red "
      "instead of taking an inverter to zero in the field",
      WRITE_BUDGET_S + _HB_MAX < FIRMWARE_INGEST_VETO_S)
check("...and it is below hass.py's own 10 s HTTP timeout, so the budget is "
      "the binding one: we would rather abandon a write and re-issue it on the "
      "next tick than hold the control loop for ten seconds",
      WRITE_BUDGET_S < 10.0)

# THE EXPERIMENT. Inverter A's transport accepts and never answers; inverter B's
# is healthy. The budget is shortened for the run - the shipped value is graded
# above - because a 4 s wait per case is 8 s of suite time to measure something
# that is already a ratio.
_REAL_BUDGET = app_main.WRITE_BUDGET_S
app_main.WRITE_BUDGET_S = 0.30


async def _hung_heartbeat():
    """The SHIPPED heartbeat, against a transport that never answers for A.

    ⚠️ `send_heartbeats()` IS THE PRODUCTION METHOD `run_control()` CALLS, and
    driving it rather than a gather written here is the difference between
    grading the add-on and grading this file. Measured the hard way: while the
    dispatch was five lines inline in `run_control()`, this part could only
    gather its own `write_setpoint()` calls - and the mutation that replaced the
    shipped dispatch with a `for` loop PASSED.
    """
    store = MemStore()
    hass = Hass({}, {}, hang={"number.sp_a"}, hang_s=30.0)
    c, _ = built(store=store, hass=hass, second=inverter_b(store))
    c.inverters[0].target, c.inverters[1].target = 500.0, 700.0
    arm(c, {"inverter-a": 500.0, "inverter-b": 700.0})
    t0 = time.monotonic()
    await c.send_heartbeats(1000.0, 10.0)
    return c, hass, time.monotonic() - t0, t0


_c6, _h6, _elapsed, _t0 = run(_hung_heartbeat())
_b_at = [t - _t0 for e, _v, t in _h6.write_at if e == "number.sp_b"]
print(f"    A hangs forever, B is healthy: the pair took {_elapsed:.2f} s "
      f"(budget {app_main.WRITE_BUDGET_S:.2f} s); B's write reached the wire "
      f"{(_b_at[0] if _b_at else float('nan')):.3f} s in; wire saw "
      f"{_h6.writes}")
check("⚠️ AC 6. B'S WRITE REACHED THE WIRE WHILE A WAS STILL HANGING. That is "
      "the whole criterion: a hung write to A cannot consume B's layer-1 "
      "budget, because B's write was never behind A's in a queue",
      _h6.writes == [("number.sp_b", 700.0)]
      and bool(_b_at) and _b_at[0] < app_main.WRITE_BUDGET_S / 2.0)
check("...and the pair returned in about ONE budget rather than in two, which "
      "is what says the writes were concurrent and not merely bounded. Two "
      "bounded writes in series is still a sibling waiting for a corpse",
      _elapsed < app_main.WRITE_BUDGET_S * 1.9)
check("...and the hung inverter is recorded as having FAILED its write, so "
      "`last_write` is not advanced and the next tick re-issues it by itself",
      _c6.inverters[0].last_write is None
      and _c6.inverters[0].last_write_ok is False)
check("...while the healthy one recorded its write and will not re-send an "
      "unchanged value",
      _c6.inverters[1].last_write == 700.0
      and _c6.inverters[1].last_write_ok is True)
check("...and the operator is told, in the action log, that a write was "
      "abandoned - a write that vanishes silently is a command an operator "
      "believes is standing",
      any("timed out" in e and "inverter-a" in e for e in _c6.events))

# THE COUNTERFACTUAL: the serialised write this replaces, run on the same
# transport. Without it "concurrent" is a word rather than a measurement.
async def _serialised():
    store = MemStore()
    hass = Hass({}, {}, hang={"number.sp_a"}, hang_s=30.0)
    c, _ = built(store=store, hass=hass, second=inverter_b(store))
    c.inverters[0].target, c.inverters[1].target = 500.0, 700.0
    arm(c, {"inverter-a": 500.0, "inverter-b": 700.0})
    t0 = time.monotonic()
    b_at = None
    for inv in c.inverters:                    # <- the shape this card replaces
        await c.write_setpoint(inv.target, force=True, unit=inv)
        if inv.name == "inverter-b":
            b_at = time.monotonic() - t0
    return b_at, time.monotonic() - t0


_ser_b_at, _ser_total = run(_serialised())
print(f"    the same transport, written SERIALLY: B's refresh arrived "
      f"{_ser_b_at:.2f} s in, against {abs(_b_at[0]) if _b_at else 0:.2f} s "
      f"concurrently")
check("⚠️ AND THE DEFECT IS REPRODUCED BESIDE THE FIX. Written in series, B's "
      "refresh waits out A's entire budget before it is even attempted. At the "
      "SHIPPED budget that is a whole extra second-scale delay per tick on "
      "every inverter behind the hung one, and with the unbounded write this "
      "card replaces it was the HTTP stack's timeout - which is what put a "
      "healthy sibling within reach of layer 1's veto",
      _ser_b_at >= app_main.WRITE_BUDGET_S)

# AND THE DEADLINE ITSELF IS PER UNIT - graded through the shipped method, for
# the reason the experiment above is: a `due` list computed in this file is a
# check on this file. A refresh sent to A must not mark B refreshed.
async def _split_deadlines():
    store = MemStore()
    hass = Hass()
    c, _ = built(store=store, hass=hass, second=inverter_b(store))
    c.inverters[0].target, c.inverters[1].target = 500.0, 700.0
    arm(c, {"inverter-a": 500.0, "inverter-b": 700.0})
    # A was refreshed a moment ago; B has not been refreshed for a long time.
    c.inverters[0].last_beat, c.inverters[1].last_beat = 1000.0, 0.0
    await c.send_heartbeats(1005.0, 10.0)
    return c, hass


_c6b, _h6b = run(_split_deadlines())
print(f"    A fed at t=1000, B never: at t=1005 with a 10 s heartbeat the wire "
      f"saw {_h6b.writes}")
check("⚠️ AND THE DEADLINE IS PER INVERTER TOO, not one shared clock. Only the "
      "overdue inverter was refreshed. With ONE `last_beat` a refresh sent to A "
      "marked B refreshed as well - so an inverter whose own writes kept "
      "failing was recorded as fed, forever, by its sibling's success, and the "
      "one thing that proves this add-on is alive was never sent to it",
      _h6b.writes == [("number.sp_b", 700.0)])
check("...and the refreshed unit's deadline moved while the other's did not, "
      "so the next tick asks the same question again about the same unit",
      _c6b.inverters[1].last_beat == 1005.0
      and _c6b.inverters[0].last_beat == 1000.0)
_beat_src = inspect.getsource(Controller.send_heartbeats)
check("...and the heartbeat re-emits each unit's OWN share rather than the "
      "site target. A heartbeat that re-sent `self.target` would be the "
      "broadcast fault arriving through the back door, once every "
      "`heartbeat_s`, on a path with no allocator in it",
      "inv.target, force=True, unit=inv" in _beat_src
      and "self.target, force=True" not in _beat_src)


# ⚠️ AND THE CYCLE'S OWN WRITE DISPATCH IS THE OTHER HALF OF THE SAME PATH. The
# heartbeat is not the only place a hung inverter can hold the loop: every cycle
# writes every unit's share, and in series that is the same defect at the cycle
# rate rather than at the heartbeat rate.
app_main.WRITE_BUDGET_S = 0.30


async def _hung_cycle():
    store = MemStore()
    hass = Hass({"sensor.grid": 1000.0, "sensor.soc_a": 50.0,
                 "sensor.batt_a": 0.0, "sensor.soc_b": 50.0,
                 "sensor.batt_b": 0.0},
                {"number.sp_a": (-A_RATING, A_RATING, 1.0)},
                hang={"number.sp_a"}, hang_s=30.0)
    c, _ = built({"auto_start": True}, store=store, hass=hass,
                 second=inverter_b(store))
    c.auto = True
    await c.read_inputs()
    t0 = time.monotonic()
    await c.cycle()
    return c, hass, time.monotonic() - t0, t0


_c6c, _h6c, _cyc_elapsed, _cyc_t0 = run(_hung_cycle())
_cyc_b = [t - _cyc_t0 for e, _v, t in _h6c.write_at if e == "number.sp_b"]
print(f"    a REAL cycle with A's transport hung: took {_cyc_elapsed:.2f} s, "
      f"B's share reached the wire {(_cyc_b[0] if _cyc_b else float('nan')):.3f} s "
      f"in; wire saw {_h6c.writes}")
check("⚠️ AND A REAL CYCLE DELIVERS B'S SHARE WHILE A HANGS. This is the same "
      "criterion one layer up from the heartbeat: the per-unit writes in "
      "`cycle()` go out together, so a hung inverter costs the loop one budget "
      "and costs its siblings nothing",
      _h6c.writes and _h6c.writes[0][0] == "number.sp_b"
      and bool(_cyc_b) and _cyc_b[0] < app_main.WRITE_BUDGET_S / 2.0
      and _cyc_elapsed < app_main.WRITE_BUDGET_S * 1.9)
check("...and the share B was told is the allocator's, not the site target - "
      "the hung sibling does not turn the cycle into a broadcast",
      _h6c.writes[0][1] == float(_c6c.allocation.shares["inverter-b"])
      and _h6c.writes[0][1] != _c6c.target)

app_main.WRITE_BUDGET_S = _REAL_BUDGET
check("...and the shipped budget is restored after the experiments, so nothing "
      "below runs against a 0.3 s one",
      app_main.WRITE_BUDGET_S == WRITE_BUDGET_S)


# ⚠️ AND THE OPERATOR'S STOP HAS TO MOVE THE SAME TWO THINGS. This is a hazard
# THIS CARD CREATED and it is worth stating plainly: before DEV-02 the heartbeat
# re-emitted `self.target`, so the stop button zeroing that one scalar was
# enough. It now re-emits `inv.target` - it must, or it would broadcast the site
# number - so a stop that zeroed only the site target would leave each unit's
# last share standing for the very next refresh to put back on the wire, seconds
# after the operator watched the inverter go to zero.
async def _stop_then_beat():
    store = MemStore()
    hass = Hass()
    c, _ = built({"auto_start": True}, store=store, hass=hass,
                 second=inverter_b(store))
    c.auto = True
    c.target = 1990.0
    c.inverters[0].target, c.inverters[1].target = 995.0, 995.0
    await c.handle_action("auto_toggle")            # -> stopped
    stopped_at = list(hass.writes)
    # A heartbeat racing the press, before any cycle has run.
    await c.send_heartbeats(1000.0, 10.0)
    return c, hass, stopped_at


_c6s, _h6s, _stop_writes = run(_stop_then_beat())
print(f"    stop pressed with shares standing at 995/995: wrote {_stop_writes}, "
      f"then the next heartbeat wrote {_h6s.writes[len(_stop_writes):]}")
check("the stop button commands 0 W to EVERY inverter",
      sorted(_stop_writes) == [("number.sp_a", 0.0), ("number.sp_b", 0.0)])
check("⚠️ AND THE NEXT HEARTBEAT DOES NOT PUT THE OLD SHARES BACK. Each unit's "
      "own target was zeroed with the write, so the refresh that proves this "
      "add-on is alive carries the zero rather than resurrecting the command "
      "the operator just stopped",
      all(v == 0.0 for _e, v in _h6s.writes)
      and all(i.target == 0.0 for i in _c6s.inverters))


# ==============================================================================
print()
print("7. AC 7: live_units() answers per-instance liveness honestly")

# ⚠️ THE ALLOCATOR DROPS WHAT THIS DOES NOT LIST and the survivors carry the
# whole site target, so both directions of a wrong answer cost something real:
# a false "not live" divides the site target among fewer inverters than exist,
# and a false "live" allocates into a link nobody can reach and calls the site
# target met.
_now = datetime.now(timezone.utc)
_c7, _ = built(store=MemStore(), second=inverter_b())
check("a healthy roster is entirely live, which is the ordinary answer and has "
      "to stay cheap", _c7.live_units() == ("inverter-a", "inverter-b"))
_c7.inverters[1].note_write(False, _now - timedelta(seconds=5))
check("...and ONE rejected write does not drop anything. `set_number()` returns "
      "False for a transient 400 as readily as for a device that has gone, and "
      "dropping a unit on the first one would hand its share to a sibling every "
      "time Home Assistant hiccupped",
      _c7.live_units() == ("inverter-a", "inverter-b"))
# ⚠️ THE SECOND REJECTION DOES NOT MOVE THE START OF THE RUN, and that is the
# shipped behaviour rather than a fixture convenience: `rejected_since` records
# when the CURRENT run of failures began and only an ACCEPTANCE clears it. So
# the run is restarted here from a timestamp a whole window ago, through the
# same `note_write()` the write path calls, rather than by writing the field.
_c7.inverters[1].note_write(True, _now)
_c7.inverters[1].note_write(False, _now - timedelta(seconds=LINK_STALE_S + 1))
check("⚠️ ...BUT A CONTINUOUS RUN OF THEM FOR A WHOLE LAYER-1 WINDOW DOES. "
      "30 s is not a new opinion about links: it is `FIRMWARE_INGEST_VETO_S`, "
      "the point at which the ESP32 takes that inverter to 0 W on its own. "
      "Claiming it as commandable past there is the add-on promising the "
      "allocator watts the firmware has already decided it will not deliver",
      _c7.live_units() == ("inverter-a",)
      and LINK_STALE_S == FIRMWARE_INGEST_VETO_S)
_c7.inverters[1].note_write(True, _now)
check("...and ONE acceptance brings it straight back, because the window is "
      "about a run of failures and not about a counter that has to be reset by "
      "hand", _c7.live_units() == ("inverter-a", "inverter-b"))
check("...and it is answered PER TICK from a clock, never cached: the same "
      "object gives two answers as the rejection ages past the window, which "
      "is what says nothing was decided at construction",
      _c7.inverters[1].is_live(_now)[0] is True
      and (_c7.inverters[1].note_write(False, _now)
           or _c7.inverters[1].is_live(_now + timedelta(seconds=LINK_STALE_S))[0])
      is False)
check("...and the reason is carried, not just the verdict, so the status page "
      "and the log can say WHICH link and for how long",
      "rejected" in _c7.inverters[1].is_live(
          _now + timedelta(seconds=LINK_STALE_S))[1])
_c7.inverters[1].note_write(True, _now)
check("⚠️ AND A UNIT WITH NO `setpoint_entity` IS DELIBERATELY STILL LIVE HERE. "
      "It is a CONFIGURATION that cannot be delivered rather than a link that "
      "went away, and the two want opposite answers: a dropped link is an event "
      "the survivors absorb and the record announces; a roster entry nobody can "
      "write is a site that must STOP rather than run at part strength on a "
      "roster somebody wrote down wrong. Absorbing it here would delete the "
      "last trigger main.py's refusal branch has - and test_peakguard.py 10h-1b "
      "drives that branch",
      inverter_b(setpoint="").is_live(_now)[0] is True)
check("...and the status page says so in its own row, which is the surface an "
      "operator actually reads",
      any("“Setpoint entity” is not set" in n for n in
          built(store=MemStore(), second=inverter_b(setpoint=""))[0]
          .liveness_notes()))
check("...and a healthy site adds NO liveness rows at all, so the page is not "
      "carrying a line per inverter that says nothing",
      built(store=MemStore(), second=inverter_b())[0].liveness_notes() == [])


# ==============================================================================
print()
print("8. AC 8: at N=1 the behaviour is unchanged, and the views REFUSE at N>=2")

# ⚠️ THE REAL EVIDENCE FOR AC 8 IS THE OTHER SEVENTEEN FILES, WHICH RUN
# UNMODIFIED. test_freeze, test_loop, test_arbitrage, test_static_tariff,
# test_price, test_tariff and test_validate all build Controllers, set
# `c.dev_min`/`c.dev_max`, read `c.soc`/`c.batt`/`c.sign_wd`, and none of them
# was touched by this card. What is graded HERE is the part they cannot see,
# because every one of them is a site of one: that the compatibility views do
# not quietly answer for the first inverter once there are two.
_c8, _ = built(store=MemStore())
_c8.dev_min, _c8.dev_max = -1234.0, 4321.0
_c8.soc, _c8.batt = 44.0, 55.0
check("at N=1 every view reads and WRITES through to the sole inverter, which "
      "is §6.2's identity and is why an existing install and every existing "
      "replay is byte-identical",
      (_c8.inverters[0].dev_min, _c8.inverters[0].dev_max) == (-1234.0, 4321.0)
      and (_c8.soc, _c8.batt) == (44.0, 55.0)
      and _c8.unit == UNIT_A and _c8.units == (UNIT_A,))
check("...and `sign_wd` and `clamp_last` are the sole inverter's own objects, "
      "not copies - so a test that trips `c.sign_wd` trips the one the cycle "
      "consults", _c8.sign_wd is _c8.inverters[0].sign_wd)

_c8.inverters.append(inverter_b())
_refused = []
for name in VIEWS:
    try:
        getattr(_c8, name)
    except AttributeError as err:
        _refused.append((name, "no single per-device value" in str(err)))
print(f"    with two inverters, {len(_refused)} of {len(VIEWS)} views refuse "
      f"with an explanation")
check("⚠️ AND AT N>=2 EVERY VIEW REFUSES RATHER THAN ANSWERING FOR THE FIRST "
      "INVERTER. THIS is the tripwire on the whole card: a consumer that was "
      "never made per-instance fails on the line that reads it, loudly, "
      "instead of quietly grading inverter B against inverter A's number - "
      "which is the defect, and which is what a view that answered "
      "`self.inverters[0]` would restore",
      len(_refused) == len(VIEWS) and all(ok for _n, ok in _refused))
_why = ""
try:
    _c8.soc
except AttributeError as _err:
    _why = str(_err)
check("...and the refusal NAMES THE WAY OUT - `for inv in "
      "controller.inverters` - because the reader of that traceback is "
      "somebody adding the second inverter and the answer is always the same "
      "one. A bare AttributeError would send them looking for a deleted field",
      "controller.inverters" in _why and "N=1" in _why)


def _settable(obj, name, value):
    try:
        setattr(obj, name, value)
        return True
    except AttributeError:
        return False


check("...and `units` has no setter at all, so a site cannot claim two "
      "actuators while holding one writer. It WAS an assignable tuple, and a "
      "roster written that way is a half-driven site with nothing to say so - "
      "the roster is grown by adding an `Inverter`, which is the object that "
      "carries the writer",
      not _settable(_c8, "units", ("x",))
      and _c8.units == ("inverter-a", "inverter-b"))



# ==============================================================================
print()
print("9. CFG-04: a two-inverter CONFIG becomes a two-unit roster")

# ⚠️ EVERY SECOND UNIT ABOVE WAS CONSTRUCTED BY THIS FILE. That was the whole of
# DEV-02's honesty: the code stopped assuming one inverter, and there was still
# no way for an OPERATOR to declare a second - `build_roster()` returned a list
# of exactly one and said so in its own docstring. This part grades the config
# path: an options dict, through the shipped `build_roster()`, into a roster the
# shipped `live_units()` answers for.
_TWO_OPTS = dict(BASE_OPTS)
_TWO_OPTS.pop("setpoint_entity")
_TWO_OPTS.pop("soc_entity")
_TWO_OPTS.pop("batt_entity")
_TWO_OPTS["inverters"] = [
    dict(name="inverter-a", setpoint_entity="number.sp_a",
         soc_entity="sensor.soc_a", batt_entity="sensor.batt_a",
         max_w=A_RATING, bms_soc_floor=A_BMS_FLOOR),
    # ⚠️ 3000 W IS A HYPOTHESIS AND IS MARKED AS ONE WHEREVER IT APPEARS. It is
    # here because it DIFFERS from A's 5000 W, which is the only property any
    # check below uses. SPIKE-04 measures the real device.
    # ⚠️ AND SO IS ITS 15 % CUT-OUT (SAFETY-32). There is no second pack to read
    # register 0x0560 off; the value is SIMULATED, and it differs from A's 10 %
    # for MUTATION.md habit 1's reason and no other.
    dict(name="inverter-b", setpoint_entity="number.sp_b",
         soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
         max_w=B_RATING_HYPOTHESIS, bms_soc_floor=B_BMS_FLOOR_SIM),
]
_c9, _hass9 = built(_TWO_OPTS)
print(f"    the roster built from `inverters:` -> {[repr(i) for i in _c9.inverters]}")
check("a two-inverter config produces a TWO-unit roster, each with its own "
      "actuator, its own sensors and its own rating - which is the thing this "
      "card exists to make possible at all",
      [i.name for i in _c9.inverters] == ["inverter-a", "inverter-b"]
      and [i.setpoint_entity for i in _c9.inverters] == ["number.sp_a",
                                                         "number.sp_b"]
      and [i.max_w for i in _c9.inverters] == [A_RATING, B_RATING_HYPOTHESIS]
      and [i.soc_entity for i in _c9.inverters] == ["sensor.soc_a",
                                                    "sensor.soc_b"])
check("...and `live_units()` - the method the allocator asks, and the one that "
      "decides what may be commanded this tick - lists BOTH of them",
      _c9.live_units() == ("inverter-a", "inverter-b"))
check("...and the site bound the loop runs on is the SUM, not one unit's "
      f"rating: {A_RATING:.0f} + {B_RATING_HYPOTHESIS:.0f} W",
      _c9.site_max_w == A_RATING + B_RATING_HYPOTHESIS
      and _c9.tuning.max_w == A_RATING + B_RATING_HYPOTHESIS)
check("...and each unit keeps its OWN Store namespace, so a SAFETY-02 trip on "
      "one cannot latch the other - DEV-02's property, now reachable from a "
      "config file instead of from a test fixture",
      _c9.inverters[0].key("sign_tripped") != _c9.inverters[1].key("sign_tripped"))

# ⚠️ AND THE SHAPE IS ANNOUNCED, because the other shape's keys are still
# sitting in the Configuration UI with values in them.
check("...and the action log says which shape is in force and what it is NOT "
      "reading, at every start - an operator whose site runs at half the "
      "rating they typed has one line that says why",
      any("roster:" in e and "inverter-b" in e and "NOT" in e
          for e in _c9.events))

# --- the other half, and it is the half an upgrade depends on ----------------
_c9a, _ = built()                        # BASE_OPTS: the flat keys, no list
_c9b, _ = built({"inverters": []})       # what config.yaml actually ships
check("⚠️ AC 3: an EMPTY `inverters:` is not 'unconfigured', it is the "
      "single-inverter configuration - the flat keys describe the one unit, "
      "under the same name, with the same entities and the same rating as "
      "before this card. That is what an install that upgrades reads, with "
      "nobody touching anything",
      [(i.name, i.setpoint_entity, i.soc_entity, i.batt_entity, i.max_w)
       for i in _c9b.inverters]
      == [(i.name, i.setpoint_entity, i.soc_entity, i.batt_entity, i.max_w)
          for i in _c9a.inverters]
      == [(UNIT_A, "number.sp_a", "sensor.soc_a", "sensor.batt_a", A_RATING)])
check("...and the key ABSENT and the key EMPTY are the same configuration, not "
      "two code paths - config.yaml ships `[]` and an options dict written "
      "before this version has neither",
      _c9b.units == _c9a.units == (UNIT_A,))
# ⚠️ AND THE SHIPPED DEFAULT IS NOT COSMETIC - IT IS WHAT MAKES THE UPGRADE
# SURVIVABLE, AND IT WAS MEASURED. Supervisor 2026.08.0's own AppOptions
# validator - the version this site runs, run against this very file by
# tools/supervisor_shape_probe.py section B - answers `Missing option
# 'inverters' in root` for the shipped `options:` block with ONLY this key
# removed (one key varied, so the refusal is about this key; section B2's
# reconstructed blob is refused for unrelated missing keys and is NOT this
# measurement), and `App.test_update_schema()`
# turns that into "new schema is not compatible" at UPDATE time. The Supervisor
# merges `options:` into the operator's stored blob before validating, so this
# one line is the difference between an upgrade nobody notices and an add-on
# that cannot start. See DECISIONS/CFG-04-inverter-roster.md; deploy-rollback.md
# §4 is the same trap in the other direction.
_cfg_txt_9 = io.open(HERE / "config.yaml", encoding="utf-8").read()
check("⚠️ config.yaml SHIPS `inverters: []` in its options block. Without it "
      "Supervisor refuses every options blob written before this version - "
      "measured on 2026.08.0, not assumed - and the operator is left with an "
      "add-on that will not start",
      "\n  inverters: []\n" in _cfg_txt_9[:_cfg_txt_9.index("\nschema:")])
_sch9 = _cfg_txt_9[_cfg_txt_9.index("\nschema:"):]
_entry9 = _sch9[_sch9.index("\n  inverters:"):]
_entry9 = _entry9[:_entry9.index("\n\n")]
_decl9 = dict(ln.strip().lstrip("- ").split(": ", 1)
              for ln in _entry9.splitlines()[1:] if ": " in ln)
print(f"    the shipped per-entry schema: {_decl9}")
check("...and the schema half declares it as a LIST OF OBJECTS whose `name`, "
      "`setpoint_entity`, `max_w` and `bms_soc_floor` are REQUIRED - no "
      "trailing `?` - so the Supervisor refuses a half-written entry before the "
      "add-on sees it, and an entry can never fall back on a sibling's rating "
      "or on a sibling's BMS cut-out because there is no entry without one. "
      "⚠️ SAFETY-32 ADDED THE FOURTH ONE AS REQUIRED RATHER THAN AS "
      "`int(0,50)?`, and the difference is the whole of that card: an optional "
      "per-pack cut-out is a DEFAULTED one, and a default is one number for "
      "two batteries - the shallower pack protected by the deeper pack's BMS. "
      "⚠️ OPT-09 ADDED `capacity_kwh` AND MADE THE OPPOSITE CHOICE ON PURPOSE: "
      "it is OPTIONAL (`float(0.1,200)?`) because an absent capacity has an "
      "honest answer - the add-on then has no opinion about the SPLIT and every "
      "unit weighs 1.0, which is exactly what shipped - where an absent cut-out "
      "would have to be invented. If you are here because you added a REQUIRED "
      "key, read `Inverter.__init__` before changing this line: a required key "
      "refuses every options blob written before it existed",
      set(_decl9) == {"name", "setpoint_entity", "max_w", "bms_soc_floor",
                      "soc_entity", "batt_entity", "batt_invert",
                      "capacity_kwh"}
      and not any(_decl9[k].endswith("?")
                  for k in ("name", "setpoint_entity", "max_w",
                            "bms_soc_floor"))
      and all(_decl9[k].endswith("?")
              for k in ("soc_entity", "batt_entity", "batt_invert",
                        "capacity_kwh")))

# ⚠️ NOTHING IS INHERITED, AND THIS IS DEV-02 RESTATED AS A CONFIG RULE.
# ⚠️ THE FLAT KEYS ARE SET TO REAL ENTITIES HERE, AND THAT IS THE WHOLE CHECK.
# The first draft left them blank, so "inherited" and "not inherited" produced
# the SAME empty string and the check passed against a `build_roster()` that
# DID inherit - a mutation this file ran and watched survive. An assertion that
# X is not Y has to be run with X and Y actually different (MUTATION.md).
_c9c, _ = built({"inverters": [dict(name="inverter-a",
                                    setpoint_entity="number.sp_a",
                                    max_w=A_RATING)],
                 "soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a"})
check("an entry that omits `soc_entity` has NO SoC sensor and does not borrow "
      "the flat one - which is set to a real entity right here, so inheriting "
      "and not inheriting are visibly different answers. A unit silently "
      "graded against a sibling's measurement is the exact defect DEV-02 spent "
      "a card removing, and a three-line fallback here would rebuild it in YAML",
      _c9c.inverters[0].soc_entity == "" and _c9c.inverters[0].batt_entity == ""
      and _c9c.o["soc_entity"] == "sensor.soc_a")

_c9d, _ = built({"inverters": [
    dict(name="inverter-a", setpoint_entity="number.sp_a", max_w=A_RATING),
    dict(name="inverter-b", setpoint_entity="number.sp_b")]})
check("...and an entry that omits `max_w` does not inherit the flat rating "
      "either - it gets 0 W and validate.py rule 9 refuses it by name, which "
      "is a refusal an operator can act on rather than a site quietly running "
      "at somebody else's number",
      _c9d.inverters[1].max_w == 0.0)


# ==============================================================================
print()
print("10. CFG-04: rule 9, and SAFETY-28's hazard now expressible for the "
      "first time")

from app.validate import (  # noqa: E402
    actuator_writers, unit_caps, unit_caps_indexed, validate_startup_config,
    validate_unit_ranges, _unit_label,
)


def _why(problems, needle):
    return any(needle in p for p in problems)


def _opts(**kw):
    """A complete, ACCEPTED options dict, so a refusal below is the new one."""
    o = dict(BASE_OPTS, meter_source="off", sign_disagree_w=300,
             sign_disagree_s=180, maintenance_charge_w=2000,
             maintenance_discharge_w=2000, peak_soc_floor=20, bms_soc_floor=10)
    o.update(kw)
    return o


check("the fixture this part refuses things out of is itself ACCEPTED, or "
      "every check below would pass on a config that was already broken",
      validate_startup_config(_opts(), -A_RATING, A_RATING) == [])

# --- AC 5: the two-writers hazard --------------------------------------------
_same = _opts(setpoint_entity="", inverters=[
    dict(name="inverter-a", setpoint_entity="number.sp_a", max_w=A_RATING),
    dict(name="inverter-b", setpoint_entity="NUMBER.SP_A",
         max_w=B_RATING_HYPOTHESIS)])
_p_same = validate_startup_config(_same)
print(f"    two inverters, one entity -> {len(_p_same)} refusal(s)")
check("⚠️ AC 5: two inverters naming the SAME setpoint_entity are REFUSED. "
      "specs §1.5's non-goal has been enforced 'by construction' for the whole "
      "life of this add-on because the schema had exactly one such key; this "
      "is the first configuration in which it is expressible, and it is the "
      "first time that guarantee has had to be a check",
      _why(_p_same, "point at the same inverter entity") and _why(_p_same, "number.sp_a"))
check("...and case-folded, so the hazard cannot be smuggled past on "
      "NUMBER.SP_A versus number.sp_a - which is exactly how it would be "
      "written by someone copying an entity id out of the HA UI",
      _why(_p_same, "inverters[0].setpoint_entity")
      and _why(_p_same, "inverters[1].setpoint_entity"))
check("...and two inverters on two DIFFERENT entities are accepted, because "
      "that is the configuration this card exists for and a validator that "
      "fires on the legitimate case gets switched off",
      validate_startup_config(_opts(setpoint_entity="", inverters=[
          dict(name="inverter-a", setpoint_entity="number.sp_a", max_w=A_RATING,
               bms_soc_floor=A_BMS_FLOOR),
          dict(name="inverter-b", setpoint_entity="number.sp_b",
               max_w=B_RATING_HYPOTHESIS,
               bms_soc_floor=B_BMS_FLOOR_SIM)])) == [])

# --- the rest of rule 9 -------------------------------------------------------
_p_flat = validate_startup_config(_opts(inverters=[
    dict(name="inverter-b", setpoint_entity="number.sp_b",
         max_w=B_RATING_HYPOTHESIS)]))
check("a flat setpoint_entity left set BESIDE a declared roster is refused. It "
      "names an actuator the roster does not, so nothing would ever write it - "
      "a site running quietly smaller than its own configuration, which is the "
      "'silently a different config' outcome this card ranks below a refusal",
      _why(_p_flat, "but the “Inverter list” (inverters) is also filled in"))
_p_dup = validate_startup_config(_opts(setpoint_entity="", inverters=[
    dict(name="twin", setpoint_entity="number.sp_a", max_w=A_RATING),
    dict(name="twin", setpoint_entity="number.sp_b",
         max_w=B_RATING_HYPOTHESIS)]))
check("two entries with the SAME name are refused - the Store namespaces every "
      "per-unit latch by name, so one name is one SAFETY-02 latch for two "
      "units and an operator clearing it clears both",
      _why(_p_dup, "both called") and _why(_p_dup, "twin"))
for _what, _bad, _needle in (
        ("with no name at all",
         dict(setpoint_entity="number.sp_b", max_w=3000), "has no name"),
        ("this controller cannot command",
         dict(name="inverter-b", max_w=3000),
         "has no “Inverter setpoint entity” (setpoint_entity)"),
        ("with no rating",
         dict(name="inverter-b", setpoint_entity="number.sp_b"),
         "no usable “Inverter rating” (max_w)"),
        ("whose rating is not a number",
         dict(name="inverter-b", setpoint_entity="number.sp_b", max_w="lots"),
         "no usable “Inverter rating” (max_w)")):
    check(f"an entry {_what} is refused, and the refusal says which entry and "
          f"which key ({_needle!r}) rather than that something is wrong",
          _why(validate_startup_config(_opts(setpoint_entity="",
                                             inverters=[_bad])), _needle))
check("`inverters` that is not a list at all is refused with the shape it "
      "should have been, rather than crashing inside build_roster()",
      _why(validate_startup_config(_opts(inverters="number.sp_b")),
           "(inverters) is str 'number.sp_b', not a list"))

# --- the permissive readers INSIDE validate.py --------------------------------
# ⚠️ RUN BOTH WAYS, WHICH IS MUTATION.md's RULE. A check that only asserts the
# refusal happens would also pass if the rule fired on everything; the
# counterfactual is the flat reader's own answer on the same config.
_weak = _opts(max_w=A_RATING, sign_disagree_w=3000, setpoint_entity="",
              inverters=[dict(name="inverter-a", setpoint_entity="number.sp_a",
                              max_w=2000),
                         dict(name="inverter-b", setpoint_entity="number.sp_b",
                              max_w=2000)])
print(f"    flat max_w {_weak['max_w']:.0f} W, roster caps {unit_caps(_weak)}, "
      f"sign_disagree_w {_weak['sign_disagree_w']:.0f} W")
check("⚠️ A PERMISSIVE READER INSIDE THE VALIDATOR, FIXED AND GRADED. Rule 5 "
      "compared sign_disagree_w against the FLAT max_w. With 5000 left over "
      "from a single-inverter install and a roster of two 2000 W units, 3000 W "
      "passed that comparison while the SAFETY-02 watchdog could fire on "
      "NEITHER unit - rule 5's own failure mode, reached through rule 5. It "
      "now reads the roster's caps and refuses",
      _why(validate_startup_config(_weak), "sign_disagree_w"))
check("...and the counterfactual: the number the old reader used, on the same "
      "config, does not trip it. Without this line the check above would pass "
      "for a rule that refused everything",
      3000.0 < float(_weak["max_w"]) and min(unit_caps(_weak)) == 2000.0)
check("...and at one inverter `unit_caps()` IS `[max_w]`, so the rule is the "
      "identical comparison it always was and no existing install changes "
      "verdict", unit_caps(_opts()) == [A_RATING])

# --- rules 5 and 6 named the WRONG INVERTER when the two lists desynchronised -
# ⚠️ WHAT THE DEFECT WAS. `unit_caps()` SKIPS an entry with no usable `max_w`,
# so caps position i stopped being roster position i the moment one entry was
# malformed. Rule 5 did `_unit_label(opts, caps.index(weakest))` and rule 6 did
# `caps[0]` beside a hard-wired `_unit_label(opts, 0)` - both feeding a CAPS
# index to a function that indexes the ROSTER. The number in the message was
# right and the NAME beside it was another inverter's: an operator sent to the
# wrong device by the message that exists to stop exactly that.
#
# It is latent, not live - rule 9(d) refuses a junk `max_w` in the same pass, so
# the operator sees both refusals. But "unreachable" was a property of today's
# rule ordering rather than of the code, and `validate_startup_config()` is also
# `sim/inject.py`'s entry point. The index travels with the cap now.
_skew = _opts(setpoint_entity="", sign_disagree_w=1500, maintenance_charge_w=3000,
              maintenance_discharge_w=1000, inverters=[
                  dict(name="inverter-a", setpoint_entity="number.sp_a",
                       max_w="junk"),                       # SKIPPED by unit_caps()
                  dict(name="inverter-b", setpoint_entity="number.sp_b", max_w=2000),
                  dict(name="inverter-c", setpoint_entity="number.sp_c", max_w=1000)])
_p_skew = validate_startup_config(_skew)
print(f"    caps {unit_caps(_skew)} but pairs {unit_caps_indexed(_skew)} - "
      f"roster is {len(_skew['inverters'])} long")
check("the fixture actually desynchronises the two lists, or the two checks "
      "below prove nothing: one entry is skipped, so caps is SHORTER than the "
      "roster and every caps index is off by one",
      unit_caps(_skew) == [2000.0, 1000.0]
      and unit_caps_indexed(_skew) == [(1, 2000.0), (2, 1000.0)])
check("⚠️ RULE 5 NAMES THE UNIT THE NUMBER CAME OFF. The weakest cap is "
      "inverter-c's 1000 W; it sits at caps index 1, and `_unit_label(opts, 1)` "
      "is inverter-B. The refusal says inverter-c",
      _why(_p_skew, "for inverter-c (inverters.max_w) is 1000 W")
      and not _why(_p_skew, "for inverter-b (inverters.max_w) is 1000 W"))
# ⚠️ MAINT-05 MOVED RULE 6's SUBJECT, AND THIS CHECK MOVED WITH IT. It used to
# assert the refusal named inverter-b, because rule 6 graded the FIRST entry
# with a usable cap - `Controller.maint_unit` was `inverters[0]` and the shaping
# claim could be filed nowhere else. There is now one `Maintenance` per roster
# entry, so the shaping claim can be filed on ANY pack and the cap that
# contradicts it first is the SMALLEST - inverter-c's 1000 W. The property this
# check exists for is unchanged and is the reason it is still here: the index
# has to travel with the cap, or the message quotes one unit's rating under
# another unit's name.
check("⚠️ RULE 6 TOO, AND SINCE MAINT-05 IT IS THE WEAKEST PACK IT NAMES. "
      "inverter-c's 1000 W sits at caps index 1, and `_unit_label(opts, 1)` is "
      "inverter-B - so a caps index leaking into a roster index would say "
      "inverter-b here. The refusal says inverter-c, with inverter-c's number",
      _why(_p_skew, "(maintenance_charge_w) is 3000 W, but the “Inverter rating” for "
                    "inverter-c (inverters.max_w) is 1000 W")
      and not _why(_p_skew, "(maintenance_charge_w) is 3000 W, but the “Inverter rating” for "
                            "inverter-b"))
check("...and it is no longer graded against the FIRST usable cap at all, which "
      "is the behaviour change MAINT-05 makes here: inverter-b's 2000 W was the "
      "pre-card answer and appears nowhere in the maintenance refusal now",
      not _why(_p_skew, "(maintenance_charge_w) is 3000 W, but the “Inverter rating” for "
                        "inverter-b (inverters.max_w) is 2000 W"))
check("...and the counterfactual, so this is not a check that would pass on a "
      "validator that named nothing: the OLD expressions still evaluate, on "
      "this same config, to the wrong names",
      _unit_label(_skew, unit_caps(_skew).index(1000.0)) == " (inverter-b)"
      and _unit_label(_skew, 0) == " (inverter-a)")
check("...and rule 9(d) refuses this config anyway - which is WHY the defect "
      "was latent, and is recorded rather than used as a reason not to fix it",
      _why(_p_skew, "inverter-a"))

# --- validate_unit_ranges: rule 2, per unit, against ITS OWN device -----------
_c10, _ = built(_TWO_OPTS)
_c10.inverters[0].dev_min, _c10.inverters[0].dev_max = -A_RATING, A_RATING
# B's ESPHome reports a 2000 W device while the config claims 3000 W for it.
_c10.inverters[1].dev_min, _c10.inverters[1].dev_max = -2000.0, 2000.0
_p_rng = validate_unit_ranges(_c10.inverters)
print(f"    per-unit range check -> {_p_rng}")
check("⚠️ A READER THAT SILENTLY PINNED INVERTER #1, IN main.py. run_control() "
      "re-ran rule 2 behind `if self.o.get('setpoint_entity')` - the FLAT key, "
      "which a roster config leaves blank - so on a two-inverter site NO "
      "unit's max_w was graded against the device it is written to, and the "
      "skip was silent. The per-unit form catches B against B's own ESPHome",
      len(_p_rng) == 1 and _why(_p_rng, "inverter-b")
      and _why(_p_rng, "Lower this inverter's rating to"))
check("...and it says INVERTER-B, not 'the setpoint device' - a site with "
      "three inverters and one bad rating needs to be told which one",
      _why(_p_rng, "number.sp_b") and not _why(_p_rng, "inverter-a"))
check("...and a unit whose ESPHome has not answered is SKIPPED rather than "
      "guessed at, which is the same (None, None) rule the flat check has "
      "always had - refusing there would refuse every cold boot",
      validate_unit_ranges([_c9.inverters[1]]) == [])
check("...and at N=1 the unit the flat call already graded is skipped, so one "
      "conflict is reported once and not twice under two wordings",
      validate_unit_ranges(_c9a.inverters, skip_entity="number.sp_a") == [])
# ⚠️ AND THE ADD-ON IS DRIVEN, NOT GREPPED. The first draft asserted the call's
# SOURCE LINE was in main.py, and the mutation that puts it back inside `if
# entity:` - which is one level of indentation - left that string intact and the
# check green. So `run_control()` is run, on a two-inverter config with no flat
# `setpoint_entity` at all, against an ESPHome that reports a device narrower
# than what B's entry claims.
_hass_rc = Hass({"sensor.grid": 0.0, "sensor.soc_a": 50.0, "sensor.batt_a": 0.0,
                 "sensor.soc_b": 50.0, "sensor.batt_b": 0.0},
                {"number.sp_a": (-A_RATING, A_RATING, 1.0),
                 # B's real device is 2000 W; its roster entry claims 3000 W.
                 "number.sp_b": (-2000.0, 2000.0, 1.0)})
_rc_opts = {k: v for k, v in _TWO_OPTS.items() if k != "setpoint_entity"}
_c_rc = Controller(dict({k: v for k, v in BASE_OPTS.items()
                         if k not in ("setpoint_entity", "soc_entity",
                                      "batt_entity")},
                        **_rc_opts),
                   _hass_rc, MemStore(), Pub(), stop=asyncio.Event())
check("the fixture really does have NO flat setpoint_entity, which is the "
      "condition the old re-check silently skipped on",
      not _c_rc.o.get("setpoint_entity") and len(_c_rc.inverters) == 2)
# ⚠️ WITH A DEADLINE, BECAUSE THE FAILURE MODE IS NOT AN EXCEPTION. A
# `run_control()` that does NOT refuse enters the control loop and never
# returns, so the mutation this check exists to catch turns a red into a HANG -
# and a check that hangs is worse than one that passes wrongly, because CI
# times out with no name attached. The timeout makes "did not refuse" a
# failure with a sentence on it.
async def _drive_rc():
    try:
        await asyncio.wait_for(_c_rc.run_control(), timeout=10.0)
        return ""
    except asyncio.TimeoutError:
        return ("run_control() entered the control loop instead of refusing - "
                "the per-unit range check did not run")


_rc_why = asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
    _drive_rc())
print(f"    run_control() on a two-inverter config -> refused={_c_rc.refused}"
      + (f" [{_rc_why}]" if _rc_why else ""))
check("⚠️ RUN, NOT GREPPED: the shipped `run_control()` REFUSES this config and "
      "never enters the loop. Before this card the whole re-check hung off the "
      "flat `setpoint_entity`, so a roster config skipped it and B's 3000 W "
      "against a 2000 W device was never noticed by anything",
      not _rc_why and _c_rc.refused
      and any("inverter-b" in p for p in _c_rc.refused)
      and _c_rc.stop_event.is_set())
check("...and it wrote nothing on the way out - a refusal is a config that "
      "never ran, not one that ran once",
      not _hass_rc.writes)



# ==============================================================================
print()
print("11. CFG-04: every config.yaml reader in the suite, FED a two-inverter "
      "config")

# ⚠️ THE HAZARD THIS PART EXISTS FOR, AND IT IS THE ONE THE PM PROOFREAD NAMED.
# CFG-03 made config.yaml the single source of truth for defaults, so nine
# checks across this suite now READ it rather than transcribing it. Every one of
# those readers was written when a key appeared at most twice - once under
# `options:`, once under `schema:` - and several disambiguated by taking the
# FIRST match on a `line.strip()`, which throws the indentation away.
#
# CFG-04's `inverters:` is a list of objects and its entries carry a `max_w:` of
# their own. So `max_w` is now declared at TWO nesting levels, and a stripped
# first-match reader answers with the PER-INVERTER one. On the shipped file that
# is harmless by luck - the two bounds are the same `int(100,5000)` - which is
# precisely what makes it dangerous: the reader silently returns inverter #1's
# number and every check that consumes it stays green. This project's rule is
# that a check which stays green after you break the thing it watches was never
# watching, and a reader that is right by coincidence is the same defect one
# layer down.
#
# So: a SYNTHETIC config.yaml is built from the shipped one with the nested
# per-inverter numbers made DIFFERENT from the flat ones (MUTATION.md's rule -
# an assertion that "X is not Y" must be run once with X and Y actually made
# different), and every reader in the suite is executed against it.
import re as _re                                                  # noqa: E402
import tempfile as _tempfile                                      # noqa: E402

_SHIPPED_CFG = io.open(HERE / "config.yaml", encoding="utf-8").read()

# The per-inverter numbers, chosen only because they differ from BOTH flat
# answers (options `max_w: 2000`, schema `int(100,5000)`). Nothing is measured
# here and inverter B still does not exist.
_NESTED_W = "3141"

_two_cfg = _SHIPPED_CFG.replace(
    "  inverters: []\n",
    "  inverters:\n"
    "    - name: inverter-a\n"
    "      setpoint_entity: number.sp_a\n"
    f"      max_w: {_NESTED_W}\n"
    "    - name: inverter-b\n"
    "      setpoint_entity: number.sp_b\n"
    f"      max_w: {_NESTED_W}\n", 1)
_two_cfg = _two_cfg.replace("      max_w: int(100,5000)\n",
                            f"      max_w: int(100,{_NESTED_W})\n", 1)
check("the synthetic two-inverter config really did get built - it differs "
      "from the shipped file, and the per-inverter numbers really are "
      "different from the flat ones, or every reader below would 'pass' by "
      "answering the same number either way",
      _two_cfg != _SHIPPED_CFG
      and _two_cfg.count(f"max_w: {_NESTED_W}") == 2
      and f"max_w: int(100,{_NESTED_W})" in _two_cfg
      and _NESTED_W not in _SHIPPED_CFG)

# ⚠️ AND THE SHIPPED FILE ITSELF STILL BAITS THE TRAP. If somebody moves
# `inverters:` below `max_w:` in the schema block, the stripped-first-match bug
# stops being reachable on the shipped file and this whole part quietly becomes
# a check about a synthetic fixture. Assert the ordering rather than trusting a
# comment to preserve it.
_schema_half = _SHIPPED_CFG[_SHIPPED_CFG.index("\nschema:"):]
check("⚠️ the shipped config.yaml still declares the NESTED per-inverter "
      "`max_w:` BEFORE the flat one under `schema:`, so a stripped-line reader "
      "would genuinely take the wrong one. Reorder that and the readers below "
      "become right by accident and this part stops watching anything",
      _schema_half.index("      max_w: int(") < _schema_half.index("\n  max_w: int("))

_TMPCFG = Path(_tempfile.mkdtemp(prefix="cfg04-"))
io.open(_TMPCFG / "config.yaml", "w", encoding="utf-8").write(_two_cfg)

# --- who the readers ARE, found rather than listed ---------------------------
# ⚠️ THE TABLE IS AUDITED, THE DISCOVERY IS MECHANICAL, AND BOTH HALVES MATTER.
# A hand-written list of readers is a list that goes stale the moment somebody
# adds the tenth one - and the tenth one is exactly the one that gets a
# stripped-line idiom typed for it. So every function in every test file whose
# body touches config.yaml is DISCOVERED by AST, and a discovered function that
# is not in the table below fails this part rather than being skipped.
_AUDITED = {
    ("test_peakguard.py", "read_option"),
    ("test_peakguard.py", "read_schema_max"),
    ("test_peakguard.py", "read_schema_min"),
    ("test_peakguard.py", "_schema_bound"),
    ("test_inverter.py", "read_schema_max"),
    ("test_maintenance.py", "shipped_default"),
    ("test_maintenance.py", "schema_range"),
    ("test_freeze.py", "_cfg_default"),
    # SAFETY-37. Found by this very check on its first run - `option_default()`
    # reads the two consumer windows the freshness bound is derived from, so it
    # is a key-taking reader and belongs here rather than outside the audit.
    ("test_freshness.py", "option_default"),
    ("test_price.py", "shipped"),
    ("test_tariff.py", "shipped_option"),
    # SAFETY-42. `read_option()` there takes a key and reads config.yaml, so the
    # signature puts it in this table whether anybody remembered it or not -
    # which is what this table is for. It answers the SITE value because it is
    # anchored at exactly two spaces; a loose-whitespace pattern would have
    # answered a per-inverter `max_w`, which is the defect below.
    ("test_delivery.py", "read_option"),
}
# Helpers the audited readers call. Not called with a key themselves, so there
# is nothing to feed them; they are exec'd so their callers work.
_HELPERS = {("test_peakguard.py", "_cfg_halves"), ("test_peakguard.py", "_top_level")}
_CFG_GLOBALS = ("_CFG_TXT", "_CFG_TEXT", "_cfg_lines", "_cfg_txt", "_cfg_src",
                "_SCHEMA_AT", "_txt")
# ⚠️ AND THE INDIRECT ONES, WHICH IS WHERE THE FIRST DRAFT OF THIS AUDIT WENT
# WRONG. `read_schema_max()` and `_schema_bound()` in test_peakguard.py do not
# contain the string "config.yaml" at all - they go through `_cfg_halves()` -
# so a predicate that only looks for the filename finds nine readers and misses
# the two that CFG-04 actually had to change. Naming a helper counts as reading.
_HELPER_NAMES = tuple(n for _f, n in _HELPERS)


def _reads_cfg(src):
    return ("config.yaml" in src
            or any(g in src for g in _CFG_GLOBALS)
            or any(f"{h}(" in src for h in _HELPER_NAMES))


_discovered, _sources, _all_fns = set(), {}, {}
for _path in sorted(HERE.glob("test_*.py")):
    _body = io.open(_path, encoding="utf-8").read()
    if "config.yaml" not in _body:
        continue
    for _node in ast.walk(ast.parse(_body)):
        if not isinstance(_node, ast.FunctionDef):
            continue
        _seg = ast.get_source_segment(_body, _node) or ""
        _all_fns[(_path.name, _node.name)] = _seg
        if _reads_cfg(_seg):
            _discovered.add((_path.name, _node.name))
_sources.update({k: _all_fns[k] for k in _discovered})
# ⚠️ TO A FIXPOINT, because reading is TRANSITIVE and the first draft of this
# audit stopped at one hop. `read_schema_max()` calls `_schema_bound()` which
# calls `_cfg_halves()`; only the last of those three names the file, and the
# first two are among the readers CFG-04 had to change. A one-hop predicate
# found nine readers, missed those two, and would have reported "all clear" for
# the exact functions this part was written about.
while True:
    _grew = False
    for (_f, _n), _seg in _all_fns.items():
        if (_f, _n) in _discovered:
            continue
        if any(_df == _f and f"{_dn}(" in _seg for _df, _dn in _discovered):
            _discovered.add((_f, _n))
            _sources[(_f, _n)] = _seg
            _grew = True
    if not _grew:
        break

# ⚠️ WHAT MAKES A DISCOVERED FUNCTION AUDITABLE IS ITS SIGNATURE, NOT A LIST
# SOMEBODY MAINTAINS. A reader is a thing you hand a key and get a value from,
# so `def f(key)` / `def f(key, which)` is the shape, and EVERY function of that
# shape in the closure above has to be in the table - which makes the table
# self-maintaining: write the eleventh reader and this check names it on the
# first run. The rest of the closure are CONSUMERS (a `replay()` that takes a
# number `_cfg_default()` produced), and there is no key to feed them; what they
# are held to instead is that they do not do their own key matching, one check
# further down.
def _key_shaped(src):
    _sig = src.split("(", 1)[1].split(")", 1)[0] if "(" in src else ""
    _args = [a.split("=")[0].split(":")[0].strip()
             for a in _sig.split(",") if a.strip()]
    return _args[:1] == ["key"]


_key_takers = {k for k in _discovered if _key_shaped(_sources[k])}
_consumers = _discovered - _key_takers - _HELPERS
print(f"    {len(_discovered)} functions in the config.yaml read-closure; "
      f"{len(_key_takers)} take a key and are fed the two-inverter config, "
      f"{len(_HELPERS)} are helpers, {len(_consumers)} are consumers")
check("every key-taking reader in the suite is in the audited table - the "
      "table is not maintained by hand, the SIGNATURE decides, so the eleventh "
      f"reader cannot arrive unwatched. Unaudited: {sorted(_key_takers - _AUDITED)}",
      _key_takers == _AUDITED)
check("...and the closure is transitive, so a reader that never names the file "
      "because it goes through a helper is still in it - which is where the "
      "first draft of this audit failed, on the two functions CFG-04 changed",
      ("test_peakguard.py", "read_schema_max") in _discovered
      and "config.yaml" not in _sources[("test_peakguard.py", "read_schema_max")])

# --- and now RUN each of them against the two-inverter config -----------------
_answers, _broken = {}, []
for _key in sorted(_AUDITED):
    _file, _fn = _key
    _ns = {"io": io, "re": _re, "ast": ast, "Path": Path, "HERE": _TMPCFG,
           "_CFG_TXT": _two_cfg, "_CFG_TEXT": _two_cfg, "_txt": _two_cfg,
           "_cfg_txt": _two_cfg, "_cfg_src": _two_cfg,
           "_cfg_lines": _two_cfg.splitlines(),
           "_SCHEMA_AT": _two_cfg.index("\nschema:")}
    # Every function of that file that is in the read-closure, so a reader
    # that delegates (read_schema_max -> _schema_bound -> _cfg_halves) resolves.
    for _dep in sorted(_HELPERS | _key_takers):
        if _dep[0] == _file and _dep != _key:
            exec(_sources[_dep], _ns)                             # noqa: S102
    exec(_sources[_key], _ns)                                     # noqa: S102
    _f = _ns[_fn]
    try:
        _answers[_key] = _f("max_w", 1) if _fn == "_schema_bound" else _f("max_w")
    except Exception as _err:                                     # noqa: BLE001
        # ⚠️ AN EXCEPTION IS A PASS HERE. "Handle the per-inverter shape or FAIL
        # LOUDLY on it" - a reader that refuses to answer an ambiguous question
        # is doing the second, and it is the answer that quietly names inverter
        # #1 that this part is hunting.
        _answers[_key] = f"raised {type(_err).__name__}"
for _key, _got in sorted(_answers.items()):
    print(f"      {_key[0]}:{_key[1]}('max_w') -> {_got!r}")
    if _NESTED_W in str(_got):
        _broken.append(f"{_key[0]}:{_key[1]}")
    elif _got is None:
        _broken.append(f"{_key[0]}:{_key[1]} (None - answered nothing at all)")

check("⚠️ AC 4/AC 6: NOT ONE reader answers with the PER-INVERTER number when "
      "fed a two-inverter config. A reader that silently returns inverter #1's "
      f"value fails right here. Offenders: {_broken}", not _broken)
check("...and every one of them still answers SOMETHING - a reader that went "
      "None on the new shape would take its whole check with it, which is "
      "fail-safe but is not the same as handling it",
      all(v is not None for v in _answers.values()))
check("...and the answers are the SITE numbers: the options half says the "
      "shipped default 2000, the schema half says the shipped ceiling 5000",
      _answers[("test_peakguard.py", "read_option")] == 2000.0
      and _answers[("test_peakguard.py", "read_schema_max")] == 5000.0
      and _answers[("test_inverter.py", "read_schema_max")] == 5000.0
      and _answers[("test_tariff.py", "shipped_option")] == 2000.0
      and _answers[("test_freeze.py", "_cfg_default")] == 2000.0
      and _answers[("test_price.py", "shipped")] == 2000
      and _answers[("test_maintenance.py", "shipped_default")] == "2000"
      and _answers[("test_maintenance.py", "schema_range")] == (100.0, 5000.0))

# --- the reads that are NOT functions ----------------------------------------
# ⚠️ SAID PLAINLY: THIS ONE IS A TEXT BACKSTOP, NOT A BEHAVIOURAL CHECK, and it
# is here because six files read config.yaml at MODULE level (test_validate's
# `_shipped` regex, test_watchdog's `circulation_*` scrape, test_arbitrage's and
# test_price_forecast's `\n  key:` counts) where there is no function to call.
# Every one of those is already anchored at exactly two spaces and none of them
# changed verdict on the two-inverter file - but the way to keep that true is to
# ban the idiom that was not: a stripped line, matched by prefix, first hit
# wins. If a module-level read ever needs to become auditable, make it a
# function and it lands in the table above automatically.
# ⚠️ THE NEEDLE IS ASSEMBLED AT RUNTIME so that this check's own source does
# not match it, and it is scoped to lines within ten of a config.yaml
# reference: the same three tokens over a plugin DOCSTRING, looking for a
# SOURCES: line, is an entirely different thing (test_price.py part 4) and a
# blanket grep fails on it.
_NEEDLE = ".strip()" + ".startswith("
_NEAR = ("config.yaml", "_CFG", "_cfg", "_txt")
_permissive = []
for _path in sorted(HERE.glob("test_*.py")):
    _body = io.open(_path, encoding="utf-8").read()
    if "config.yaml" not in _body:
        continue
    _lines = _body.splitlines()
    for _n, _line in enumerate(_lines, 1):
        if _NEEDLE not in _line:
            continue
        _window = "\n".join(_lines[max(0, _n - 11):_n])
        if any(_w in _window for _w in _NEAR):
            _permissive.append(f"{_path.name}:{_n}")
check("no check in this suite matches a config.yaml key on a STRIPPED line any "
      "more - the idiom that made indentation invisible, and with it the "
      f"difference between the site's max_w and one inverter's. Found: {_permissive}",
      not _permissive)

# ⚠️ AND THE ADD-ON ITSELF READS NONE OF THIS. The Dockerfile does not copy
# config.yaml into the image, so every reader above is a CHECK reading the
# shipped file, and app/ code cannot have this class of bug at all. Asserted,
# because "the running add-on cannot read it" is load-bearing for CFG-03's whole
# single-source argument and is exactly the kind of fact that stops being true.
# ⚠️ A STRING CONSTANT, NOT A MENTION. Half of app/ discusses config.yaml in
# its comments - that is CFG-03 working - and grepping the text would fail on
# the documentation rather than on a read.
_app_opens = []
for _f in sorted((HERE / "app").glob("*.py")):
    for _node in ast.walk(ast.parse(io.open(_f, encoding="utf-8").read())):
        if isinstance(_node, ast.Constant) and _node.value == "config.yaml":
            _app_opens.append(_f.name)
check("no module under app/ so much as names config.yaml as a PATH, so the "
      "first-match hazard is confined to the checks - CFG-03's in-code mirrors "
      f"exist precisely because the running add-on cannot open this file. "
      f"Found: {sorted(set(_app_opens))}", not _app_opens)


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

# Each row is a one-line defeat of something above, with the check that catches
# it. ⚠️ THESE WERE RUN, NOT REASONED: every one was applied to the shipped
# module on a scratch copy, the tree was PURGED, and the named file was observed
# RED via `tools/mutation_run.sh`. `python3 -B` alone does not save you - a
# size-preserving edit reverted inside one second is invisible to CPython's
# staleness test (TEST-15). MUTATION.md is the method; the table is the record.
MUTATIONS = [
    ("`self.soc = None` restored in Controller.__init__",
     "test_inverter.py part 1: the AST scan of __init__, at the line"),
    ("a view answering `self.inverters[0]` instead of `_only()`",
     "test_inverter.py part 8: 'every view REFUSES rather than answering for "
     "the first inverter'"),
    ("`actual_w = self.inverters[0].batt` - the loop fed one unit's power",
     "test_inverter.py part 5, the call-site pin. ⚠️ NOT test_alloc.py part 8, "
     "and that is worth saying: part 8 closes its loop around a plant IT "
     "models, so it grades the PROPERTY (one unit's power diverges, the site "
     "sum does not) and is blind to which of them main.py passes. Measured: "
     "this mutation left test_alloc.py GREEN"),
    ("the store migration performed but not read back",
     "test_inverter.py part 2: 'the migration REPORTS that it failed'"),
    ("`_grade_store_migration()` reduced to a log line",
     "test_inverter.py part 2: 'the controller REFUSES to come up disarmed'"),
    # ⚠️ THE THREE ROWS BELOW ARE ONE DEFECT SEEN FROM THREE SIDES, and it was
    # found by review rather than by a check: the latch was silently DISARMED
    # by the rollback ROUND TRIP store.py's rule 2 exists for. The guard used to
    # return early on `not rep.ran`, which is every start after the first, so a
    # trip taken inside a downgrade window was discarded by the version marker.
    ("`_grade_store_migration()` gated on `rep.ran` again",
     "test_inverter.py part 2, the ROLLBACK ROUND TRIP: 'AND THE LATCH IS "
     "CAUGHT ANYWAY'. Every other check in that part passes with the gate "
     "restored, because every other one runs on the start that migrates"),
    ("the operator's clear leaving the pre-DEV-02 flat keys set",
     "test_inverter.py part 2: 'THE OPERATOR CAN STILL GET OUT OF IT - ONCE' "
     "and part 3's 'the restart after that clear comes up UNLATCHED' - i.e. "
     "the re-arm-forever hazard, which is why the `rep.ran` gate existed"),
    ("a trip writing only the namespaced keys, not the flat rollback view",
     "test_inverter.py part 3: 'THE TRIP IS MIRRORED ONTO THE PRE-DEV-02 FLAT "
     "PAIR'. A downgrade after a post-migration trip found no latch at all"),
    ("the guard asking only `self.inverters[0]` whether anything is latched "
     "instead of asking the STORE - so B's own mirrored trip re-arms A",
     "test_inverter.py part 3: 'a brand new Controller comes back with B "
     "latched and A NOT' - the flat-key defect returning through its own guard"),
    ("one shared SignWatchdog: every unit graded on inverter A's command",
     "test_inverter.py part 3: 'inverter A did NOT trip'"),
    ("`sign_tripped` written flat again instead of namespaced",
     "test_inverter.py part 3 AND test_watchdog.py's store-key checks"),
    ("`clamp_to_rating(value, self.o['max_w'], A's device range)`",
     "test_inverter.py part 4: 'the clamp for B uses B's numbers'"),
    ("`send_heartbeats()` dispatching in series instead of gathering",
     "test_inverter.py part 6: 'the pair returned in about ONE budget'"),
    ("`cycle()` writing the shares in series",
     "test_inverter.py part 6: 'a REAL cycle delivers B's share while A hangs'"),
    ("`asyncio.wait_for` dropped from the write",
     "test_inverter.py part 6 - and it takes 94 s to say so, which is the "
     "defect measuring itself"),
    ("one shared `last_beat` restored",
     "test_inverter.py part 6: 'the deadline is per inverter too'"),
    ("`live_units()` returning the whole roster again",
     "test_inverter.py part 7 AND test_peakguard.py 10h-2's PRECONDITION"),
    ("the heartbeat re-emitting `self.target` instead of `inv.target`",
     "test_inverter.py part 6: only the overdue unit's own share is written"),
    ("the failsafe substitution keeping `shares=resolution.shares`",
     "test_peakguard.py 10h-1b: 'it WRITES that 0 W to the unit it CAN reach'"),
    ("`_read_unit()` reading `self.o['soc_entity']` for every unit",
     "test_inverter.py part 5: 'read_inputs() fills them from the RIGHT "
     "sensors'"),
    ("`pack_floor_soc` averaging instead of taking the minimum",
     "test_inverter.py part 4: 'the pack floor is the EMPTIEST pack'"),
    ("`device`/`supervised` filed without `inv` - B's rating on A",
     "test_peakguard.py 10g: the static routing pin, at the line"),
    # ⚠️ THE ROW ABOVE IS THE *ACTUATOR* VARIANT AND IT WAS THE ONLY ONE HERE
    # FOR A WHOLE REVIEW ROUND. The BOUND variant below routes every claim
    # correctly and builds the NUMBERS from the wrong unit - a different defect
    # with a different fix, invisible to the routing pin. It survived all 19
    # files, because test_peakguard.py's AC 4 check compared B's bound against
    # B's `max_w` while the fixture defaulted B's `max_w` to A's. Both halves
    # are fixed: B now has a different rating and the bound is asserted against
    # literal numbers. ANY CHECK ASSERTING "X DIFFERS FROM Y" MUST BE RUN ONCE
    # WITH X AND Y ACTUALLY MADE DIFFERENT.
    ("the per-unit `device`/`supervised` BOUNDS read off `self.inverters[0]` "
     "while the `actuator=` stays `inv` - right envelope, wrong numbers",
     "test_peakguard.py 10g: 'AND EACH UNIT'S supervised BOUND IS ITS OWN "
     "max_w' and the two lines under it, which read literal 5000/3000 rather "
     "than an expression that can compare a number against itself"),
    ("the stop button zeroing `self.target` but not each `inv.target`",
     "test_inverter.py part 6: 'the next heartbeat does NOT put the old shares "
     "back' - a hazard THIS CARD created, because the heartbeat re-emits a "
     "share now and not the site number"),
    # --- CFG-04's ten, run the same way (tools/mutation_run.sh) -------------
    ("test_inverter.py's `read_schema_max()` back to matching a STRIPPED line",
     "part 11: 'NOT ONE reader answers with the PER-INVERTER number' names it "
     "as an offender. ⚠️ THE SHIPPED FILE CANNOT CATCH THIS - the two `max_w` "
     "bounds are the same int(100,5000) - which is why part 11 executes the "
     "readers against a synthetic config where they differ"),
    ("test_peakguard.py's `read_option()` back to whole-file first match",
     "part 11, same check, plus the text backstop naming test_peakguard.py:234"),
    ("test_peakguard.py's `_schema_bound()` back to a stripped line",
     "part 11, which names BOTH it and `read_schema_max()` - the caller "
     "delegates, so one edit breaks two readers and the transitive closure "
     "finds both"),
    ("`build_roster()` inheriting the flat `soc_entity` for an entry that "
     "omits it",
     "part 9: 'an entry that omits soc_entity has NO SoC sensor'. ⚠️ THIS ONE "
     "SURVIVED THE FIRST RUN and the FIXTURE was changed, not the rule: the "
     "flat keys were blank, so inherited and not-inherited were the same empty "
     "string. MUTATION.md's own failure mode, found in this file"),
    ("`build_roster()` inheriting the flat `max_w` for an entry that omits it",
     "part 9: '...and an entry that omits max_w does not inherit the flat "
     "rating either'"),
    ("validate.py rule 9(e) dropped - a flat `setpoint_entity` beside a "
     "declared roster silently ignored",
     "part 10: 'a flat setpoint_entity left set BESIDE a declared roster is "
     "refused'"),
    ("validate.py rule 5 back to grading `sign_disagree_w` against the FLAT "
     "`max_w`",
     "part 10: 'A PERMISSIVE READER INSIDE THE VALIDATOR', with the "
     "counterfactual beside it so a rule that refused everything would not pass"),
    ("main.py's per-unit range check put back inside `if entity:`",
     "part 10: 'RUN, NOT GREPPED'. ⚠️ THIS ONE SURVIVED THE FIRST RUN TOO - the "
     "check asserted the call's source line was present, and re-indenting it "
     "leaves the line present. It drives `run_control()` now, under a deadline, "
     "because the mutation's real symptom is a HANG and not a red"),
    ("config.yaml no longer shipping the `inverters: []` default",
     "part 9: 'config.yaml SHIPS inverters: []'. The consequence is not in "
     "this suite at all - Supervisor answers `Missing option 'inverters' in "
     "root` and the add-on will not start after the upgrade - which is why the "
     "check is on the file and the measurement is in "
     "DECISIONS/CFG-04-inverter-roster.md §B. ⚠️ THE PROBE ITSELF WAS "
     "REGRESSION-BLIND TO THIS MUTATION until the review: its section B "
     "asserted `not ok` on a DIFFERENT blob, so deleting the default left it "
     "printing PASS. Section B now varies one key and asserts on the KEY NAME "
     "in the refusal, and the mutation was re-run against the real image: "
     "`2 CASE(S) FAILED`, exit 1"),
    ("validate.py allowing two roster entries with the SAME name",
     "part 10: 'two entries with the SAME name are refused' - one Store "
     "namespace for two units is DEV-02's defect rebuilt in YAML"),
    ("`migrate_unit_keys()`'s marker gate `if rep.ok:` -> `if rep.ok or "
     "rep.kept:` - the schema marker stamped on a migration that DROPPED THE "
     "LATCH, because one key happened to be namespaced already",
     "part 10: 'AND THE STORE IS NOT MARKED MIGRATED'. ⚠️ THIS ONE SURVIVED "
     "ALL 19 FILES and was found by MAINT-05's round-3 battery, not by this "
     "card: every migration check here reached the gate on a store where "
     "NOTHING was namespaced yet, so `rep.kept` was empty in all of them and "
     "the `or` never fired. Stamping there disarms the latch AND tells the "
     "next start the migration finished, so it never re-arms - DEV-02's own "
     "worst case filed under 'done'"),
]
# ⚠️ FOUR OF THESE SURVIVED THE FIRST TIME AND THE CHECKS WERE CHANGED, NOT THE
# TABLE - two in DEV-02, two more in CFG-04, and the CFG-04 pair are recorded in
# their own rows above because both are the same mistake in different clothes: a
# check that observed the TEST FIXTURE (blank flat keys) or the SOURCE TEXT (a
# line that survives re-indentation) rather than the behaviour it was named
# after. The serialised dispatch and the shared `last_beat` both passed while
# part 6 gathered its OWN `write_setpoint()` calls and computed its OWN `due`
# list - a check on this file rather than on the add-on. `send_heartbeats()` was
# extracted from `run_control()` so that the shipped dispatch is reachable by a
# check at all, and both are now driven through it. That is the failure mode
# this repository keeps finding, found once more here, and it is recorded rather
# than quietly fixed.
for what, caught in MUTATIONS:
    print(f"      {what}\n          -> {caught}")
check(f"all {len(MUTATIONS)} defeats of this card are caught by a named check "
      f"above. A guard whose checks survive its own defeat is not a guard, and "
      f"this repository has three separate records of exactly that",
      len(MUTATIONS) == 35)


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