"""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. the seam is a seam, and NOT an allocator") 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", allocate(1500.0, {"inverter-a": UnitBounds()}) == ({"inverter-a": 1500.0}, True)) _shares, _ok = allocate(1500.0, {"inverter-a": UnitBounds(), "inverter-b": UnitBounds()}) print(f" two units, no allocator: shares={_shares} allocated={_ok}") check("with two it DECLINES rather than inventing a split. ⚠️ THE OBVIOUS " "PLACEHOLDER - hand the target to each - IS THE FAULT ARCH-01 WAS FILED " "ABOUT: each inverter supplies it in full, the gain at the meter becomes " "N times the per-unit gain, and no watchdog in the table fires", _shares == {} and _ok is False) check("...and specifically it does not broadcast", 1500.0 not in _shares.values()) 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]) == []) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")