"""Runnable check for the site allocator. `python3 test_alloc.py` ALLOC-01. The site grid target is ALLOCATED across N inverters and NEVER broadcast, because the control primitive is meter emulation: the master reports a FABRICATED grid reading and the inverter regulates itself to it, so one number means "the battery moves X W" here and "the grid is at X W" there - an identity that holds ONLY at N=1. Hand both inverters the same number and each supplies it IN FULL. ⚠️ WHAT THIS FILE CAN AND CANNOT CLAIM, SAID BEFORE THE FIRST CHECK. The two-master rig lives in admin/goodwe (TEST-19, `sim/sitesim.py`), where `compute()` closes the loop against `plantsim.Plant` with the firmware's own encoder on the wire. CI checks admin/goodwe out at `.gitea/infra-pin` (CI-12), but this file does not import the rig. The split: * `tools/alloc_sim.py` drives the REAL rig, and its output is quoted in the pull request. It is where "no limit cycle" is a claim about the shipped control law against the modelled hardware. * THIS FILE grades the allocator's own arithmetic exhaustively - which needs no plant at all - and closes a loop around the REAL `compute()` against a plant IT MODELS AND SAYS IT MODELS (part 8). The modelled plant is the `lag_s=0` limit ARCH-01's evidence was computed in; it is not evidence about a GW5000S-BP. ⚠️ INVERTER B DOES NOT EXIST YET. Nothing here contains a measured parameter for it. Every two-unit case below is built from a rating passed IN, and the second unit's rating is either the first one's or an explicit hypothetical. ⚠️ 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 random import sys from math import floor, inf from app.allocator import ( DROP_CONTRADICTION, DROP_NO_INTEGER, DROP_STALE, REFUSE_EMPTY, REFUSE_SCOPE, REFUSE_SIGN, Allocation, UnitPlan, _clamp_int, plan, ) from app.arbiter import ( ACTUATOR_SITE, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, UnitBounds, Claim, allocate, resolve, ) from app.control import Tuning, compute fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def _raises_value(fn): """True if `fn()` raises ValueError - a check that something is REFUSED.""" try: fn() except ValueError: return True return False def B(lo=-inf, hi=inf, gate=None, pin=None, by=None, contra=False): """One unit's interval, the way ARB-03 stage 1 hands it over.""" return UnitBounds(lo=lo, hi=hi, gate=gate, pinned_w=pin, pinned_by=by, contradiction=contra) A, Bee = "inverter-a", "inverter-b" RATING = 5000.0 # GW5000S-BP, measured. Inverter B's is a HYPOTHESIS. TWO = {A: B(-RATING, RATING), Bee: B(-RATING, RATING)} # ============================================================================== print("1. AC 1: the shares sum to the site target, and it is INTEGER arithmetic") # ⚠️ THE INVARIANT IS NOT `sum == S`. It is `sum + unmet == S`, and the # difference is the whole of AC 4: an allocation the envelopes could not fully # absorb is still a SAFE allocation, and reporting the shortfall is better than # either refusing or pretending. `unmet_w` is 0 whenever the envelopes allow. _r = plan(1990.0, TWO) print(f" {_r.explain()}") check("two open envelopes, an even weighting: the site target is halved and " "the halves sum back to it", _r.shares == {A: 995, Bee: 995} and _r.unmet_w == 0) check("...and 995 is not 1990, which is the sentence ARCH-01 is about. A " "broadcast hands each unit the site target itself, each supplies it IN " "FULL, and the loop gain at the meter becomes N times the per-unit gain", all(v != _r.sum_w for v in _r.shares.values())) # THE LONG REPLAY. AC 1 says "verified over a long replay rather than a spot # check", and it says so because the failure mode is a BIAS - a fraction of a # watt per tick that only shows up in aggregate. A spot check at a round number # cannot see it; 250 000 randomised ticks can. # # ⚠️ AND THE GENERATOR IS THE OTHER HALF OF THE CHECK, WHICH THIS FILE GOT # WRONG ONCE AND IS THE REASON THIS PARAGRAPH EXISTS. The first version drew # every envelope as `B(-uniform(0, RATING), uniform(0, RATING))` - finite on # both ends and containing zero, EVERY TIME. The check it fed says "no share # ever leaves the envelope it was clamped into", which reads as a statement # about `UnitBounds`; the data could only ever exercise a symmetric closed # interval around zero. So it reported 0 escapes over 250 000 ticks while a # reviewer's own 120 000 - drawn over the whole shape space - reported 6 254, # against a real clamp defect on half-open intervals. # # `UnitBounds` DEFAULTS TO `(-inf, +inf)`, so an envelope with one end missing # is the ordinary case at N>=2, not an edge. `_shape()` therefore draws from # every shape the type can express, and the counters below are printed so that # "the replay covers half-open intervals" is a number a reader can see rather # than a sentence this file asserts about itself. def _shape(rnd): """One random envelope, drawn across EVERY SHAPE `UnitBounds` CAN HOLD.""" kind = rnd.randrange(9) if kind <= 2: # finite, straddling zero return B(-rnd.uniform(0.0, RATING), rnd.uniform(0.0, RATING)) if kind == 3: # a FLOOR and no ceiling return B(rnd.uniform(-RATING, RATING), inf) if kind == 4: # a ceiling and no floor return B(-inf, rnd.uniform(-RATING, RATING)) if kind == 5: # finite, EXCLUDES zero, +ve _lo = rnd.uniform(1.0, RATING) return B(_lo, _lo + rnd.uniform(0.0, RATING)) if kind == 6: # finite, EXCLUDES zero, -ve _hi = -rnd.uniform(1.0, RATING) return B(_hi - rnd.uniform(0.0, RATING), _hi) if kind == 7: # half-open AND excludes zero return (B(rnd.uniform(1.0, RATING), inf) if rnd.random() < 0.5 else B(-inf, -rnd.uniform(1.0, RATING))) return B() # bare: (-inf, +inf) def _half_open(b): return (b.lo == -inf) != (b.hi == inf) def _excludes_zero(b): return b.lo > 0.0 or b.hi < 0.0 random.seed(20260902) _N = 250_000 _bad_sum = _bad_env = _refused = 0 _refused_ordinary = _ordinary = 0 _bad_record = 0 _quant = 0.0 _span = _n_open = _n_half = _n_nozero = _n_bare = 0 _esc_half = _esc_nozero = 0 for _i in range(_N): _S = random.uniform(-6000.0, 6000.0) _ba, _bb = _shape(random), _shape(random) _env = {A: _ba, Bee: _bb} for _b in (_ba, _bb): _n_half += _half_open(_b) _n_nozero += _excludes_zero(_b) _n_bare += (_b.lo == -inf and _b.hi == inf) _n_open += (_b.lo == -inf or _b.hi == inf) # AN ORDINARY TICK: no pin, and every envelope contains zero. That is the # population part 6 asks the sign gate never to fire on - and it is a # POPULATION now rather than the whole replay, because the widened shapes # include the forced floors the gate exists for. _ord = all(_b.lo <= 0.0 <= _b.hi for _b in (_ba, _bb)) _ordinary += _ord _wa = random.random() _p = plan(_S, _env, weights={A: _wa, Bee: 1.0 - _wa}) if not _p.ok: _refused += 1 _refused_ordinary += _ord continue if sum(_p.shares.values()) + _p.unmet_w != _p.sum_w: _bad_sum += 1 for _u, _s in _p.shares.items(): _b = _env[_u] # ⚠️ NO TOLERANCE. `_int_envelope()` narrows INWARD, so every integer # the allocator may command is inside the real interval - there is # nothing for a slack of a watt to absorb except a defect. if not (_b.lo <= _s <= _b.hi): _bad_env += 1 _esc_half += _half_open(_b) _esc_nozero += _excludes_zero(_b) # AND THE RECORD IS GRADED AGAINST THE SAME REPLAY, because the defect that # produced this generator also produced a `UnitPlan.explain()` line reading # `inverter-a -> +500 W in [2000, +inf]` - the record agreeing with itself # about a share the interval excludes. if not all(_pl.within_envelope for _pl in _p.plans): _bad_record += 1 _quant += _p.quantisation_w if _p.unmet_w == 0: _span += 1 print(f" replayed {_N} ticks at random targets, random envelopes and random " f"weights: {_bad_sum} sum failures, {_bad_env} envelope escapes, " f"{_refused} refusals") print(f" envelope shapes drawn ({2 * _N} in total): {_n_open} with an " f"infinite end, of which {_n_half} HALF-OPEN and {_n_bare} bare " f"UnitBounds(); {_n_nozero} that EXCLUDE zero") check("⚠️ OVER A QUARTER OF A MILLION TICKS THE SUM IS EXACT EVERY TIME. Not " "within a tolerance - EXACT - because `sum_w` is an int, the shares are " "ints and the leftover is settled in integer arithmetic, so there is no " "float in the identity for an error to accumulate in", _bad_sum == 0) check("...and no share ever leaves the envelope it was clamped into, over the " "same replay - INCLUDING the half-open ones and the ones that exclude " "zero, which is the coverage this check claimed before it had it. An " "allocator that meets its sum by over-driving one unit past its rating " "has met the wrong invariant", _bad_env == 0) check("...and the replay is NOT VACUOUS about that: it really did draw " f"{_n_half} half-open intervals, {_n_bare} bare `UnitBounds()` and " f"{_n_nozero} intervals excluding zero. Revert `_clamp_int` to the form " "that substitutes `v` for a missing bound and this check goes red on the " "half-open draws; under the old generator it could not", _n_half > 10_000 and _n_bare > 10_000 and _n_nozero > 10_000) check("...and every plan in the record AGREES with the share it carries: no " "`UnitPlan` was built asserting a number is inside an interval that " "excludes it. The record is what SAFETY-28 gates on, so a record that " "describes a share it did not check is worth less than no record", _bad_record == 0) check("⚠️ AND THE ONLY ROUNDING LEFT IS THE SITE-LEVEL ONE, WHICH DOES NOT " f"DRIFT. Over {_N - _refused} allocations the total of " f"`quantisation_w` - `sum_w - target_w`, at most half a watt each and " f"NEVER distributed into a per-unit bias - is {_quant:+.1f} W, i.e. " f"{_quant / max(1, _N - _refused):+.5f} W per tick. A 1 W systematic " "bias at 1-2 Hz is not a rounding artefact - it is a permanent standing " "error with the site integrator wound around it. (Largest remainder " "against the per-unit alternative is measured in part 2; the sum " f"identity above holds on all {_span} unconstrained ticks.)", abs(_quant) / max(1, _N - _refused) < 0.01) # ============================================================================== print() print("2. AC 1 continued: WHY largest remainder, measured against the " "alternative") # The obvious implementation is `round(w_i * S)` per unit. It is wrong, and the # amount it is wrong by is worth printing rather than asserting. def _naive(S, ws): """The obvious implementation: round each unit's own share, independently.""" return [round(w * S) for w in ws] random.seed(4242) _naive2 = _naive3 = _ours = 0 _worst2 = _worst3 = 0 _C = "inverter-c" _THREE = {A: B(-RATING, RATING), Bee: B(-RATING, RATING), _C: B(-RATING, RATING)} for _i in range(50_000): _S = random.randrange(-4000, 4000) _wa = random.random() _n2 = _naive(_S, [_wa, 1.0 - _wa]) _naive2 += sum(_n2) - _S _worst2 = max(_worst2, abs(sum(_n2) - _S)) _x, _y = sorted((random.random(), random.random())) _w3 = [_x, _y - _x, 1.0 - _y] _n3 = _naive(_S, _w3) _naive3 += sum(_n3) - _S _worst3 = max(_worst3, abs(sum(_n3) - _S)) _p3 = plan(float(_S), _THREE, weights=dict(zip([A, Bee, _C], _w3))) _ours += sum(_p3.shares.values()) - _p3.sum_w print(f" 50 000 ticks, per-unit round(): at N=2 it drifts {_naive2:+d} W in " f"total (worst tick {_worst2} W); at N=3, {_naive3:+d} W (worst tick " f"{_worst3} W). Largest remainder at N=3: {_ours:+d} W") check("⚠️ AT THREE INVERTERS THE NAIVE PER-UNIT round() MISSES THE SUM ON REAL " "INPUTS, repeatedly. This is what the card means by calling the rounding " "a safety property: every missed watt is a standing error at the meter " "that the site integrator chases and never reaches", _worst3 >= 1) check("...and largest remainder does not miss it, ever, over the same inputs " "at the same roster - which is also this file's only check that the " "allocator is general in N rather than special-cased at two", _ours == 0 and len(plan(1000.0, _THREE).shares) == 3) check("⚠️ AND AT TWO INVERTERS THE NAIVE VERSION IS ALMOST RIGHT, WHICH IS THE " "DANGEROUS HALF AND IS WHY IT IS MEASURED RATHER THAN ASSUMED. With two " "COMPLEMENTARY weights the two roundings cancel except on an exact " f"half-watt tie, so over 50 000 ticks it drifted {_naive2} W in total. An " "implementation that shipped at N=2 on per-unit rounding would look " "correct and start leaking the day a third inverter arrived - and the tie " "case below is wrong even at N=2", _worst2 <= 1) check("the two-unit case a per-unit round() gets wrong every time is the " "half-watt one: 1501 W split evenly is 750.5 twice, which rounds to " "either 1500 or 1502 - never 1501. Largest remainder gives 751 and 750, " "and gives the extra watt to a NAMED unit rather than to whichever one " "the tie-break happened to favour", plan(1501.0, TWO).shares == {A: 751, Bee: 750} and sum(plan(1501.0, TWO).shares.values()) == 1501) check("...and the tie-break is DETERMINISTIC, so the same inputs give the same " "allocation on every tick and on every host. A tie broken by dict order " "would move a watt between two inverters at 1-2 Hz forever", all(plan(1501.0, TWO).shares == {A: 751, Bee: 750} for _ in range(50))) # THE SITE-LEVEL ROUNDING, graded over a replay for the same reason the sum is: # the failure mode is a BIAS, and one spot check at one fractional target cannot # see one. `int()` instead of `round()` is the mutation that matters here - it # truncates toward zero, which is a systematic under-command of up to a watt on # EVERY tick, in a consistent direction, which is precisely the standing error # the card calls a safety property. random.seed(99) _q = [plan(random.uniform(-4000.0, 4000.0), TWO).quantisation_w for _ in range(20_000)] print(f" 20 000 fractional targets: |quantisation| <= {max(map(abs, _q)):.3f} W, " f"mean {sum(_q) / len(_q):+.5f} W") check("the site target itself is rounded ONCE, at the top, and the record says " "by how much. It is never more than half a watt - a TRUNCATION would be " "up to a whole one - and that half watt is a single site-level rounding " "the integrator corrects on the next tick, not a per-unit bias it cannot " "see", max(map(abs, _q)) <= 0.5 and plan(1500.4, TWO).sum_w == 1500) check("...and it is UNBIASED over the replay, which truncation is not: " "`int(x)` rounds toward zero, so it would under-command on every single " "tick in a consistent direction - a permanent standing error rather than " "a rounding that averages out", abs(sum(_q) / len(_q)) < 0.01) check("...and the allocation does not depend on the ORDER the envelopes arrive " "in, which is what makes it reproducible tick to tick: `plan()` sorts the " "roster before it does anything else, so a dict built the other way round " "gives the same shares", plan(1501.0, {Bee: B(-RATING, RATING), A: B(-RATING, RATING)}).shares == plan(1501.0, {A: B(-RATING, RATING), Bee: B(-RATING, RATING)}).shares == {A: 751, Bee: 750}) # ============================================================================== print() print("3. AC 4: saturation - the surplus moves, and what will not fit is NAMED") _sat = plan(4000.0, {A: B(-RATING, 1000.0), Bee: B(-RATING, RATING)}, weights={A: 0.5, Bee: 0.5}) print(f" {_sat.explain()}") check("one inverter at its rating gives its surplus to the other rather than " "being over-driven. An even weighting would have asked A for 2000 W; its " "envelope stops at 1000 W and B takes the other 3000", _sat.shares == {A: 1000, Bee: 3000} and _sat.unmet_w == 0) check("...and the record says WHY it got 1000 and not 2000, per unit, in words " "an operator can act on rather than as a silent clamp", any(p.unit == A and "envelope full" in p.why for p in _sat.plans)) _unmet = plan(9000.0, {A: B(-RATING, 1000.0), Bee: B(-RATING, 2000.0)}) print(f" {_unmet.explain()}") check("⚠️ AND WHEN NEITHER CAN TAKE IT, `unmet_w` IS REPORTED RATHER THAN THE " "SUM QUIETLY MISSING. 9000 W asked for, 3000 W of envelope: both units go " "to their ceilings and 6000 W is stated as unmet", _unmet.shares == {A: 1000, Bee: 2000} and _unmet.unmet_w == 6000) check("...and the allocation is still OK, which is the decision worth stating: " "a saturated site is not a broken allocation. Commanding 3000 W of the " "9000 the loop asked for is strictly better than commanding 0 W, and the " "shortfall is a fact the write gate can act on. Fusing `unmet_w` into " "`ok` would let one saturated inverter park the whole site", _unmet.ok is True) check("the identity that holds in every case, saturated or not: shares + unmet " "== the site target", sum(_unmet.shares.values()) + _unmet.unmet_w == _unmet.sum_w) # ============================================================================== print() print("4. AC 5: a pack at its SoC floor - 0 W discharging, STILL CHARGEABLE") # The floor arrives as a `hi` of 0 on that unit's interval: it may not # discharge. It says nothing about charging, and reading it as "this unit is # out" is the mistake - a unit at its floor is precisely the unit that most # wants the charge. _floor = {A: B(-RATING, 0.0), Bee: B(-RATING, RATING)} _dis = plan(1990.0, _floor) _chg = plan(-1990.0, _floor) print(f" discharging: {_dis.explain()}") print(f" charging: {_chg.explain()}") check("in the DISCHARGING direction the floored unit is allocated 0 W and the " "other carries the whole site target", _dis.shares == {A: 0, Bee: 1990} and _dis.unmet_w == 0) check("⚠️ AND IN THE CHARGING DIRECTION IT IS STILL AVAILABLE, which is the " "half a `if soc <= floor: skip this unit` implementation gets wrong. Its " "envelope forbids discharge, not participation, and a pack at its floor " "is the one that most needs the charge", _chg.shares == {A: -995, Bee: -995}) check("...and 0 W is not a sign of its own: the floored unit's 0 W sits happily " "beside the other's +1990 W without tripping the sign-coherence gate, " "because zero is coherent with every direction", _dis.ok is True and _dis.refused is None) # ============================================================================== print() print("5. AC 6: a dead link leaves the envelope set, and IS ANNOUNCED") _dead = plan(1990.0, TWO, live=(A,)) print(f" {_dead.explain()}") check("a unit whose master is stale is REMOVED from the envelope set. " "Allocating into a link nobody can reach is writing into a void and " "calling the site target met", Bee not in _dead.shares and _dead.dropped == ((Bee, DROP_STALE),)) check("...and the survivor carries the whole site target, which is right: there " "is one integrator on P1 and its output still has to be delivered", _dead.shares == {A: 1990} and _dead.unmet_w == 0) check("⚠️ AND IT DOES NOT ABSORB IT SILENTLY. The record NAMES the unit and the " "reason, so `main.py` can put it in the action log and SAFETY-28's write " "gate can refuse on it. A site running at half its inverters and " "reporting nothing reads as a healthy site on every dashboard", _dead.dropped[0][0] == Bee and _dead.dropped[0][1] == DROP_STALE and any(p.unit == Bee and "DROPPED" in p.why for p in _dead.plans)) check("...and the per-unit plan for the dropped unit is present rather than " "absent, so a gate iterating the roster finds an entry for every inverter " "instead of having to notice a missing key", {p.unit for p in _dead.plans} == {A, Bee}) check("a unit contradicting ITSELF - its own lo above its own hi - leaves the " "set for its own reason, and does not park the site. That is stage 1's " "decision honoured, not re-litigated here", plan(1990.0, {A: B(-RATING, RATING), Bee: B(500.0, -500.0, contra=True)}).dropped == ((Bee, DROP_CONTRADICTION),)) check("an interval that contains NO INTEGER is dropped too, and says which case " "it is. The register is 1 W: there is no number that unit could be " "commanded that both of its own bounds accept", plan(0.0, {A: B(-RATING, RATING), Bee: B(100.2, 100.8)}).dropped == ((Bee, DROP_NO_INTEGER),)) check("⚠️ AND EVERY UNIT GONE IS A REFUSAL, NOT AN EMPTY SUCCESS. With nobody " "in the envelope set there is nothing to allocate to, and inventing a " "share for a link nobody can reach is the failure the refusal branch " "exists for. THIS is the trigger `main.py`'s `if not resolution." "allocated:` was re-aimed onto when ALLOC-01 removed its old one", plan(1990.0, TWO, live=()).ok is False and REFUSE_EMPTY in plan(1990.0, TWO, live=()).refused) # ============================================================================== print() print("6. AC 7: opposite directions are REFUSED, not averaged") # ⚠️ WHY THIS IS A SAFETY GATE AND NOT AN EFFICIENCY ONE. One pack discharging # while another charges round-trips energy through two conversion stages, and it # is INVISIBLE: the meter sees the net, which is exactly S, so the site # integrator is content; each unit is meeting the target it was given; the sign # watchdog compares a command against the meter and both agree. Nothing in this # repository would ever report it. _circ = plan(1000.0, {A: B(-RATING, RATING, gate=P_MAINTENANCE, pin=2500.0, by="maintenance"), Bee: B(-RATING, RATING)}, site_priority=P_LOOP) print(f" {_circ.explain()}") check("a maintenance drain of 2500 W on one pack against a site target of " "1000 W would need the other pack to CHARGE at 1500 W to make the sum. " "That nets to exactly the right number at the meter, which is precisely " "why it has to be refused here - no downstream guard can see it", _circ.ok is False and REFUSE_SIGN in _circ.refused) check("...and the refusal NAMES both sides, so an operator is told which two " "inverters were about to push against each other", A in _circ.refused and Bee in _circ.refused) check("...and it is a REFUSAL rather than a quiet re-plan: dropping the pinned " "unit and re-filling would silently override a claim that outranked the " "loop, which is a worse answer than stopping and saying so", _circ.shares == {}) check("the same pin with a site target BEYOND it is fine, because the shares " "then agree in sign - the gate grades the OUTPUT, not the presence of a " "pin", plan(3000.0, {A: B(-RATING, RATING, gate=P_MAINTENANCE, pin=2500.0, by="maintenance"), Bee: B(-RATING, RATING)}, site_priority=P_LOOP).shares == {A: 2500, Bee: 500}) check("⚠️ AND IT NEVER FIRES ON AN ORDINARY TICK, which is what makes it a gate " "rather than a nuisance. Of the 250 000-tick replay in part 1 - random " f"targets, random envelopes, random weights, no pins - {_ordinary} ticks " "drew envelopes that all CONTAIN zero, half-open ones and bare " f"`UnitBounds()` among them, and it refused {_refused_ordinary} of " f"those. The other {_refused} refusals in that replay are this gate doing " "its job on the forced floors the widened generator now draws", _refused_ordinary == 0 and _ordinary > 50_000) check("...and it cannot: with no pin and an envelope that contains zero, the " "fill only ever moves shares in ONE direction away from zero, so two " "signs are not expressible. The gate exists for the envelope that " "EXCLUDES zero, which is a forced floor or a pin", all(v >= 0 for v in plan(1990.0, TWO).shares.values()) and all(v <= 0 for v in plan(-1990.0, TWO).shares.values())) # ⚠️ THE HALF-OPEN ENVELOPE, WHICH IS WHERE THIS GATE WAS ONCE SILENTLY # DISARMED - and the ticket's first named deliverable, "each share clipped to # that unit's UnitBounds interval", with it. `_clamp_int()` used to substitute # `v` for a missing bound and feed both ends into one `min(max(...))`; with # `hi == inf` that collapses to `min(max(v, lo), v) == v`, so THE FLOOR WAS # DISCARDED BY THE ABSENCE OF A CEILING. The worked case: inverter-a with a # 2000 W floor and no ceiling, inverter-b symmetric, a +1000 W site target. _HALF = {A: B(2000.0, inf), Bee: B(-5000.0, 5000.0)} _FIN = {A: B(2000.0, 9_000_000.0), Bee: B(-5000.0, 5000.0)} _half, _fin = plan(1000.0, _HALF), plan(1000.0, _FIN) print(f" floor 2000 W, NO ceiling: {_half.explain()}") print(f" the same with a finite ceiling: {_fin.explain()}") check("⚠️ A ONE-SIDED ENVELOPE REFUSES EXACTLY AS THE TWO-SIDED ONE DOES. Both " "put inverter-a at its 2000 W floor against a +1000 W site target, so " "inverter-b must charge for the shares to sum - circulating power, and " "the gate fires. The defect made the half-open case return a=+500, " "b=+500 instead: inverter-a commanded at a quarter of its own floor, and " "a refusal that should have fired suppressed with it", _half.ok is False and REFUSE_SIGN in _half.refused and _fin.ok is False and REFUSE_SIGN in _fin.refused) check("...and the two ends are applied INDEPENDENTLY, which is the shape of the " "fix rather than a patch of the one case above: a missing ceiling does " "not discard the floor, a missing floor does not discard the ceiling, and " "a bare `UnitBounds()` binds nothing", _clamp_int(0, 2142, inf) == 2142 and _clamp_int(9999, 2142, inf) == 9999 and _clamp_int(0, -inf, -500) == -500 and _clamp_int(-9999, -inf, -500) == -9999 and _clamp_int(-7, -inf, inf) == -7 and _clamp_int(0, 2142, 5000) == 2142) check("...and the RECORD can no longer assert a share is inside an interval " "that excludes it. The defect's log line read `inverter-a -> +500 W in " "[2000, +inf] (weighted share)` - the record naming the floor and the " "share beneath it in one breath and agreeing with itself. If that state " "is ever reachable again the line says OUTSIDE, and says it in the line " "an operator is already reading", UnitPlan(A, 500, 2000.0, inf, 0.5, "weighted share").within_envelope is False and "OUTSIDE" in UnitPlan(A, 500, 2000.0, inf, 0.5, "x").explain() and "DO NOT WRITE" in UnitPlan(A, 500, 2000.0, inf, 0.5, "x").explain() and "in [2000, +inf]" in UnitPlan(A, 2500, 2000.0, inf, 0.5, "weighted share").explain()) check("...and a DROPPED unit is not shouted about: it was commanded nothing, " "and its interval may be a contradiction that no number is inside. That " "case is already reported by name, and a violation banner on it would be " "the guard crying wolf on the one row that is working", "not commanded" in UnitPlan(A, 0, 3000.0, -3000.0, 0.0, f"DROPPED: {DROP_CONTRADICTION}", dropped=True).explain() and UnitPlan(A, 0, 3000.0, -3000.0, 0.0, "DROPPED: x", dropped=True).within_envelope is True) # ============================================================================== print() print("7. AC 8: at N<=1 it is a pass-through, byte-identical to ARB-03") check("with no units at all the whole target goes to the site actuator, as a " "float", allocate(1500.0, {}) == ({ACTUATOR_SITE: 1500.0}, True)) check("with one unit the share IS the site target - §6.2's N=1 identity: with " "one inverter the fabricated grid reading and the battery setpoint are " "the same number, so there is nothing to divide", allocate(1500.0, {A: UnitBounds()}) == ({A: 1500.0}, True)) check("⚠️ AND IT IS A FLOAT, UNROUNDED. The general path rounds to the 1 W " "register; the pass-through must not, because every N=1 replay in this " "repository - test_arbiter, test_watchdog, test_validate, test_peakguard " "- expects the number it saw before ALLOC-01. This is the line a 'tidy " "the special case away' refactor goes red on", allocate(1500.4, {A: UnitBounds()}) == ({A: 1500.4}, True) and isinstance(allocate(1500.4, {A: UnitBounds()})[0][A], float)) check("...and it is UNCLAMPED too, which is the other half of byte-identity: at " "N=1 every per-inverter claim routes to the SITE actuator anyway, so " "stage 2 has already applied the rating and re-applying an interval here " "would be a second, different clamp on a live safety path", allocate(9999.0, {A: B(-RATING, RATING)}) == ({A: 9999.0}, True)) _thru = plan(1500.0, {A: UnitBounds()}) check("...and the pass-through still produces a RECORD, so SAFETY-28's write " "gate reads the same shape at every roster size and does not need an N=1 " "special case of its own", _thru.ok and _thru.passthrough and _thru.plans and _thru.unmet_w == 0) # ============================================================================== print() print("8. AC 2 and AC 3, on a MODELLED plant: split-invariance and the sweep") # ⚠️ READ THE FILE HEADER BEFORE READING THESE NUMBERS. The plant here is # modelled IN THIS FILE - `lag_s=0`, unbounded ramp, which is the limit # ARCH-01's own evidence was computed in - and the loop is the REAL `compute()` # out of app/control.py. So the claim these checks support is about THE CONTROL # LAW AND THE ALLOCATOR, not about a GW5000S-BP. The hardware claim is # `tools/alloc_sim.py` against admin/goodwe's rig, quoted in the pull request. CYCLE_S = 5.0 def close_loop(gain, split, cycles=200, load_w=2000.0, mode="alloc", feedback="site", rating=RATING): """One integrator on the meter, N units, each told its OWN share. `feedback` is what `compute()` is handed as `actual_w`, and it is a PARAMETER because it turns out to be the whole of the AC 2 finding - see below. "site" is the sum of the units' power; "unit" is inverter A's alone, which is what the shipped add-on's single `inverter_power` sensor gives it. """ # ⚠️ `slew_w` IS PINNED HERE AND WAS NOT BEFORE, AND THAT IS SAFETY-39. # ARCH-01's 4000 W peak-to-peak broadcast limit cycle was measured at the # shipped pair of the day - `max_w: 2000` with `slew_w: 1000` - and this # function reached that 1000 through `Tuning`'s DEFAULT. SAFETY-39 made that # default 0, meaning "follow the rail" (`SLEW_RAIL_FRAC * max_w`, 222 W at # this rail), so the same call would now sweep a DIFFERENT law and report # ARCH-01's measurement as absent. A reproduction that silently re-tunes # itself when a default moves is not a reproduction. The number is written # here, next to the measurement it belongs to. tuning = Tuning(gain=gain, slew_w=1000.0) prev, i_w, sat = 0.0, None, 0 power = [0.0, 0.0] grid = load_w trace = [] for _c in range(cycles): d = compute(prev, grid, sum(power) if feedback == "site" else power[0], tuning, sat, i_w) prev, i_w, sat = d.target_w, d.i_w, d.sat_count if mode == "broadcast": cmds = [d.target_w, d.target_w] else: p = plan(d.target_w, {A: B(-rating, rating), Bee: B(-rating, rating)}, weights={A: split, Bee: 1.0 - split}) assert p.ok, p.refused cmds = [float(p.shares[A]), float(p.shares[Bee])] # The modelled unit: it obeys within one cycle, bounded by its rating. power = [max(-rating, min(rating, c)) for c in cmds] grid = load_w - sum(power) trace.append((round(d.target_w, 6), round(grid, 6))) return trace def p2p(trace, k=20): tail = [g for _t, g in trace[-k:]] return max(tail) - min(tail) _ref = close_loop(1.0, 0.5) _splits = [0.5, 0.9, 0.1, 0.99, 1.0, 0.0] _same = {s: close_loop(1.0, s) == _ref for s in _splits} print(f" aggregate trace vs the 50/50 reference, gain 1.0: " + ", ".join(f"{int(s*100)}/{int((1-s)*100)}={'same' if v else 'DIFFERS'}" for s, v in _same.items())) check("⚠️ AC 2. A 90/10 AND A 50/50 SPLIT PRODUCE AN IDENTICAL AGGREGATE TRACE " "AT THE METER - and so do 99/1 and 100/0. Not close: the same numbers. " "That is what says the weights are a COMFORT CHOICE and not a tuning " "parameter, and it is the reason §6.2 chose share allocation over a " "regulator/ballast arrangement, where the ballast unit's share is a " "function of the shared measurement and the gain question comes back", all(_same.values())) check("...and it is not vacuous, because the SHARES really are different. An " "allocator that ignored its weights would pass the line above trivially", close_loop(1.0, 0.9) == close_loop(1.0, 0.5) and plan(2000.0, TWO, weights={A: 0.9, Bee: 0.1}).shares != plan(2000.0, TWO, weights={A: 0.5, Bee: 0.5}).shares) check("...and it holds at EVERY gain the schema accepts, not only at the one " "the reference experiment uses", all(close_loop(g, 0.9) == close_loop(g, 0.5) for g in (0.05, 0.2, 0.4, 0.6, 0.8, 1.0))) # THE SWEEP. AC 3 says the sweep is the check, not an argument. _GAINS = (0.05, 0.2, 0.4, 0.6, 0.8, 0.9, 1.0) _alloc_worst = max(p2p(close_loop(g, s, feedback=f)) for g in _GAINS for s in _splits for f in ("site", "unit")) # ⚠️ THE BROADCAST RUNS ARE FED ONE INVERTER'S POWER, because that is what a # broadcast INSTALL would have: the add-on binds a single `inverter_power` # sensor, so an operator who wires up a second inverter has a loop that sees # unit A only. Feeding it the site sum here would model an add-on that does not # exist, and it would do so in the KIND direction - the sum tracks the command # much worse at N=2, so the saturation freeze would fire and damp the very limit # cycle this comparison exists to reproduce. sitesim.run_site() makes the same # choice for the same reason, and the two agree to the watt: 40 W p2p at gain # 0.8, 120 W at 0.9, 4000 W at 1.0. _bcast = {g: p2p(close_loop(g, 0.5, mode="broadcast", feedback="unit")) for g in _GAINS} print(f" broadcast tail p2p by gain: " + ", ".join(f"{g}={v:.0f}" for g, v in _bcast.items())) print(f" allocated, worst over {len(_GAINS)}x{len(_splits)}x2 " f"(gain x split x feedback): {_alloc_worst:.0f} W") check("⚠️ AC 3. SWEEPING THE GAIN TO 1.0 - WHICH config.yaml ACCEPTS AS ITS " "DOCUMENTED MAXIMUM - AT EVERY SPLIT SHOWS NO LIMIT CYCLE. The tail " "settles to a flat line at all 84 combinations - at BOTH the site-level " "feedback the loop should have and the single-sensor feedback it has", _alloc_worst == 0.0) check("⚠️ AND THE SWEEP IS NOT VACUOUS, because the SAME sweep reproduces the " "fault at the same gain. Broadcast at gain 1.0 is a permanent limit cycle " "- this is ARCH-01's measured 4000 W peak-to-peak, rebuilt here from the " "shipped control law rather than quoted", _bcast[1.0] >= 3900.0) check("...and the fault has an ONSET, which says it is a gain-margin failure " "and not a bug in the rig: quiet at 0.6, visible at 0.9, unbounded at " "1.0. An allocator that 'fixed' a cycle that was never there would show a " "flat broadcast row", _bcast[0.6] == 0.0 and 0.0 < _bcast[0.9] < _bcast[1.0]) check("...and the broadcast export overshoot at gain 0.6 - the half that has no " "gain threshold at all - is present here too, at ARCH-01's measured " "-400 W, while the allocated run does not export", min(g for _t, g in close_loop(0.6, 0.5, mode="broadcast", feedback="unit")[1:]) <= -400.0 and min(g for _t, g in close_loop(0.6, 0.5)[1:]) >= 0.0) # ⚠️ THE ONE THAT DID NOT HOLD, AND DEV-02 IS WHAT CLOSED IT. This block was # written as a DISCLOSURE - "the allocator is not the cause, and it is DEV-02's # to close" - and it is kept, with its stimulus intact, because the defect it # describes is exactly the thing a regression would restore. What has changed is # only which of the two rows the SHIPPED add-on is on. # # THE DEFECT. There was one `batt_entity`, so at a roster of two `actual_w` was # ONE unit's power used as a site measurement - the SENSING half of the same # N=1 identity ALLOC-01 removed from the COMMANDING half. `compute()`'s # anti-windup test is `abs(prev_w - actual_w) > saturation_w` with `prev_w` the # SITE target, so that comparison put a site number against a unit number, # tripped the saturation freeze permanently and left a standing grid error. # # MEASURED, on admin/goodwe's own two-master rig rather than only here # (`tools/alloc_sim.py --splits --feedback unit`, 2 kW load, 40 cycles): # # gain 0.2 worst |delta| between splits 510 W # gain 0.4 250 W # gain 0.6 50 W # gain 0.8 and above 0 W # # and with `--feedback site` - the counterfactual, which is what DEV-02 makes # the shipped sensing - every split is bit-identical at every one of those # gains. Both rows are still run below, because the comparison is the evidence. _unit_fb = {s: close_loop(0.6, s, feedback="unit") == close_loop(0.6, 0.5, feedback="unit") for s in _splits} print(f" with the loop fed ONE inverter's power - the PRE-DEV-02 sensing - " f"at gain 0.6: " + ", ".join(f"{int(s*100)}/{int((1-s)*100)}=" f"{'same' if v else 'DIFFERS'}" for s, v in _unit_fb.items())) _worst_unit = max( max(abs(a[1] - b[1]) for a, b in zip(close_loop(g, s, feedback="unit"), close_loop(g, 0.5, feedback="unit"))) for g in (0.2, 0.4, 0.6) for s in _splits) print(f" worst split-dependence under that sensing, gains 0.2-0.6: " f"{_worst_unit:.0f} W at the meter") check("⚠️ THE DEFECT IS STILL REPRODUCIBLE, WHICH IS WHAT MAKES THE FIX BELOW " "MEAN SOMETHING. Fed ONE inverter's power the aggregate trace at the " "meter is a function of the split, by hundreds of watts below gain 1.0. " "The allocator is not the cause - its shares sum to S in every one of " "these runs - and a check that had DELETED this row when the sensing was " "fixed would leave the next reader unable to tell a working loop from an " "untested one", not all(_unit_fb.values()) and _worst_unit > 100.0) check("⚠️ AND DEV-02 PUT THE SHIPPED ADD-ON ON THE OTHER ROW: `actual_w` is now " "the SITE SUM of the per-instance battery readings " "(`Controller.site_batt_w`, pinned at main.py's call site by " "test_inverter.py part 5), and every split is bit-identical again at " "every gain. This is the line that goes red if anyone hands the loop one " "unit's power again - which is the tempting shape, because it is what a " "single `batt_entity` gives you", all(close_loop(g, s, feedback="site") == close_loop(g, 0.5, feedback="site") for g in (0.05, 0.2, 0.4, 0.6, 0.8, 0.9, 1.0) for s in _splits)) check("...and the fix is not vacuous in the other direction either: the SHARES " "still differ between splits under site feedback, so 'bit-identical at " "the meter' is a statement about the aggregate and not about an allocator " "that stopped allocating", close_loop(0.6, 0.9, feedback="site") == close_loop(0.6, 0.5, feedback="site") and plan(2000.0, TWO, weights={A: 0.9, Bee: 0.1}).shares != plan(2000.0, TWO, weights={A: 0.5, Bee: 0.5}).shares) # ============================================================================== print() print("9. the structural claim itself: the gain at the meter is a CONVEX " "combination") # ⚠️ THIS IS THE ARGUMENT THE WHOLE CARD RESTS ON, SO IT IS MEASURED. Give the # two modelled units DIFFERENT per-unit gains - unit i delivers g_i of what it # is told - and read the gain the site actually shows. Under broadcast it is the # SUM. Under allocation it is a weighted average, so it lies between the two # whatever the weights are. def site_gain(g_a, g_b, split, mode): told = 1000.0 if mode == "broadcast": return (g_a * told + g_b * told) / told p = plan(told, TWO, weights={A: split, Bee: 1.0 - split}) return (g_a * p.shares[A] + g_b * p.shares[Bee]) / told _GA, _GB = 0.4, 1.0 _alloc_gains = [site_gain(_GA, _GB, s, "alloc") for s in (0.0, 0.25, 0.5, 0.75, 1.0)] print(f" per-unit gains {_GA} and {_GB}: allocated site gain over splits " f"0/25/50/75/100 = {[round(x, 3) for x in _alloc_gains]}; " f"broadcast = {site_gain(_GA, _GB, 0.5, 'broadcast'):.3f}") check("⚠️ FOR ANY SPLIT the gain at the meter lies between the two per-unit " "gains. THAT is why a bad split cannot destabilise the site and why the " "weights may be chosen for SoC, thermals and wear without anyone " "re-deriving a stability margin", all(_GA - 1e-9 <= g <= _GB + 1e-9 for g in _alloc_gains)) check("...and under broadcast it is their SUM, which is outside that interval " "and grows with N. At two units of gain 1.0 that is 2.0 against a loop " "whose config maximum is 1.0 - the 2.12 ARCH-01 measured, and the reason " "the limit cycle has no upper bound", site_gain(_GA, _GB, 0.5, "broadcast") > _GB) check("...and the same holds for two units with the SAME gain, which is the " "case that will actually ship: allocation leaves it at g, broadcast " "doubles it", site_gain(0.8, 0.8, 0.3, "alloc") == 0.8 and site_gain(0.8, 0.8, 0.3, "broadcast") == 1.6) # ============================================================================== print() print("10. the envelope is read PER TICK, and the weights are not a constraint") _moving = {A: B(-RATING, RATING), Bee: B(-RATING, RATING)} _first = plan(2000.0, _moving) _moving[Bee] = B(-RATING, 0.0) # B's pack hits its floor mid-run _second = plan(2000.0, _moving) print(f" same allocator object, envelope moved between ticks: " f"{_first.shares} then {_second.shares}") check("⚠️ THE SAME CALL WITH A MOVED ENVELOPE GIVES A DIFFERENT ANSWER, which " "is what says nothing was cached at construction. There is no allocator " "OBJECT to cache in - `plan()` is a pure function of this tick's " "intervals - and an allocator holding a construction-time envelope " "over-allocates into a unit that cannot take it", _first.shares == {A: 1000, Bee: 1000} and _second.shares == {A: 2000, Bee: 0}) check("...and the module holds no state at all, which is how that is enforced " "rather than promised: nothing in it is written outside a function", not [n for n, v in vars(sys.modules["app.allocator"]).items() if not n.startswith("_") and isinstance(v, (list, dict, set))]) _zero = plan(4000.0, {A: B(-RATING, 1000.0), Bee: B(-RATING, RATING)}, weights={A: 1.0, Bee: 0.0}) print(f" weight 0 on B, but A's envelope is full: {_zero.explain()}") check("a weight of 0 is honoured while the preferred unit has headroom", plan(1000.0, TWO, weights={A: 1.0, Bee: 0.0}).shares == {A: 1000, Bee: 0}) check("⚠️ BUT IT IS A PREFERENCE, NOT A PERMISSION. When the preferred unit " "fills up the rest goes to the zero-weight unit anyway, because the SUM " "is the safety property and stranding capacity behind a comfort choice " "would manufacture a standing error at the meter for the site integrator " "to chase. To make a unit unusable, close its envelope", _zero.shares == {A: 1000, Bee: 3000} and _zero.unmet_w == 0) check("...and the record says that is what happened, rather than leaving a " "reader to wonder why the weights were disregarded", any(p.unit == Bee and "last resort" in p.why for p in _zero.plans)) check("a negative weight is REFUSED rather than normalised away - it is a " "signed target wearing a comfort weight's clothes, and silently taking " "its magnitude would allocate against the direction the caller asked for", _raises_value(lambda: plan(1000.0, TWO, weights={A: -1.0, Bee: 1.0}))) # ============================================================================== print() print("11. the seam above: what resolve() hands stage 3, and the pin gate") _res = resolve([Claim.set("loop", P_LOOP, 900.0), Claim.limit("device", P_SAFETY, -2000, 2000, "rating", actuator=A), Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w", actuator=A)], units=(A, Bee)) print(f" {_res.explain()} -> {_res.allocation.explain()}") check("resolve() carries the full record through, so SAFETY-28 has one place " "to gate on and does not have to re-derive anything", isinstance(_res.allocation, Allocation) and _res.allocation.ok and _res.shares == _res.allocation.shares) check("...and the priority stage 3 weighs a unit's own SET against is the " "priority of the claim `explain()` NAMES, taken from stage 2 rather than " "recomputed - so the two cannot drift apart", _res.winner_priority == P_LOOP and _res.winner == "loop") # THE PIN GATE. A unit's own winning SET competes with the site's. _pin = {A: B(-RATING, RATING, gate=P_MAINTENANCE, pin=-1200.0, by="maintenance"), Bee: B(-RATING, RATING)} _low = plan(-3000.0, _pin, site_priority=P_LOOP) _high = plan(0.0, _pin, site_priority=P_SAFETY) print(f" maintenance pin vs the loop: {_low.explain()}") print(f" maintenance pin vs a safety stop: {_high.explain()}") check("a unit's own maintenance SET at P_MAINTENANCE outranks the grid-" "following loop, so that pack does what its own cycle asked and the OTHER " "unit carries the remainder of the site target. The shares still sum to S", _low.shares == {A: -1200, Bee: -1800} and sum(_low.shares.values()) == _low.sum_w) check("⚠️ AND A SITE SAFETY STOP OUTRANKS THE PIN, which is the case that has " "to be right: `safety` files a SITE set at P_SAFETY for 'stopped' and " "'inputs missing', and a maintenance drain that survived it would be an " "inverter this controller had stopped and then kept driving", _high.shares == {A: 0, Bee: 0}) check("...and with no site priority named at all, the unit's own claim stands - " "a caller that says nothing about the site has not said the site outranks " "anything, and honouring the unit's highest-priority claim is the only " "safe reading of silence", plan(-3000.0, _pin).shares == {A: -1200, Bee: -1800}) # ⚠️ THE RESIDUAL, STATED RATHER THAN GLOSSED. `pre_resolve()` filters a unit's # LIMITS by that unit's OWN gate, and clause 2 of the arbiter's rule says a # limit binds only at or above the WINNER's priority - which at N>=2 may be the # site's, chosen afterwards. Where the site winner outranks a unit's gate, a # unit limit whose priority falls between them is applied more tightly than # clause 2 strictly requires. Enumerated over what main.py can actually file: _UNIT_LIMITS = {"device": P_SAFETY, "supervised": P_SAFETY, "maintenance": P_MAINT_SHAPE} _UNIT_SETS = {"failsafe": P_SAFETY, "maintenance": P_MAINTENANCE} _SITE_SETS = {"loop": P_LOOP, "arbitrage": P_ARBITRAGE, "safety": P_SAFETY} _affected = [(ls, lp, us, up, ss, sp) for ls, lp in _UNIT_LIMITS.items() for us, up in list(_UNIT_SETS.items()) + [(None, None)] for ss, sp in _SITE_SETS.items() # included by pre_resolve (no gate, or the gate does not exclude it) if (up is None or lp >= up) # but clause 2 against the SITE winner would have excluded it and lp < sp] print(f" unit-limit x unit-set x site-set combinations main.py can file: " f"{len(_UNIT_LIMITS) * (len(_UNIT_SETS) + 1) * len(_SITE_SETS)}; " f"tighter than clause 2 strictly requires: {len(_affected)}") for _row in _affected: print(f" {_row[0]}@{_row[1]} under {_row[2]}@{_row[3]} " f"vs site {_row[4]}@{_row[5]}") check("⚠️ THE ONLY COMBINATIONS AFFECTED ARE THE MAINTENANCE CHARGE-ONLY LIMIT " "UNDER A SITE SAFETY STOP, and a safety stop resolves to 0 W - a " "charge-only `hi` of 0 does not move 0. So the tighter reading is not " "observable on anything main.py can file today. It is written down here " "because a ninth source could change that, and because erring TIGHT is " "the safe direction: a tighter envelope allocates less, never more", all(row[0] == "maintenance" and row[4] == "safety" for row in _affected)) # ============================================================================== print() print("12. mutation: what defeats these guards, and which check bites") MUT = [] def _largest_remainder_replaced_by_round(S, ws): return {u: round(w * S) for u, w in ws.items()} _m = _largest_remainder_replaced_by_round(1501.0, {A: 0.5, Bee: 0.5}) MUT.append(("largest remainder replaced by a per-unit round()", sum(_m.values()) != 1501, "part 2: 'largest remainder does not miss it, ever'")) # The envelope clamp dropped: shares placed by weight alone. _m2 = {A: 0.5 * 4000.0, Bee: 0.5 * 4000.0} MUT.append(("the envelope clamp dropped - shares placed by weight alone", _m2[A] > 1000.0 and plan(4000.0, {A: B(-RATING, 1000.0), Bee: B(-RATING, RATING)}).shares[A] == 1000, "part 3: 'one inverter at its rating gives its surplus to the other'")) # The liveness argument ignored - `live` never consulted. MUT.append(("`live` ignored, so a stale master keeps being allocated into", plan(1990.0, TWO, live=(A,)).shares != plan(1990.0, TWO).shares, "part 5: 'a unit whose master is stale is REMOVED'")) # The sign gate removed - the circulating case allocated instead of refused. MUT.append(("the sign-coherence gate removed", _circ.ok is False, "part 6: 'opposite directions are REFUSED, not averaged'")) # The N<=1 branch folded into the general path. MUT.append(("the N<=1 pass-through folded into the general path (it would " "round, and clamp)", allocate(1500.4, {A: UnitBounds()})[0][A] == 1500.4 and allocate(9999.0, {A: B(-RATING, RATING)})[0][A] == 9999.0, "part 7: 'a float, unrounded' and 'unclamped too'")) # The allocator broadcasts - the ARCH-01 fault, rebuilt. _bc = {u: 1990 for u in TWO} MUT.append(("the allocator replaced by a broadcast - ARCH-01's own fault", sum(_bc.values()) != 1990 and plan(1990.0, TWO).unmet_w == 0, "part 1: 'the halves sum back to it', and part 8's sweep")) # unmet_w folded into ok. MUT.append(("`unmet_w != 0` folded into `ok`, so a saturated inverter parks " "the site", _unmet.ok is True and _unmet.unmet_w == 6000, "part 3: 'a saturated site is not a broken allocation'")) # The pin gate inverted: the site winner never outranks a unit SET. MUT.append(("the pin gate inverted - a maintenance drain survives a site " "safety stop", _high.shares == {A: 0, Bee: 0}, "part 11: 'a site safety stop outranks the pin'")) # scope_errors ignored by stage 3. _mis = resolve([Claim.set("loop", P_LOOP, -1500.0), Claim.limit("peak", P_PEAK, lo=1500.0, reason="q", actuator=Bee)]) MUT.append(("`scope_errors` ignored by stage 3 - a mis-routed claim divided " "anyway", _mis.allocated is False and REFUSE_SCOPE in _mis.allocation.refused, "test_arbiter.py part 8: 'stage 3 REFUSES to allocate on a claim " "list it has been told is mis-routed'")) MUT.append(("the site target TRUNCATED instead of rounded - a bias toward zero " "on every tick", max(map(abs, _q)) <= 0.5, "part 2: '|quantisation| is never more than half a watt'")) # ⚠️ THE ONE THAT SHIPPED, AND THE ROW THIS TABLE IS MOST INDEBTED TO. This # mutation IS the code that was on `186e13c`, so it is not hypothetical: the # question it answers is not "would a check catch it" but "why did none". _collapsed = min(max(int(0), 2142), 0) # the old form, evaluated at hi=inf MUT.append(("the two ends of `_clamp_int` collapsed back into one min(max()), " "so a MISSING CEILING discards the FLOOR", _collapsed == 0 and _clamp_int(0, 2142, inf) == 2142 and _half.ok is False, "part 6: 'a one-sided envelope refuses exactly as the two-sided " "one does', and part 1's replay - 39 991 envelope escapes over the " "same 250 000 ticks. ⚠️ AND ZERO ESCAPES UNDER THE GENERATOR THIS " "FILE ORIGINALLY SHIPPED WITH, which drew every envelope finite on " "both ends and containing zero: the mutation was already live and " "the table would have called it caught. The fix is graded by the " "SHAPES the generator can express, not by the number of ticks")) # ⚠️ AND NOW THE ONES THAT DO NOT BITE. Three of them, found by actually running # the mutations rather than by reasoning about them, and each is reported with # WHY - because a mutation table that only lists its successes is the same # unverifiable claim as a green check that cannot fail. MUT.append(("the FLOOR replaced by a round() before the leftover pass", False, "NOTHING - and it should not, because THE SUM IS NOT PROTECTED BY " "THE FLOOR. It is protected by the integer leftover pass that " "follows: whatever per-unit integerisation happens first, `debt = " "remainder - sum(share)` is computed from it and handed out until " "it is zero. Rounding first produces a DIFFERENT but still exactly " "sum-invariant allocation. Largest remainder is what decides WHICH " "unit is credited the leftover watt - the one that was shorted " "most - which is a fairness property, not a safety one. Remove the " "leftover pass itself and part 1 goes red, which is the row above")) MUT.append(("the largest-remainder TIE-BREAK on the unit name dropped", False, "NOTHING - and by construction: `plan()` sorts the roster before it " "does anything else and Python's sort is stable, so the order is " "already by name when the tie-break is reached. The determinism is " "delivered upstream; the explicit key is belt and braces. Part 2's " "dict-order check is what actually grades the property")) # ⚠️ THE ONE THAT DOES NOT BITE, AND IT IS STATED RATHER THAN HIDDEN. MUT.append(("the WEIGHTS replaced by any other non-negative numbers", False, "NOTHING - AND BY DESIGN. That is AC 2: the aggregate trace at the " "meter is identical for every split, so no check here can tell one " "weighting from another by watching the site. The weights are " "graded by what they do to the SHARES (parts 2 and 10), never by " "what they do to the meter - and the fact that nothing at the " "meter can see them is the property, not a gap")) for _name, _bites, _where in MUT: print(f" {'caught' if _bites else 'NOT CAUGHT':>10} {_name}") print(f" {_where}") check("the mutations that CAN disarm these guards are each caught by a check " "above - a guard whose checks survive its own defeat is not a guard. " "Every row here was RUN through tools/mutation_run.sh on a purged tree, " "not reasoned about: three of them were reported as caught by an earlier " "draft of this table and were not", all(b for _n, b, _w in MUT[:-3])) check("...and the three that are NOT caught are reported as such, with the " "reason each one cannot be. Two are properties delivered somewhere other " "than the line the mutation touches; the third is the acceptance " "criterion itself - a weighting that changed the aggregate trace would be " "a FAILURE of AC 2, and part 8 is the line that would go red", all(b is False for _n, b, _w in MUT[-3:])) print() print("13. OPT-11: the mode layer - applied at or below 35, released above it") # A mode is a SOFT unit limit beside the hard ones (UnitBounds.mode_*). The # allocator applies it only when the site target was decided at or below the # mode's own priority (P_MODE, 35), drops it when its range leaves out 0 W, # and never lets it narrow a hard limit into nothing. from app.allocator import (MODE_FAULT, MODE_OUTRANKED, MODE_RELEASED, # noqa: E402 loop_bound, unit_interval) from app.arbiter import P_MODE # noqa: E402 def M(lo=-RATING, hi=RATING, mlo=-inf, mhi=inf, prio=P_MODE, **kw): """A unit's bounds with a mode range beside its hard limits.""" b = B(lo, hi, **kw) return UnitBounds(lo=b.lo, hi=b.hi, gate=b.gate, pinned_w=b.pinned_w, pinned_by=b.pinned_by, contradiction=b.contradiction, mode_lo=mlo, mode_hi=mhi, mode_priority=prio) _off = {A: M(), Bee: M(mlo=0.0, mhi=0.0)} _at35 = plan(3000, _off, deciding_priority=P_MODE) _at50 = plan(3000, _off, deciding_priority=P_PEAK) _at10 = plan(3000, _off, deciding_priority=P_LOOP) print(f" B off, 3000 W: decided at 10 {_at10.explain()} | at 35 " f"{_at35.explain()} | at 50 {_at50.explain()}") check("⭐ RELEASE AT 35: a target decided at P_MODE itself keeps the mode - " "B (Off) is held at 0 W and A carries all 3000 W", _at35.shares == {A: 3000, Bee: 0} and _at35.unmet_w == 0) check("...and below it too (the loop, 10): the same split", _at10.shares == _at35.shares) check("⭐ RELEASE AT 50: a target decided by the peak floor releases the mode " "and B takes its weighted half", _at50.shares == {A: 1500, Bee: 1500} and _at50.unmet_w == 0) check("...and the boundary is exactly 35: at 36 the mode is released", plan(3000, _off, deciding_priority=P_MODE + 1).shares == {A: 1500, Bee: 1500}) check("...and no deciding priority named means the mode applies (the safe " "reading for a direct caller)", plan(3000, _off).shares == {A: 3000, Bee: 0}) check("unit_interval() reports why: None when applied, MODE_RELEASED above " "35, and the hard limits always", unit_interval(_off[Bee], P_MODE) == (0.0, 0.0, None) and unit_interval(_off[Bee], P_PEAK) == (-RATING, RATING, MODE_RELEASED)) # CONTAINS 0: a mode that would FORCE a direction is a software fault. _force = {A: M(mlo=500.0, mhi=RATING), Bee: M()} _f = plan(-1000, _force, deciding_priority=P_LOOP) print(f" A's mode is [500, 5000] (leaves out 0 W), site charges 1000 W: " f"{_f.explain()}, faults {_f.mode_faults}") check("⭐ CONTAINS 0: a mode range that leaves out 0 W is dropped and named " "as a fault, so REFUSE_SIGN stays unreachable: both units charge and " "the split is allowed", _f.ok and _f.mode_faults == ((A, MODE_FAULT),) and all(v <= 0 for v in _f.shares.values()) and sum(_f.shares.values()) == -1000) _f_obeyed = plan(-1000, {A: B(500.0, RATING), Bee: B(-RATING, RATING)}) check("...and that is not a hypothetical: the same range obeyed as a hard " "limit forces A to give 500 W while B charges 1500 W, which the " "allocator REFUSES as circulating power", not _f_obeyed.ok and REFUSE_SIGN in _f_obeyed.refused) check("...and a mode whose own lo is above its own hi is the same fault", unit_interval(M(mlo=300.0, mhi=-300.0))[2] == MODE_FAULT) _out = unit_interval(M(lo=-2000.0, hi=-500.0, mlo=0.0, mhi=0.0)) check("a hard limit that leaves the mode no room wins outright " "(MODE_OUTRANKED): the unit keeps its hard interval", _out == (-2000.0, -500.0, MODE_OUTRANKED)) check("a unit with no mode (every unit with battery modes off) gets exactly " "its hard limits, and plan() is the same record as before the mode " "fields existed", unit_interval(B(-700.0, 900.0)) == (-700.0, 900.0, None) and plan(1234, TWO, deciding_priority=P_PEAK) == plan(1234, TWO)) # The sweep: sum and sign kept whatever the modes, the target and the # deciding priority; every share inside its effective interval. _rng = random.Random(1135) _bad, _nm = [], 0 for _ in range(20000): _units = {} for u in (A, Bee, "inverter-c")[:_rng.choice((2, 3))]: r = _rng.choice((2000.0, 3600.0, 5000.0)) mlo = _rng.choice((-r, -700.0, 0.0)) mhi = _rng.choice((0.0, 450.0, r)) _units[u] = M(-r, r, mlo, mhi) _d = _rng.choice((P_LOOP, P_ARBITRAGE, P_MODE, P_MODE + 1, P_PEAK, None)) _t = _rng.randint(-15000, 15000) _p = plan(_t, _units, deciding_priority=_d) _nm += 1 _eff = {u: unit_interval(b, _d)[:2] for u, b in _units.items()} _sh = _p.shares if (not _p.ok or sum(_sh.values()) + _p.unmet_w != _p.sum_w or (any(v > 0 for v in _sh.values()) and any(v < 0 for v in _sh.values())) or any(not (_eff[u][0] <= _sh[u] <= _eff[u][1]) for u in _sh) or (_d is not None and _d > P_MODE and any(_eff[u] != (b.lo, b.hi) for u, b in _units.items()))): _bad.append((_t, _d, _p.explain())) for x in _bad[:5]: print(f" BAD {x}") check(f"over {_nm} random ticks (2-3 units, mode ranges holding 0 W, targets " f"+-15 kW, deciding at 10/30/35/36/50/None) the shares sum to S with " f"`unmet`, never split the sign, and each stays inside its interval - " f"the mode's above 35 released, applied at or below", not _bad and _nm == 20000) # loop_bound(): what the loop may ask when it decides. _lb = loop_bound({A: M(mlo=-3000.0, mhi=0.0), Bee: M(mlo=0.0, mhi=0.0)}) check("loop_bound() sums each unit's interval with its mode applied: A may " "only charge 3000 W, B is off, so the loop may ask [-3000, 0]", _lb == (-3000.0, 0.0)) check("...a unit that is not live adds nothing", loop_bound({A: M(), Bee: M()}, live=(A,)) == (-RATING, RATING)) check("...a PINNED unit adds the span from 0 to its pin, so a 200 W probe " "(SAFETY-73: the closed limit becomes a pin) stays reachable", loop_bound({A: M(gate=P_SAFETY, pin=200.0), Bee: M(mlo=0.0, mhi=0.0)}) == (0.0, 200.0)) check("...and the pin is clamped into the hard limits first, as plan() " "clamps it", loop_bound({A: M(lo=-RATING, hi=0.0, gate=P_SAFETY, pin=200.0), Bee: M(mlo=0.0, mhi=0.0)}) == (0.0, 0.0)) check("...a mode that leaves out 0 W adds the hard interval (dropped, as in " "plan()), so every term holds 0 and so does the sum", loop_bound({A: M(mlo=500.0, mhi=900.0), Bee: M(mlo=0.0, mhi=0.0)}) == (-RATING, RATING)) check("...and a unit contradicting itself adds nothing", loop_bound({A: M(contra=True), Bee: M(mlo=0.0, mhi=0.0)}) == (0.0, 0.0)) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")