goodwe-addon/goodwe_controller/test_arbiter.py

"""Runnable check for the precedence arbiter.  `python3 test_arbiter.py`

Each assertion is a precedence question someone will eventually ask in the
field: "why did it charge during a peak?", "why did it discharge while the
maintenance cycle was charging?", "what happens if two things disagree?".

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

import sys
from math import inf

from app.arbiter import (
    ACTUATOR_SITE, CLAIM_SCOPE, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE,
    P_MAINTENANCE, P_PEAK, P_SAFETY, SCOPE_SITE, SCOPE_UNIT, UnitBounds,
    Claim, actuator_for, allocate, pre_resolve, resolve, scope_violations,
    site_resolve,
)

fails = []


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


def _raises(fn):
    """True if `fn()` raises - a check that something is REFUSED, not defaulted."""
    try:
        fn()
    except Exception:
        return True
    return False


print("basics")
r = resolve([])
check("no claims at all means 0 W", r.target_w == 0.0 and r.winner == "failsafe")

r = resolve([Claim.set("loop", P_LOOP, 900)])
check("a lone claim is honoured", r.target_w == 900)

r = resolve([Claim.set("loop", P_LOOP, 900),
             Claim.set("maintenance", P_MAINTENANCE, 2500)])
check("higher priority set wins", r.target_w == 2500 and r.winner == "maintenance")

print("limits")
r = resolve([Claim.set("loop", P_LOOP, 4000),
             Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")])
check("safety limit binds the loop", r.target_w == 2000)
check("and says what bound it", "supervised" in (r.bound_by or ""))

r = resolve([Claim.set("loop", P_LOOP, 4000),
             Claim.limit("a", P_SAFETY, hi=3000),
             Claim.limit("b", P_SAFETY, hi=1500)])
check("most restrictive limit wins", r.target_w == 1500)

print("the rule that matters: limits only bind claims at or below their priority")
# Maintenance charge phase forbids discharging...
charge_only = Claim.limit("maintenance", P_MAINT_SHAPE, hi=0, reason="charge-only")

r = resolve([Claim.set("loop", P_LOOP, 900), charge_only])
check("charge-only stops the ordinary loop discharging", r.target_w == 0)

# ...but peak shaving outranks it and must be able to discharge anyway.
r = resolve([Claim.set("loop", P_LOOP, 900), charge_only,
             Claim.set("peak", P_PEAK, 2500)])
check("peak shaving is NOT bound by charge-only", r.target_w == 2500)
check("peak shaving is the winner", r.winner == "peak")

# Safety limits still bind everything, including peak shaving.
r = resolve([Claim.set("peak", P_PEAK, 9000), charge_only,
             Claim.limit("device", P_SAFETY, lo=-5000, hi=5000, reason="rating")])
check("safety limits bind even the highest strategy", r.target_w == 5000)

print("safety stops")
r = resolve([Claim.set("loop", P_LOOP, 900),
             Claim.set("safety", P_SAFETY, 0, "inputs missing")])
check("a safety stop beats everything", r.target_w == 0 and r.winner == "safety")

r = resolve([Claim.set("maintenance", P_MAINTENANCE, 2500),
             Claim.set("safety", P_SAFETY, 0, "stopped")])
check("safety stop beats maintenance too", r.target_w == 0)

print("contradictions are bugs, not ties")
r = resolve([Claim.set("loop", P_LOOP, 500),
             Claim.limit("x", P_SAFETY, lo=1000),
             Claim.limit("y", P_SAFETY, hi=200)])
check("impossible limits command 0 W", r.target_w == 0.0)
check("and are flagged, not silently clipped", r.contradiction)
check("explain() says so", "CONTRADICTION" in r.explain())

print("explainability")
r = resolve([Claim.set("loop", P_LOOP, 900), charge_only])
check("explain names winner and binder",
      "loop" in r.explain() and "maintenance" in r.explain())
print(f"       e.g. {r.explain()!r}")

print("sign handling")
r = resolve([Claim.set("loop", P_LOOP, -3000),
             Claim.limit("supervised", P_SAFETY, lo=-1000, hi=1000)])
check("charging is clamped by the low limit", r.target_w == -1000)

r = resolve([Claim.set("loop", P_LOOP, -200), charge_only])
check("charge-only permits charging", r.target_w == -200)

# =============================================================================
# ARB-03: the actuator specs.md declared, and the three stages
# =============================================================================
# ⚠️ EVERY CHECK ABOVE THIS LINE IS UNEDITED, AND THAT IS ITSELF THE FIRST
# ACCEPTANCE CRITERION. Not one of them names an actuator, so every claim they
# build is a SITE claim - which is exactly the state main.py is in at N=1 - and
# they still answer what they answered before the field existed. If a line above
# had to be changed to accommodate `actuator`, stage 2 would not be the shipped
# arbiter and the refactor would not be pure.

print()
print("ARB-03 1. the field, and what its absence means")

lone = Claim.set("loop", P_LOOP, 900)
check("a claim built without naming an actuator carries one anyway",
      lone.actuator == ACTUATOR_SITE)
# ⚠️ THE TYPE ITSELF, NOT ONLY THE TWO HELPERS. Measured: with only the helper
# checked, emptying the DATACLASS default survives this suite untouched, because
# `Claim.set()` and `Claim.limit()` pass their own default down and the field's
# never runs. Anything that builds a `Claim(...)` directly - a future strategy,
# a `dataclasses.replace()` - would have got the unspecified actuator that this
# card exists to remove.
check("...and so does one built through the dataclass directly, which is the "
      "constructor the two helpers do not cover",
      Claim("loop", P_LOOP).actuator == ACTUATOR_SITE)
check("...and it is an explicit SITE, not None and not an empty string. "
      "'unspecified' is what the shipped code had, and 'unspecified' is what "
      "made every claim implicitly about the one and only actuator",
      isinstance(lone.actuator, str) and lone.actuator != ""
      and lone.actuator is not None)
check("...and `source` is still the first positional argument, which is what "
      "test_peakguard.py part 10f's AST scan reads off main.py to pin the "
      "claim-source set. specs.md declares `actuator` first; the tuple position "
      "is not the part of that declaration that carries meaning, and moving "
      "`source` off zero would blind the tripwire that guards SAFETY-25",
      Claim.limit("device", P_SAFETY, -1, 1).source == "device")

print()
print("ARB-03 2. the split: eight sources, each classified where it is BUILT")
# The classification is graded against what each claim is ABOUT, restated here
# so that the table in arbiter.py and the expectation in a test are not the same
# sentence read twice. test_peakguard.py part 10f pins this set against main.py's
# own AST, so a ninth source cannot arrive without both files noticing.
SITE_SOURCES = {
    "loop":      "regulates grid_w, the P1 measurement - one meter, one site",
    "peak":      "the quarter-hour average offtake AT the grid connection",
    "arbitrage": "buys and sells at the grid against a day-ahead price",
    "safety":    "'stopped' is the site's switch, 'inputs missing' is the "
                 "site's own meter - neither is about a pack",
}
UNIT_SOURCES = {
    "device":     "one inverter's own reported min/max rating",
    "supervised": "max_w caps what may be WRITTEN to one setpoint entity",
    "failsafe":   "site-wide evidence, per-unit remedy: SAFETY-02 grades the "
                  "command THIS controller put on THIS actuator",
    "maintenance": "the monthly recalibration is a pack procedure",
}
for _src, _why in sorted(SITE_SOURCES.items()):
    print(f"    site  {_src:<11} {_why}")
for _src, _why in sorted(UNIT_SOURCES.items()):
    print(f"    unit  {_src:<11} {_why}")
check("all eight claim sources are classified, and only those eight",
      set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES))
check("the four that are about the GRID CONNECTION are site claims",
      {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE} == set(SITE_SOURCES))
check("the four that are about a PACK, a DEVICE or one write are per-inverter",
      {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT} == set(UNIT_SOURCES))
check("a source nobody classified cannot be routed at all - actuator_for() "
      "raises rather than defaulting to the site, because defaulting to the "
      "site is the failure this card removes",
      _raises(lambda: actuator_for("reserve", "inverter-a", ("inverter-a",))))

print()
print("ARB-03 3. at N=1 the site IS the unit, so nothing moves")
# §6.2's identity: with one inverter the fabricated grid reading and the battery
# setpoint are the same number, so there is one actuator and it is both.
ONE = ("inverter-a",)
TWO = ("inverter-a", "inverter-b")
check("with one inverter every source routes to the site, per-inverter ones "
      "included - that is the identity, not the classification being ignored",
      all(actuator_for(s, "inverter-a", ONE) == ACTUATOR_SITE for s in CLAIM_SCOPE))
check("...and with none configured at all, likewise",
      all(actuator_for(s, "inverter-a", ()) == ACTUATOR_SITE for s in CLAIM_SCOPE))
check("with TWO the identity breaks and the split becomes real: the four "
      "per-inverter sources route to the unit this controller drives, the four "
      "site sources stay on the grid connection",
      {s: actuator_for(s, "inverter-a", TWO) for s in CLAIM_SCOPE}
      == {**{s: "inverter-a" for s in UNIT_SOURCES},
          **{s: ACTUATOR_SITE for s in SITE_SOURCES}})

print()
print("ARB-03 4. stage 2 IS the shipped arbiter (AC 3)")
# ⚠️ NOT ASSERTED, REPLAYED. Every claim list this file resolves above is run
# through BOTH entry points and every field the shipped code reads off a
# Resolution is compared. `resolve()` at N=1 is `site_resolve()` on the same
# list, so a single edit inside stage 2 would show up in every check above as
# well as here - and one that only affects the wrapper shows up only here.
REPLAY = [
    [],
    [Claim.set("loop", P_LOOP, 900)],
    [Claim.set("loop", P_LOOP, 900), Claim.set("maintenance", P_MAINTENANCE, 2500)],
    [Claim.set("loop", P_LOOP, 4000),
     Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")],
    [Claim.set("loop", P_LOOP, 4000), Claim.limit("a", P_SAFETY, hi=3000),
     Claim.limit("b", P_SAFETY, hi=1500)],
    [Claim.set("loop", P_LOOP, 900), charge_only],
    [Claim.set("loop", P_LOOP, 900), charge_only, Claim.set("peak", P_PEAK, 2500)],
    [Claim.set("peak", P_PEAK, 9000), charge_only,
     Claim.limit("device", P_SAFETY, lo=-5000, hi=5000, reason="rating")],
    [Claim.set("loop", P_LOOP, 900), Claim.set("safety", P_SAFETY, 0, "inputs missing")],
    [Claim.set("loop", P_LOOP, 500), Claim.limit("x", P_SAFETY, lo=1000),
     Claim.limit("y", P_SAFETY, hi=200)],
    [Claim.set("loop", P_LOOP, -3000),
     Claim.limit("supervised", P_SAFETY, lo=-1000, hi=1000)],
    [Claim.set("loop", P_LOOP, -200), charge_only],
    [Claim.set("arbitrage", P_ARBITRAGE, -2000, "buy"),
     Claim.limit("peak", P_PEAK, lo=800.0, reason="quarter peak"),
     Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")],
]


def _shape(r):
    return (r.target_w, r.winner, r.bound_by, r.contradiction,
            list(r.considered), r.explain())


_diff = [i for i, cl in enumerate(REPLAY)
         if _shape(resolve(cl)) != _shape(site_resolve(cl))]
print(f"    replayed {len(REPLAY)} claim lists through both entry points")
check("NOT ONE of them resolves differently - target, winner, binder, "
      "contradiction flag, the considered list and the operator line are equal "
      "on every one. At N=1 the three stages are the shipped arbiter with "
      "bookkeeping either side of it", _diff == [] and len(REPLAY) >= 12)
check("...and stage 3 hands the whole target to the one actuator there is, "
      "which is the N=1 identity rather than an allocation",
      resolve([Claim.set("loop", P_LOOP, 900)]).shares == {ACTUATOR_SITE: 900.0})
check("...and says it allocated, because at N<=1 there is nothing to divide",
      resolve([Claim.set("loop", P_LOOP, 900)]).allocated)

print()
print("ARB-03 5. THE CONTRADICTION THAT CANNOT BE WRITTEN DOWN (AC 4)")
# SAFETY-25 §0a, constructed. A per-inverter reserve says "this pack must not
# discharge" (`hi = 0`); the peak guard says "the site must deliver at least
# 1500 W" (`lo = 1500`). At N=1 those are two bounds on ONE number and they
# genuinely cross - which is today's behaviour, and today's hazard, and is left
# exactly as it is. Filed on TWO actuators they are bounds on two different
# numbers, and there is no expression in which one is the other's counterpart.
SITE_SIDE = [Claim.set("loop", P_LOOP, -1500.0, "grid"),
             Claim.limit("peak", P_PEAK, lo=1500.0, reason="quarter peak")]
CROSSED = SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0,
                                   reason="charge-only")]
SPLIT = SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0,
                                 reason="charge-only", actuator="inverter-b")]
r_crossed, r_split = resolve(CROSSED), resolve(SPLIT)
print(f"    both on one actuator: {r_crossed.explain()}")
print(f"    the `hi` on inverter-b: {r_split.explain()}   "
      f"inverter-b bounds {r_split.unit_bounds['inverter-b'].explain()}")
check("filed on ONE actuator the pair still contradicts and still parks the "
      "battery at 0 W in the billed quarter. That is the shipped behaviour and "
      "this card does not change it - at N=1 the two really are bounds on one "
      "number", r_crossed.contradiction and r_crossed.target_w == 0.0)
check("filed on TWO the site keeps its floor: the unit `hi` is not a term of "
      "the site's `lo > hi` test, so there is no contradiction to flag",
      not r_split.contradiction and r_split.target_w == 1500.0)
check("...and the constraint is not lost, it is CARRIED - inverter-b's own "
      "interval still says 'do not discharge', for the allocator to honour when "
      "it divides the 1500 W", r_split.unit_bounds["inverter-b"].hi == 0.0
      and r_split.unit_bounds["inverter-b"].hi_src == "maintenance")
check("...and the site target is bit-for-bit what the site claims alone "
      "resolve to, so the unit claim contributed NOTHING to it - which is the "
      "'cannot be expressed' rather than 'is guarded against'",
      _shape(site_resolve(SITE_SIDE))[:4] == _shape(r_split)[:4])

# Swept rather than argued: no per-inverter source, at any bound, on any unit,
# can reach the site's limits.
_SWEEP_N = _SWEEP_BAD = 0
for _src in sorted(UNIT_SOURCES):
    for _unit in ("inverter-a", "inverter-b"):
        for _lo, _hi in ((0.0, inf), (-inf, 0.0), (2500.0, inf), (-inf, -2500.0),
                         (900.0, 100.0)):
            _r = resolve(SITE_SIDE + [Claim.limit(_src, P_MAINT_SHAPE, lo=_lo,
                                                  hi=_hi, actuator=_unit)])
            _SWEEP_N += 1
            if _r.contradiction or _r.target_w != 1500.0:
                _SWEEP_BAD += 1
print(f"    swept {_SWEEP_N} (per-inverter source x unit x bound) filings "
      f"against a live site floor: {_SWEEP_BAD} disturbed the site target")
check(f"NOT ONE of the {_SWEEP_N} filings contradicts the site floor or moves "
      f"the site target - including the last bound of each row, which is a unit "
      f"contradicting ITSELF (lo 900 > hi 100). A unit's own impossible pair "
      f"parks that unit; it does not park the site",
      _SWEEP_BAD == 0 and _SWEEP_N >= 40)
_self_bad = resolve(SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE,
                                             lo=900.0, hi=100.0,
                                             actuator="inverter-b")])
check("...and that unit says so on its own bounds, so the fault is reported "
      "where it was made rather than swallowed",
      _self_bad.unit_bounds["inverter-b"].contradiction
      and "CONTRADICTION" in _self_bad.unit_bounds["inverter-b"].explain())

print()
print("ARB-03 6. stage 1: one unit's claims, one unit's interval")
mb = pre_resolve([Claim.limit("device", P_SAFETY, -5000, 5000, "rating"),
                  Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w"),
                  Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only")])
print(f"    a unit filing rating, max_w and charge-only: {mb.explain()}")
check("the most restrictive of the unit's own limits is the interval",
      mb.lo == -2000 and mb.hi == 0.0 and mb.hi_src == "maintenance"
      and mb.lo_src == "supervised")
owned = pre_resolve([Claim.set("maintenance", P_MAINTENANCE, -2500, "charge"),
                     Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w"),
                     Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only")])
print(f"    a unit whose maintenance owns the setpoint: {owned.explain()}")
check("a unit's own winning SET gates its own limits exactly as clause 2 does "
      "at the site - the P_MAINT_SHAPE ceiling is not applied to a "
      "P_MAINTENANCE drain that outranks it", owned.hi == 2000.0)
check("...and the SET's value travels BESIDE the interval, not inside it. "
      "Folding -2500 into [-2500, -2500] would turn 'clamped to max_w' into "
      "'contradicts max_w', which is a different answer and a wrong one",
      owned.pinned_w == -2500 and owned.pinned_by == "maintenance"
      and owned.gate == P_MAINTENANCE and not owned.contradiction)

print()
print("ARB-03 7 / ALLOC-01: the seam, and what now occupies it")
# ⚠️ THIS PART WAS "the seam is a seam, and NOT an allocator" AND IT GRADED THE
# DECLINE. ALLOC-01 filled the seam, so the decline at N>=2 is gone - and the
# checks that graded it are RE-AIMED rather than deleted, because the two halves
# they were really about are both still true and both still worth a line: the
# N<=1 pass-through is byte-identical, and the seam still does not BROADCAST.
check("with no units the whole target goes to the site actuator",
      allocate(1500.0, {}) == ({ACTUATOR_SITE: 1500.0}, True))
check("with one unit the share IS the site target - §6.2's N=1 identity, "
      "pass-through. ⚠️ A FLOAT, UNROUNDED AND UNCLAMPED: this is the "
      "compatibility promise every N=1 replay in this repository depends on, "
      "and the general path below does neither of those things",
      allocate(1500.0, {"inverter-a": UnitBounds()})
      == ({"inverter-a": 1500.0}, True))
check("...and it is byte-identical on a fractional target too, which is where a "
      "'tidy this into the general path' refactor would show up: the general "
      "path rounds to the 1 W register, the pass-through must not",
      allocate(1500.4, {"inverter-a": UnitBounds()})
      == ({"inverter-a": 1500.4}, True))
_shares, _ok = allocate(1500.0, {"inverter-a": UnitBounds(),
                                 "inverter-b": UnitBounds()})
print(f"    two units, ALLOC-01: shares={_shares} allocated={_ok}")
check("⚠️ AND THE ONE THING THE SEAM MUST NEVER DO, WHICH IS THE WHOLE OF "
      "ARCH-01: it does not BROADCAST. Hand the site target to each inverter "
      "and each supplies it IN FULL - the gain at the meter becomes N times the "
      "per-unit gain, which admin/goodwe's two-master rig measures as a 4000 W "
      "peak-to-peak limit cycle at gain 1.0 with not one guard firing. This "
      "line went red on ARB-03's decline and it goes red on a broadcast, which "
      "is the point: it grades the OUTPUT, not the mechanism",
      1500.0 not in _shares.values())
check("...it allocates instead, and the shares SUM to the site target exactly. "
      "That is the safety property: sum(s_i) = S is what makes the gain at the "
      "meter a convex combination of the per-unit gains for ANY split, so the "
      "weights are a comfort choice and a bad split cannot destabilise the site",
      _ok is True and sum(_shares.values()) == 1500
      and sorted(_shares) == ["inverter-a", "inverter-b"])
check("...and an odd target still sums exactly, which is the half a per-unit "
      "round() gets wrong: 1501/2 is 750.5 twice, and two roundings of that are "
      "either 1500 or 1502. Largest remainder gives 751 and 750",
      sum(allocate(1501.0, {"inverter-a": UnitBounds(),
                            "inverter-b": UnitBounds()})[0].values()) == 1501)
check("⚠️ AND THE DECLINE IS NOT GONE, IT IS RE-AIMED. It is still correct when "
      "no safe split exists at all, and PR #40's reviewer proved that branch was "
      "unreachable and untested once already - removing its trigger without "
      "giving it a new one would rebuild that hole. Here: every inverter's "
      "master stale, so there is no envelope set to allocate into",
      allocate(1500.0, {"inverter-a": UnitBounds(),
                        "inverter-b": UnitBounds()}, live=()) == ({}, False))

# ⚠️ AND `resolve()` HAS TO HONOUR THE ROSTER IT IS GIVEN, WHICH IS NOT THE SAME
# THING AS HAVING A `units` PARAMETER. Left None the roster is DERIVED from the
# claims, and that derivation is right only for a caller that files a claim for
# every unit every cycle. Ignore the argument and derive it always - `if units
# is None:` -> `if True:` - and a unit that filed nothing this cycle simply
# CEASES TO EXIST: the site of two looks like a site of one, the seam passes the
# whole grid target through to the single inverter that spoke, and ARCH-01's
# fault is built with the evidence for it removed. That mutation left this whole
# suite green until this check was written (PR #40 review), so it is graded on
# the one observation that distinguishes the two - a SILENT unit.
_silent = resolve([Claim.set("loop", P_LOOP, 900.0),
                   Claim.limit("device", P_SAFETY, -2000, 2000, "rating",
                               actuator="inverter-a")],
                  units=TWO)
print(f"    roster of two, only inverter-a files: bounds="
      f"{sorted(_silent.unit_bounds)} allocated={_silent.allocated} "
      f"shares={_silent.shares}")
check("resolve() HONOURS the roster it is handed rather than inferring one from "
      "who happened to speak. inverter-b filed nothing this cycle and is still "
      "a unit: it gets its own interval, and the site is still a site of TWO, "
      "so the target is SPLIT across two inverters",
      sorted(_silent.unit_bounds) == ["inverter-a", "inverter-b"]
      and _silent.shares == {"inverter-a": 450, "inverter-b": 450})
check("...and that is the whole cost of ignoring the argument, stated as the "
      "number it would produce: derive the roster from the claims instead and "
      "the site is a site of ONE, so the seam passes ALL 900 W through to "
      "inverter-a - the silent inverter's absence read as a smaller site, and "
      "inverter-a told to do the whole site's work on its own. ⚠️ ALLOC-01 "
      "makes this mutation WORSE, not better: before, both answers refused, so "
      "only the roster listing distinguished them; now the wrong one produces a "
      "plausible number that is twice what inverter-a should be given",
      resolve([Claim.set("loop", P_LOOP, 900.0),
               Claim.limit("device", P_SAFETY, -2000, 2000, "rating",
                           actuator="inverter-a")]).shares
      == {"inverter-a": 900.0})
check("...and left None the roster IS derived from the claims, which is the "
      "documented default every N=1 replay above depends on - so the parameter "
      "changes the answer only where the two disagree",
      resolve([Claim.set("loop", P_LOOP, 900.0),
               Claim.limit("device", P_SAFETY, -2000, 2000, "rating",
                           actuator="inverter-a")], units=None).allocated is True)

print()
print("ARB-03 8. verified by breaking it: a claim routed to the wrong stage (AC 6)")
# The `peak` floor, misfiled on a unit. It is the site's floor - if it lands on
# one inverter's interval the SITE keeps no floor at all, and the loop is free to
# go on charging through the quarter that sets the month's bill.
mis_peak = SITE_SIDE[:1] + [Claim.limit("peak", P_PEAK, lo=1500.0,
                                        reason="quarter peak",
                                        actuator="inverter-b")]
r_mis = resolve(mis_peak)
print(f"    peak floor misfiled on inverter-b: {r_mis.explain()}   "
      f"scope_errors={r_mis.scope_errors}")
check("the mis-routing is CAUGHT: scope_violations names the source and the "
      "actuator it does not belong on", len(r_mis.scope_errors) == 1
      and "peak" in r_mis.scope_errors[0])
check("...and the reason it must be caught, measured rather than argued: with "
      "the floor on a unit the SITE has no floor, so the loop charges 1500 W "
      "into the quarter the guard was trying to shave",
      r_mis.target_w == -1500.0 and resolve(SITE_SIDE).target_w == 1500.0)
check("a UNIT source filed on the site at a site that HAS named units is "
      "caught too - that is the direction that produces `lo > hi`",
      scope_violations(SITE_SIDE + [
          Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0),
          Claim.limit("device", P_SAFETY, -1, 1, actuator="inverter-b")]) != [])
check("...and at N=1, where every source legitimately routes to the site, the "
      "same list is clean. The check grades ROUTING, not tidiness",
      scope_violations(SITE_SIDE + [
          Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0)]) == [])
check("an unclassified ninth source is a violation wherever it is filed",
      scope_violations([Claim.limit("reserve", P_LOOP, hi=0.0)]) != [])
check("and the eight, routed by actuator_for() at a two-inverter site, are "
      "clean - which is what says the router and the checker agree",
      scope_violations(
          [Claim.limit(s, P_SAFETY, -1, 1,
                       actuator=actuator_for(s, "inverter-a", TWO))
           for s in CLAIM_SCOPE]) == [])

# --- ALLOC-01: AND NOW SOMETHING HAPPENS ABOUT IT ---------------------------
# ⚠️ ARB-03 LEFT THIS DELIBERATELY INERT - `scope_violations()` logged and did
# not change the resolution, and nothing graded the consequence. ALLOC-01 had to
# decide, and decided the allocator CONSUMES it as a REFUSAL rather than as an
# input. The reasoning is at `allocator.plan()`'s REFUSAL 1 and the short form
# is: the mis-routed claim has already been partitioned by the time stage 3
# runs, so its effect is baked into S or into somebody's interval; "consuming"
# it as an allocation input would mean RE-ROUTING it, which needs a guess about
# which inverter it belonged to that nothing here can make. What is left is that
# S itself may be wrong, and dividing a number that may be wrong is worse than
# not dividing it.
print(f"    and stage 3's answer to it: {r_mis.allocation.explain()}")
check("⚠️ the mis-routing is no longer only a log line: stage 3 REFUSES to "
      "allocate on a claim list it has been told is mis-routed, so main.py "
      "fails toward inaction and shouts instead of dividing a number that may "
      "be wrong. That is the same answer clause 3 already gives a contradiction",
      r_mis.allocated is False and r_mis.shares == {}
      and "mis-routed" in r_mis.allocation.refused)
check("...and it is NOT consumed as an ENVELOPE, which is the tempting reading "
      "and the wrong one: a unit-scoped claim that landed on the site names no "
      "unit, so promoting it means either guessing one or applying one "
      "inverter's rating to a sibling that does not have it. Nothing here is "
      "promoted - the refusal carries no shares at all",
      r_mis.allocation.plans == () and r_mis.allocation.dropped == ())
check("...and a correctly routed cycle is unaffected, which is what says this "
      "grades ROUTING and not the presence of units: the same site claims with "
      "the floor where it belongs allocate normally",
      resolve(SITE_SIDE, units=TWO).allocated is True)


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