"""Runnable check for the sign-disagreement watchdog. `python3 test_watchdog.py` Every case here is either a fault this must catch or a piece of NORMAL behaviour it must not mistake for one. The second list is the longer one on purpose: an alarm that fires on ordinary operation is worse than no alarm, because the next person switches it off. The headline validation is not here - it is `sim/replay_sign.py` in the infra repo, which drives this module over the two captured bus days. This file is the table-driven half: boundaries, timers, the latch, and the arming gate. ⚠️ 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 datetime import datetime, timedelta, timezone from app.watchdog import ( SignWatchdog, SignWatchdogConfig, loop_is_following, ) fails = [] T0 = datetime(2026, 8, 20, 15, 0, 0, tzinfo=timezone.utc) def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def drive(samples, cfg=None, following=True): """Feed (offset_s, commanded_w, grid_w) tuples. Returns (wd, trip_offset).""" wd = SignWatchdog(cfg) trip_at = None for item in samples: offset, cmd, grid = item[0], item[1], item[2] f = item[3] if len(item) > 3 else following if wd.evaluate(cmd, grid, f, T0 + timedelta(seconds=offset)) and trip_at is None: trip_at = offset return wd, trip_at def opposed(duration_s, step=5, cmd=-1000, grid=1500, following=True, start=0): return [(start + t, cmd, grid, following) for t in range(0, int(duration_s) + 1, step)] print("the fault it exists to catch") wd, at = drive(opposed(300)) check("charging into an importing house trips", wd.tripped) check("and trips at 180 s, not before", at == 180) check("the reason names both signals", "-1000" in wd.reason and "+1500" in wd.reason) print(f" e.g. {wd.reason!r}") wd, at = drive(opposed(300, cmd=2000, grid=-2000)) check("discharging into an exporting house trips too", wd.tripped and at == 180) print() print("the boundaries") wd, _ = drive(opposed(600, cmd=-300, grid=1500)) check("300 W exactly does NOT qualify (strictly greater)", not wd.tripped) wd, _ = drive(opposed(600, cmd=-301, grid=1500)) check("301 W does", wd.tripped) wd, _ = drive(opposed(600, cmd=-1000, grid=300)) check("a 300 W meter does not qualify either", not wd.tripped) wd, _ = drive(opposed(175)) check("175 s of disagreement does not trip", not wd.tripped) wd, _ = drive(opposed(180)) check("180 s does (>= hold_s)", wd.tripped) print() print("continuity - the rule that produced zero false positives on real data") broken = opposed(175) + [(180, -1000, -1500)] + opposed(175, start=185) wd, _ = drive(broken) check("179 s, one agreeing sample, then 175 s more: no trip", not wd.tripped) check("and the window really was reset", wd.disagree_since is not None) long_enough = broken + opposed(200, start=365) wd, _ = drive(long_enough) check("but a full 180 s after the reset does trip", wd.tripped) print() print("things that are NOT faults") # The load-step transient. Measured on the healthy capture: longest 30 s. wd, _ = drive(opposed(30) + [(35 + t, 800, 900) for t in range(0, 300, 5)]) check("a 30 s load-step crossover does not trip", not wd.tripped) # SoC 10 %, Standby, declining a 1350 W discharge. The battery does nothing, so # the meter stays where it was and the command agrees with it. wd, _ = drive([(t, 1350, 1350) for t in range(0, 900, 5)]) check("a battery refusing a discharge command cannot trip it", not wd.tripped) # A frozen meter drives command and reading into agreement, not disagreement. wd, _ = drive([(t, 800, 800) for t in range(0, 900, 5)]) check("a frozen meter cannot trip it", not wd.tripped) # Maintenance grid charge: 2500 W of charge against an importing house, for # hours. Opposed, both huge - and entirely correct. wd, _ = drive(opposed(7200, cmd=-2500, grid=3000, following=False)) check("two hours of maintenance grid charge does not trip", not wd.tripped) check("and the timer never even started", wd.disagree_since is None) wd, _ = drive([(t, -1000, None) for t in range(0, 900, 5)]) check("a missing meter reading does not trip", not wd.tripped) wd, _ = drive(opposed(175) + [(180, -1000, None)] + opposed(600, start=185)) check("a missing reading mid-window RESETS it rather than accumulating", wd.tripped and wd.disagree_since is not None) wd, at = drive(opposed(175) + [(180, -1000, None)] + opposed(600, start=185)) check("...so the trip lands 180 s after the gap, not after 180 s total", at == 365) wd, _ = drive([(t, 0.0, 3000) for t in range(0, 900, 5)]) check("a 0 W command is not a wrong-direction command", not wd.tripped) print() print("the latch") wd, _ = drive(opposed(200)) check("tripped after the fault", wd.tripped) # The trip commands 0 W, which removes the command side of the comparison. for t in range(200, 900, 5): wd.evaluate(0.0, 3000, True, T0 + timedelta(seconds=t)) check("stays tripped once the command is zeroed", wd.tripped) for t in range(900, 1500, 5): wd.evaluate(1000, 1000, True, T0 + timedelta(seconds=t)) check("stays tripped even once everything looks healthy again", wd.tripped) wd.reset() check("reset() is the only way out", not wd.tripped and wd.disagree_since is None) check("and the reason is cleared with it", wd.reason == "") print() print("the thresholds are configuration, not constants") cfg = SignWatchdogConfig(min_w=800.0, hold_s=60.0) wd, at = drive(opposed(300, cmd=-1000, grid=1500), cfg=cfg) check("a 60 s hold trips at 60 s", wd.tripped and at == 60) wd, _ = drive(opposed(600, cmd=-700, grid=1500), cfg=cfg) check("an 800 W gate ignores a 700 W command", not wd.tripped) print() print("the configuration that would disarm it is refused") from app.validate import validate_startup_config # noqa: E402 base = {"setpoint_entity": "number.sp", "meter_source": "off", "meter_phases": 1, "meter_max_age_s": 30} p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 300}, -5000, 5000) check("300 W under a 2000 W ceiling is accepted", p == []) p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 2000}, -5000, 5000) check("a threshold AT max_w is refused", len(p) == 1 and "never fire" in p[0]) # ⚠️ maintenance_charge_w is named here only to keep this case about the SIGN # rule. MAINT-04 added a second rule against the same ceiling, and a 500 W # max_w is also below the 2000 W shipped maintenance charge - so without this # the dict is refused twice, for two correct but different reasons, and the # `len(p) == 1` below would be measuring the interaction rather than this rule. p = validate_startup_config({**base, "max_w": 500, "sign_disagree_w": 3000, "maintenance_charge_w": 500}, -5000, 5000) check("a threshold above max_w is refused", len(p) == 1 and "never fire" in p[0]) p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 1999}, -5000, 5000) check("one watt under the ceiling is reachable, so accepted", p == []) print() print("the arming gate, derived from the REAL arbiter") # ⚠️ THIS IS THE DECISION THE WHOLE TICKET TURNS ON, and until now it was # asserted in three comment blocks and verified by nothing: every other case in # this file passes `following` in as a literal. What has to hold is that the # expression in main.py - loop won AND the emitted value points the way the loop # asked AND is no larger - is never True for a resolution the loop did not # actually get. So run real claims through the real resolve() and derive it. from app.arbiter import ( # noqa: E402 P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, Claim, resolve, ) def following_from(claims, asked): """THE SHIPPED gate, not a copy of it. ⚠️ This used to transcribe main.py's expression with a comment saying "keep the two in step", which is exactly the arrangement that does not keep two things in step. Measured: editing the shipped gate alone to `winner in ("loop", "peak")` left this suite - including the P_PEAK row below, whose entire job is to catch that edit - completely green, while the watchdog went blind for the duration of every peak event. Import it or the P_PEAK assertion is decoration. """ r = resolve(claims) return r, loop_is_following(r, asked) MAXW = Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w") DEV = Claim.limit("device", P_SAFETY, -5000, 5000, "rating") # name, claims, what the loop asked, expected `following` TABLE = [ ("the loop alone, unopposed", [Claim.set("loop", P_LOOP, 900)], 900, True), ("the loop bound by max_w - reduced, not redirected", [Claim.set("loop", P_LOOP, 4000), MAXW], 4000, True), ("charging, bound by max_w on the other side", [Claim.set("loop", P_LOOP, -4000), MAXW], -4000, True), ("maintenance owns the setpoint outright", [Claim.set("loop", P_LOOP, 900), Claim.set("maintenance", P_MAINTENANCE, -2500, "charge"), MAXW], 900, False), ("maintenance charge-only clamps the loop to 0 W", [Claim.set("loop", P_LOOP, 900), Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only"), MAXW], 900, False), # ⚠️ THE CASE `winner == \"loop\"` ALONE WOULD HAVE GOT WRONG. 1500 W, not # 2500 W: at max_w 2000 a -2500 hi against a -2000 lo is an arbiter # contradiction, not a charge. That pair was the SHIPPED default until # MAINT-04 refused it and CFG-03 gave the default one source of truth; it is # kept below as a case rather than as a configuration. ("cheap-window charge REDIRECTS the loop while it still 'wins'", [Claim.set("loop", P_LOOP, 500), Claim.limit("maintenance", P_MAINT_SHAPE, hi=-1500, reason="cheap-window"), MAXW], 500, False), ("...and the contradiction when charge_w exceeds max_w is not following either", [Claim.set("loop", P_LOOP, 500), Claim.limit("maintenance", P_MAINT_SHAPE, hi=-2500, reason="cheap-window"), MAXW], 500, False), ("a lo limit RAISES the magnitude without flipping the sign", [Claim.set("loop", P_LOOP, 200), Claim.limit("floor", P_SAFETY, lo=800), MAXW], 200, False), ("a safety stop wins outright", [Claim.set("loop", P_LOOP, 900), Claim.set("safety", P_SAFETY, 0.0, "stopped"), MAXW], 900, False), ("no claims at all - the arbiter's own failsafe", [MAXW, DEV], None, False), ("the loop asked for zero", [Claim.set("loop", P_LOOP, 0.0), MAXW], 0.0, False), # The forward hazard, asserted as it stands TODAY rather than as it should # be. When peak shaving lands, these expectations are what have to change. ("peak shaving at P_PEAK disarms the gate (loop_is_following's forward hazard)", [Claim.set("loop", P_LOOP, 900), Claim.set("peak", P_PEAK, 1800, "shaving"), MAXW], 900, False), # ⚠️ AND THE ROW ABOVE CANNOT DETECT A WIDENED GATE ON ITS OWN, which was # found by trying it: with the shipped gate edited to # `winner in ("loop", "peak")` that case STILL comes out False, because peak # shaving asked for MORE than the loop and the magnitude term rejects it # anyway. The winner term is masked. This row isolates it - a peak claim # smaller than the loop's ask and pointing the same way, so `winner` is the # only term that can decide - and it is what actually turns red if the gate # is widened without anyone revisiting the hazard. ("...and with peak BELOW the loop's ask, only the winner term separates them", [Claim.set("loop", P_LOOP, 1800), Claim.set("peak", P_PEAK, 900, "shaving"), MAXW], 1800, False), ] for name, claims, asked, expect in TABLE: r, got = following_from(claims, asked) check(f"{name} -> following={expect}", got is expect) if got is not expect: print(f" resolution was {r.explain()!r}") # And the gate is never True when the loop did not win, whatever else is going on. check("no configuration above arms the gate without the loop winning", all(not following_from(c, a)[1] for _, c, a, _ in TABLE if following_from(c, a)[0].winner != "loop")) print() print("end to end: the trip is COMMAND ZERO, not alarm-only") # ⚠️ THE FAULT INJECTED HERE IS A NEGATIVE GAIN, and that is not a contrivance - # it is the minimal faithful model of what this watchdog is for. The check # compares the loop's OUTPUT DIRECTION against the INPUT STATE it was regulating, # so what it detects is a loop whose two ends have become decoupled in sign: a # gain, a claim or a strategy with its polarity inverted, or an integrator so far # wound up that unwinding takes minutes. (Illusmart's was the last one: it # reached -14 768 W against a +-5000 W rail while writing once every 20-60 s, and # that arithmetic alone is 5-15 minutes of wrong-direction command.) # # `gain` is float(0.05,1.0) in the schema, so this is not a reachable # configuration - it is a stand-in for a code regression, which is the case a # unit test can actually hold the line on. import asyncio # noqa: E402 from app.main import Controller # noqa: E402 # ⚠️ DEV-02. THE KEY SPELLING IS IMPORTED, NEVER TRANSCRIBED. `sign_tripped` and # `sign_reason` are now NAMESPACED PER INVERTER - flat, a trip caused by B # latched A across every restart and one operator press cleared both. A check # that spelled `"unit/inverter-a/sign_tripped"` by hand here would go on passing # against a main.py that had started writing somewhere else, which is this # repository's own named failure mode: the copy that drifts is always the # grading one. `unit_key()` is the one definition and `Inverter.key()` calls it. from app.inverter import UNIT_A # noqa: E402 from app.store import unit_key # noqa: E402 TRIPPED_KEY = unit_key(UNIT_A, "sign_tripped") REASON_KEY = unit_key(UNIT_A, "sign_reason") class _Hass: def __init__(self): self.writes = [] self.calls = [] # ⚠️ ADDED FOR SAFETY-03, AND THE REASON IS WORTH READING. The e2e below # runs a DELIBERATELY INVERTED loop (gain -0.6) into a steady 2000 W import, # which is a loop that winds straight into the rail - so the output-clamp # alarm fires here, and the stub not having `call` is how that was # discovered. This scenario is now a second, independent witness that the # clamp alarm sees a real winding loop; test_clamp.py asserts it explicitly. async def call(self, domain, service, data): self.calls.append((domain, service, data)) async def limits(self, entity): return (-5000.0, 5000.0, 1.0) async def number(self, entity, invert=False): # A house importing 2000 W, steadily, whatever the battery does. return {"sensor.grid": 2000.0, "sensor.soc": 50.0, "sensor.batt": 0.0}.get(entity) async def set_number(self, entity, value): self.writes.append(float(value)) return True class _S: data = {} def set(self, *a): pass def get_time(self, *a): return None class _P: enabled = False def publish(self, *a, **k): pass def close(self): pass async def _e2e(): hass = _Hass() opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 2000, "gain": -0.6, "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True} c = Controller(opts, hass, _S(), _P(), stop=asyncio.Event()) c.auto = True c.dev_min, c.dev_max = -5000.0, 5000.0 seen_wrong_way = False for _ in range(12): await c.read_inputs() await c.cycle() if c.target < -300 and not c.sign_wd.tripped: seen_wrong_way = True await asyncio.sleep(0.25) return c, hass, seen_wrong_way ctrl, hass, wrong_way = asyncio.run(_e2e()) check("the broken loop really did command charge into an import first", wrong_way) check("the watchdog tripped", ctrl.sign_wd.tripped) check("the commanded value is 0 W", ctrl.target == 0.0) check("0 W actually reached the device", 0.0 in hass.writes) check("and it got there through the failsafe claim, not by accident", "failsafe" in ctrl.reason) print(f" status line: {ctrl.reason!r}") check("the status page says so in words", any("SIGN-DISAGREEMENT FAILSAFE" in c["text"] for c in ctrl.checks())) check("the banner leads with it", "FAILSAFE" in ctrl.status_payload()["banner"]) check("it stays at 0 W on later cycles too", hass.writes[-1] == 0.0 and hass.writes.count(0.0) >= 1) # ⚠️ THIS USED TO BE `check("MQTT reports the failsafe state", ctrl.sign_wd.tripped)` # which restates the assertion three lines above and never touches the payload. # The `status: "failsafe"` value was shipped untested. Capture what publish() # actually hands the broker. published = {} ctrl.mqtt = type("P", (), {"enabled": True, "publish": lambda _s, v: published.update(v), "close": lambda _s: None})() ctrl.publish() check("MQTT's controller-status sensor reports 'failsafe'", published.get("status") == "failsafe") ctrl.sign_wd.reset() ctrl.publish() check("...and goes back to 'running' once cleared", published.get("status") == "running") print() print("the latch survives a restart - it is the only thing between a tripped") print("controller and the first thing an operator will try") # ⚠️ THE BLOCKER THIS FILE MISSED FIRST TIME ROUND. `auto` is persisted and # `tripped` was not, so reconstructing a Controller against the same Store - # which is exactly what a container restart does - came back armed, into the # fault, with no operator involved: tripped=False, banner NOT READY, and it # resumed commanding -1000 then -2000 W before re-tripping 180 s later. class _MemStore: """A Store that keeps its dict across "restarts", like the real file does. ⚠️ SEEDED FROM `store.DEFAULTS`, NOT FROM A LITERAL. `Store.load()` merges that dict over whatever is on disk, so a fake that hand-listed three keys answered `None` where the real store answers a default - and a check reading "this latch was never written" then could not tell an absent key from a False one. It is also a place to drift: SAFETY-31 and SAFETY-30 both added flat pairs, and neither would have appeared here. """ def __init__(self): from app.store import DEFAULTS as _SDEF self.data = dict(_SDEF) def set(self, k, v): self.data[k] = v def get_time(self, *a): return None async def _restart(): shared = _MemStore() hass1 = _Hass() opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 2000, "gain": -0.6, "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True} a = Controller(opts, hass1, shared, _P(), stop=asyncio.Event()) a.auto = True a.dev_min, a.dev_max = -5000.0, 5000.0 for _ in range(12): await a.read_inputs() await a.cycle() await asyncio.sleep(0.25) # The restart. Same options, same store, brand new Controller and Hass. hass2 = _Hass() b = Controller(opts, hass2, shared, _P(), stop=asyncio.Event()) b.auto = True b.dev_min, b.dev_max = -5000.0, 5000.0 for _ in range(6): await b.read_inputs() await b.cycle() await asyncio.sleep(0.05) return a, b, shared, hass2 pre, post, store, hass2 = asyncio.run(_restart()) check("it tripped before the restart", pre.sign_wd.tripped) check("the trip reached the store, UNDER THE TRIPPED INVERTER'S OWN KEY", store.data[TRIPPED_KEY] is True) check("...with the reason beside it, so the operator is told why", "SIGN_DISAGREEMENT" in store.data[REASON_KEY]) check("⚠️ ...AND THE FLAT PRE-DEV-02 KEY IS MIRRORED, AS THE ROLLBACK VIEW. " "This assertion used to read the other way - that the flat key was NOT " "written - on the argument that writing it would latch every other " "inverter on the site. That argument was right about the READ and wrong " "about the WRITE: nothing at runtime reads the flat key, and the one " "thing that does (`_grade_store_migration()`, on every start) ignores it " "whenever ANY namespaced latch is set, which a sibling's trip always " "leaves. What the absence actually cost was the ROLLBACK: store.py's " "rule 2 keeps this key so a downgrade finds the latch, and a trip taken " "after the migration wrote nothing here - so an operator who downgraded " "came up DRIVING. `test_inverter.py` part 3 is where 'B's trip does not " "latch A across a restart' is graded, on a real roster of two", store.data.get("sign_tripped") is True and "SIGN_DISAGREEMENT" in str(store.data.get("sign_reason", ""))) check("the new Controller comes back TRIPPED", post.sign_wd.tripped) check("...and says why, restored rather than re-detected", "SIGN_DISAGREEMENT" in post.sign_wd.reason) check("...commands 0 W from its very first cycle", post.target == 0.0) check("...and never commanded anything else after the restart", hass2.writes != [] and set(hass2.writes) == {0.0}) check("the banner still leads with the failsafe", "FAILSAFE" in post.status_payload()["banner"]) # ⚠️ TWO PRESSES, and the first one must NOT clear it. `auto` is True while # tripped, so the first press stops control; only the transition back to True # clears. That is the documented behaviour and it is worth an assertion, because # a one-press clear would be reachable by an operator who never read the banner. asyncio.run(post.handle_action("auto_toggle")) # -> stopped check("pressing stop does not clear the failsafe", post.sign_wd.tripped and store.data[TRIPPED_KEY] is True) asyncio.run(post.handle_action("auto_toggle")) # -> started check("pressing start DOES clear it, and clears the store with it - BOTH " "spellings, which is what stops the every-start latch guard re-arming a " "failsafe the operator has already dealt with", not post.sign_wd.tripped and store.data[TRIPPED_KEY] is False and store.data[REASON_KEY] == "" and store.data.get("sign_tripped") is False) # One more restart from the cleared store, to prove the clear persisted too. fresh = Controller({"setpoint_entity": "number.sp", "meter_source": "off"}, _Hass(), store, _P(), stop=asyncio.Event()) check("...so the next restart comes back armed and untripped, as it should", not fresh.sign_wd.tripped) # ⚠️ THE WRITE ORDERING IS AN INVARIANT, SO IT GETS AN ASSERTION. main.py writes # `sign_reason` and THEN `sign_tripped`, each one its own `save()`, so that a # process that dies between the two leaves a reason nothing reads - harmless - # rather than a latch with no explanation for the operator who has to decide # whether to clear it. Documented reasoning is not enforcement: swapping those two # lines used to pass every check in this file. # # The death is modelled as the second write raising. In the field it is a power # cut or a container kill (Store.save() swallows OSError rather than raising), so # what is asserted here is the PROPERTY - what survives on disk - not the # mechanism. class _CrashOnFlag(_MemStore): def set(self, k, v): if k == TRIPPED_KEY and v: raise RuntimeError("power cut, mid-trip") super().set(k, v) async def _crash_mid_write(): store = _CrashOnFlag() hass = _Hass() opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 2000, "gain": -0.6, "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True} c = Controller(opts, hass, store, _P(), stop=asyncio.Event()) c.auto = True c.dev_min, c.dev_max = -5000.0, 5000.0 died = False for _ in range(12): try: await c.read_inputs() await c.cycle() except RuntimeError: died = True break await asyncio.sleep(0.25) return store, died crashed_store, died = asyncio.run(_crash_mid_write()) check("the second write is the one that sets the flag", died) check("a crash between the two writes leaves the REASON on disk", "SIGN_DISAGREEMENT" in crashed_store.data.get(REASON_KEY, "")) check("...and never a latch with nothing to explain it", not crashed_store.data.get(TRIPPED_KEY)) # ============================================================================= print() print("=" * 72) print("SAFETY-31: circulating power, which nets to zero at the meter") print("=" * 72) # ⚠️ READ THIS BEFORE ADDING A CHECK HERE. Half of this card was already built # and it is NOT the half below. ALLOC-01 shipped the SIGN-COHERENCE GATE # (`allocator.REFUSE_SIGN`, graded by test_alloc.py part 6): an allocation whose # shares carry two signs is REFUSED, both units named, no shares returned. So # the allocator can no longer COMMAND circulating power, and nothing here # re-tests that. # # What is graded here is the MEASUREMENT half: circulating power the allocator # NEVER COMMANDED. The two are complements, not duplicates - the gate refuses an # allocation before it is written and cannot see one millisecond past that; this # reads the hardware and does not care what was written. from app.inverter import LINK_STALE_S # noqa: E402 from app.circulation import ( # noqa: E402 CIRC_OPTION_DEFAULTS, CirculationConfig, CirculationWatchdog, circulating_w, opposing, ) print() print("1. the sensor: sum(|Bi|) - |sum(Bi)| over MEASURED per-unit power") check("both units discharging: the triangle equality holds and it is ZERO. " "This is the ordinary case and the sensor must be silent in it", circulating_w([1000.0, 2000.0]) == 0.0) check("...both charging, likewise", circulating_w([-1000.0, -2000.0]) == 0.0) check("...and a unit resting at exactly zero beside a working one is not " "opposition", circulating_w([0.0, 2500.0]) == 0.0) # ⚠️ ASSERTED AGAINST A LITERAL, NOT AGAINST `2 * min(abs(a), abs(b))`. # MUTATION.md's rule 2: an assertion written in terms of the expression under # test cannot tell "the right formula" from "the same wrong formula twice". check("A charging 3000 W while B discharges 3000 W reads 6000 W - the card's " "own scenario, and the figure is 2*min(|A|,|B|), the power going round " "the loop through BOTH conversion stages", circulating_w([-3000.0, 3000.0]) == 6000.0) check("...unequal opposition is bounded by the smaller unit: -3000 against " "+1000 is 2000 W, not 4000", circulating_w([-3000.0, 1000.0]) == 2000.0) print(f" [-3000, +3000] -> {circulating_w([-3000.0, 3000.0]):.0f} W " f"[-3000, +1000] -> {circulating_w([-3000.0, 1000.0]):.0f} W " f"[+1000, +2000] -> {circulating_w([1000.0, 2000.0]):.0f} W") check("three units, one out of step: it is defined over the whole roster, not " "over a pair", circulating_w([-2000.0, 1000.0, 1000.0]) == 4000.0) print() print(" AC 5: at N=1 it is IDENTICALLY zero, for every reading") # ⚠️ SWEPT, NOT SAMPLED. "Identically zero" is a claim about EVERY value, and # one call with one number cannot distinguish it from "zero for this number". _n1 = [circulating_w([v]) for v in (0.0, 1.0, -1.0, 2500.0, -2500.0, 5000.0, -5000.0, 1e9, -1e9)] check("nine readings from -1e9 to +1e9 W, every one of them 0.0", set(_n1) == {0.0}) check("...and an EMPTY roster is zero too, not a crash - `Controller.publish()` " "calls this before anything has been read", circulating_w([]) == 0.0) check("⚠️ AND A MISSING READING AT N=1 IS STILL ZERO, WHICH IS THE OPPOSITE OF " "THE N>=2 RULE BELOW AND IS DELIBERATE. With one term the algebra has no " "room for an unknown: |B| - |B| is 0 for every B, INCLUDING one nobody " "read. Returning None here would park the shipped one-inverter install in " "a permanent 'cannot tell' on a quantity that cannot be anything else", circulating_w([None]) == 0.0) print() print(" a missing reading at N>=2 is UNKNOWN, and never zero") check("⚠️ None, NOT 0.0. Zero would be this sensor reporting HEALTH FROM AN " "ABSENCE - which is the 2026-09-02 failure itself, where a subsystem " "stopped answering and its silence read as calm for 6 h 25 m", circulating_w([None, 3000.0]) is None and circulating_w([-3000.0, None]) is None and circulating_w([None, None]) is None) print() print(" it is NON-NEGATIVE, including where floating point would say otherwise") _fp = [circulating_w([0.1, 0.2, 0.3]), circulating_w([1e-9, 2e-9]), circulating_w([-0.1, -0.2, -0.30000000000000004])] print(f" {_fp}") check("a same-sign roster never publishes a negative number. ⚠️ THIS GRADES " "THE PROPERTY, NOT THE `max(0.0, ...)` GUARD THAT DEFENDS IT, and the " "difference is stated so a mutation sweep does not report a gap that is " "not one: for a same-sign roster the two sums accumulate the same " "magnitudes in the same order and are bit-identical, so the guard is an " "EQUIVALENT MUTANT - searched over 700 000 random rosters without a " "single negative. circulation.py says so at the line", all(v >= 0.0 for v in _fp)) print() print("2. CFG-03: config.yaml is the source of truth and there is ONE copy") # ⚠️ READ OFF THE FILE, NOT TRANSCRIBED. The Dockerfile does not put config.yaml # in the image, so the running add-on cannot read it and `CIRC_OPTION_DEFAULTS` # has to exist - which makes it a place to drift. This is the same pin # test_maintenance.py puts on MAINT_OPTION_DEFAULTS, for the same reason. import re as _re # noqa: E402 from pathlib import Path as _Path # noqa: E402 _cfg_txt = (_Path(__file__).resolve().parent / "config.yaml").read_text(encoding="utf-8") # ⚠️ SPLIT ON THE TOP-LEVEL KEY, NOT ON THE WORD. "schema:" appears inside the # CFG-03 comment above these very options ("options:, schema:, ..."), and # splitting on the bare word cut the file in the middle of a comment - the # options block then contained no options, the regex found none, and this pin # compared an EMPTY dict against an empty expectation. That is MUTATION.md's # third failure mode exactly: the assertion ran, the sentence was true, and the # data could not reach the failure it names. Caught by the printed line. _cfg_opts, _cfg_schema = _cfg_txt.split("\nschema:", 1) _from_yaml = {k: int(v) for k, v in _re.findall(r"^ (circulation_\w+):\s*(\d+)\s*$", _cfg_opts, _re.M)} print(f" config.yaml options: {_from_yaml}") print(f" CIRC_OPTION_DEFAULTS: {CIRC_OPTION_DEFAULTS}") check("both options are declared in config.yaml's options: block and the " "in-code mirror agrees with it key for key and value for value", _from_yaml == CIRC_OPTION_DEFAULTS and len(_from_yaml) == 2) _bounds = dict(_re.findall(r"^ (circulation_\w+):\s*int\((\d+),(\d+)\)\s*$", _cfg_schema, _re.M) and [(m[0], (int(m[1]), int(m[2]))) for m in _re.findall(r"^ (circulation_\w+):\s*int\((\d+),(\d+)\)\s*$", _cfg_schema, _re.M)]) print(f" schema bounds: {_bounds}") check("...and BOTH have a schema range, so the Supervisor refuses an " "out-of-range value rather than the add-on discovering it at runtime", set(_bounds) == set(CIRC_OPTION_DEFAULTS)) check("⚠️ AND THE THRESHOLD'S FLOOR IS THE MEASURED KNEE, NOT A ROUND NUMBER. " "On the TEST-19 rig the worst CONTINUOUS opposition a site where NOTHING " "is faulty can produce is 5.0 s at 400 W, 10.0 s at 300 W, 40.0 s at " "250 W and 505.0 s at 100 W. Below the knee, ordinary operation holds " "this sensor over the threshold for MINUTES and no window shorter than " "the 900 s billing quarter could tell that from a fault - so a " "configurable value under 400 W would be a false-alarm generator, and " "the schema does not offer one", _bounds["circulation_w"][0] == 400) check("...and the window's floor is the firmware's own veto window, " "`inverter.LINK_STALE_S`, because below it a unit still executing a " "stale share is a LINK fault the add-on already handles by dropping it " "from the split - and its ceiling is the 900 s quarter that bills the " "capacity tariff, inside which this must be caught and acted on", _bounds["circulation_hold_s"] == (int(LINK_STALE_S), 900)) check("the shipped defaults sit inside their own schema ranges", all(_bounds[k][0] <= v <= _bounds[k][1] for k, v in CIRC_OPTION_DEFAULTS.items())) check("⚠️ AND NEITHER IS A ROUND NUMBER, WHICH IS AC 3 READ LITERALLY. 700 W " "is 1.75x the measured knee - the headroom is bought for inverter B's " "ramp and lag being PLACEHOLDERS until SPIKE-04 measures them - and 75 s " "is 15x the worst legitimate episode and 2.5x the firmware veto. 60, 100, " "120, 180, 300, 500 and 1000 are all values a guess would have landed on", CIRC_OPTION_DEFAULTS["circulation_w"] not in (100, 500, 1000, 2000) and CIRC_OPTION_DEFAULTS["circulation_hold_s"] not in (30, 60, 120, 180, 300)) print() print("3. the trip: sustained opposition fires, the slew transient does not") def circ_drive(series, cfg=None, step=5.0): """Feed a list of circulating figures, one per control cycle.""" wd = CirculationWatchdog(cfg or CirculationConfig()) at = None for i, v in enumerate(series): if wd.evaluate(v, T0 + timedelta(seconds=i * step)) and at is None: at = i * step return wd, at _CFG = CirculationConfig() _sustained, _at = circ_drive([2000.0] * 40) check("2000 W of measured opposition, held, trips", _sustained.tripped) check(f"...and at {_CFG.hold_s:.0f} s, not before - the window is a window, " "not a delay somebody rounded", _at == _CFG.hold_s) check("the reason gives the operator the figure, the peak and the duration, " "and says in words why nothing else reported it", "2000" in _sustained.reason and "75 s" in _sustained.reason and "nets to zero" in _sustained.reason.lower()) print(f" e.g. {_sustained.reason!r}") print() print(" ⚠️ AND NOT ON THE TRANSIENT, WHICH IS THE FALSE-ALARM CASE") # ⚠️ THE SHAPES BELOW ARE THE RIG'S, NOT INVENTED. Measured on TEST-19 over 924 # closed-loop runs in which every unit executed the share it was given: the # worst continuous opposition above 700 W is 5.0 s - ONE control cycle - at a # peak of 1700 W. That is the serialised write path: on a reversing cycle B's # new number lands up to a whole cycle after A's, both slew through zero at # 800 W/s behind a 10 s lag, and between the two crossings they genuinely # oppose. A guard that trips on that is a false-alarm generator, and this # project's doctrine on crying wolf is explicit (SAFETY-15, SAFETY-19). _one_cycle = [0.0] * 5 + [1700.0] + [0.0] * 30 _t1, _ = circ_drive(_one_cycle) check("the worst single transient the rig can produce - 1700 W for ONE control " "cycle - does not trip", not _t1.tripped) _repeated = ([0.0] * 3 + [1700.0]) * 20 _t2, _ = circ_drive(_repeated) check("...and neither does it REPEATED every fourth cycle for 400 s, which is " "a load flapping across zero all afternoon. The window clears between " "them, so nothing accumulates", not _t2.tripped) _almost = [1500.0] * 14 + [0.0] + [1500.0] * 14 _t3, _ = circ_drive(_almost) check("...and opposition that stands for 70 s, clears for one cycle, then " "stands again for 70 s does NOT trip: 140 s of trouble, but never 75 s " "CONTINUOUSLY, which is what the guard actually claims to measure", not _t3.tripped) _below = [_CFG.min_w] * 60 _t4, _ = circ_drive(_below) check(f"...and exactly {_CFG.min_w:.0f} W forever does not trip - the " "comparison is strictly greater, so the threshold is a threshold rather " "than an off-by-one", not _t4.tripped) _just_over, _at5 = circ_drive([_CFG.min_w + 1.0] * 40) check("...while one watt above it does, held long enough. The pair is what " "makes the boundary a measured line rather than a direction", _just_over.tripped and _at5 == _CFG.hold_s) print() print(" a MISSING reading clears the window rather than accumulating it") _gap, _ = circ_drive([2000.0] * 10 + [None] + [2000.0] * 10) check("⚠️ SAFETY-02 MAKES THE SAME DECISION FOR A STALE METER, AND FOR THE SAME " "REASON: a frozen figure held against a moving one manufactures a fault " "in about a window's worth of cycles. 50 s over, one unknown, 50 s over " "again is not 100 s of opposition", not _gap.tripped) print() print(" the latch: it does not re-evaluate, because the trip removes the evidence") _latched, _ = circ_drive([2000.0] * 40 + [0.0] * 40) check("still tripped after the opposition has gone - and the opposition ALWAYS " "goes, because the trip commands 0 W everywhere and two inverters at zero " "cannot oppose. A non-latching version would de-assert and start the " "whole thing again, three cycles at a time, forever", _latched.tripped) _r = CirculationWatchdog(_CFG, tripped=True, reason="restored from the store") check("...and one restored as tripped stays tripped without being re-driven", _r.evaluate(0.0, T0) and _r.reason == "restored from the store") _r.reset() check("...until it is reset, which is the operator's press and nothing else", not _r.tripped and _r.reason == "" and _r.held_s(T0) == 0.0) print() print(" `opposing()` names the two sides, for the sentence the operator reads") check("indices of who is discharging and who is charging", opposing([-3000.0, 3000.0]) == ([1], [0])) check("...and a unit idling below the floor is not named as a participant in " "somebody else's loop", opposing([-3000.0, 3000.0, 3.0], min_w=350.0) == ([1], [0])) print() print("4. END TO END, AND THIS IS AC 2 - the sensor is MEASURED, so it sees") print(" circulation the allocator NEVER COMMANDED") # ⚠️ THIS IS THE CHECK MUTATION.md's THIRD FAILURE MODE IS ABOUT, so it is # built the way that section says to build one. The claim is "X differs from Y" # - the MEASURED sensor differs from a sensor computed off the allocation # record - and X and Y are ACTUALLY MADE DIFFERENT here, with the difference # asserted as its own precondition and PRINTED, before anything is concluded # from it. DEV-02 shipped a check comparing a number against itself; ALLOC-01 # shipped a generator that found 0 escapes where a wider one found 39 991. from app.inverter import Inverter # noqa: E402 # ⚠️ INVERTER B DOES NOT PHYSICALLY EXIST - it is ~10 days out and SPIKE-04 # measures it. Its rating here is a HYPOTHESIS, named as one, exactly as # test_peakguard.py's helper names its own. B_RATING_HYPOTHESIS = 3000.0 class _CircHass(_Hass): """A house at rest, and two inverters MEASURED to be fighting each other. ⚠️ THE MEETER READS ZERO AND THAT IS THE WHOLE POINT. grid = load - sum(B), and with A at -3000 W and B at +3000 W the sum is zero, so P1 reads exactly what an idle site reads. Every existing guard in this add-on is a consumer of that number. """ def __init__(self, a_w=-3000.0, b_w=3000.0): super().__init__() self.a_w, self.b_w = a_w, b_w async def number(self, entity, invert=False): return {"sensor.grid": 0.0, "sensor.soc": 50.0, "sensor.soc_b": 50.0, "sensor.batt": self.a_w, "sensor.batt_b": self.b_w}.get(entity) def _two_inverter(hass, store, *, hold_s=1, min_w=700, **over): opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 5000, "gain": 0.6, "circulation_w": min_w, "circulation_hold_s": hold_s, "auto_start": True} opts.update(over) c = Controller(opts, hass, store, _P(), stop=asyncio.Event()) b = Inverter("inverter-b", setpoint_entity="number.sp_b", soc_entity="sensor.soc_b", batt_entity="sensor.batt_b", max_w=B_RATING_HYPOTHESIS, store=store) b.dev_min, b.dev_max = -B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS c.inverters.append(b) c.auto = True for inv in c.inverters: inv.dev_min = inv.dev_min if inv.dev_min is not None else -5000.0 inv.dev_max = inv.dev_max if inv.dev_max is not None else 5000.0 return c async def _circ_e2e(): hass = _CircHass() c = _two_inverter(hass, _MemStore()) seen = [] for _ in range(14): await c.read_inputs() await c.cycle() seen.append(dict(circ=c.site_circulating_w, grid=c.grid, target=c.target, alloc=c.allocation, tripped=c.circ_wd.tripped)) await asyncio.sleep(0.25) return c, hass, seen _c, _chass, _seen = asyncio.run(_circ_e2e()) _first = _seen[0] _alloc = _first["alloc"] _shares = dict(_alloc.shares) _meas = [inv.batt for inv in _c.inverters] print(f" MEASURED per-inverter power : {dict(zip(_c.units, _meas))}") print(f" ALLOCATION RECORD : ok={_alloc.ok} {_alloc.explain()}") print(f" meter (P1) : {_first['grid']:+.0f} W") print(f" sensor over MEASURED power : {circulating_w(_meas):.0f} W") print(f" sensor over the RECORD : " f"{circulating_w(list(_shares.values())):.0f} W") # --- THE PRECONDITION, AS ITS OWN CHECK ------------------------------------- # ⚠️ MUTATION.md RULE 1. If a later edit makes the record and the measurement # agree again, THIS goes red rather than the claim below going quietly vacuous. check("PRECONDITION: the allocation record and the measured power are " "ACTUALLY DIFFERENT in this fixture. The record is coherent - it is `ok`, " "its shares carry one sign, and they sum to the site target - while the " "hardware is measured at -3000 W and +3000 W. Without this line the two " "claims below could both be true of a fixture in which nothing differed", _alloc.ok is True and len({(s > 0) - (s < 0) for s in _shares.values() if s}) <= 1 and sorted(_meas) == [-3000.0, 3000.0] and set(_shares.values()) != set(_meas)) check("ALLOC-01's gate is SATISFIED here, and that is the point rather than an " "accident: the allocator refused nothing, because it commanded nothing " "wrong. `REFUSE_SIGN` cannot fire on a record whose shares agree in sign", _alloc.refused is None) check("⚠️ SO A SENSOR COMPUTED FROM THE ALLOCATION RECORD READS EXACTLY 0 W. " "That is the cheap version of this card, and it is worth nothing: " "ALLOC-01's sign-coherence gate already guarantees the record is " "sign-coherent, so a record-derived sensor is zero BY CONSTRUCTION and " "can only ever confirm that the allocator agrees with itself", circulating_w(list(_shares.values())) == 0.0) check("⚠️ AND THE SHIPPED SENSOR READS 6000 W ON THE SAME CYCLE, because it is " "derived from `Inverter.batt` - each unit's OWN measurement. This is the " "whole card: it detects circulation THE ALLOCATOR NEVER COMMANDED - a " "unit ignoring or lagging its share, a stale share still being executed, " "or HW-01's power stage running its last instruction for 6 h 25 m with " "the comms dead and the record impeccable throughout", _c.site_circulating_w == 6000.0) check("...and the METER READS ZERO through all of it, which is why no existing " "guard fires. SAFETY-02 gates on abs(grid_w) > 300 W and that gate never " "opens; the peak guard and the capacity tariff both bill grid exchange", _first["grid"] == 0.0 and not any(i.sign_wd.tripped for i in _c.inverters)) print() print(" ...and the trip ACTS: a site-wide stop, not a notification") check("the watchdog tripped", _c.circ_wd.tripped) check("the site is commanded 0 W", _c.target == 0.0) check("...and 0 W actually reached BOTH inverters, not just the one somebody " "decided was to blame", _chass.writes and _chass.writes[-1] == 0.0 and all(i.target == 0.0 for i in _c.inverters)) check("...and it got there through the `circulation` claim, by name, rather " "than by some other guard happening to stop the site", "circulation" in _c.reason) print(f" status line: {_c.reason!r}") check("the status page says so in words, and says the meter cannot see it", any("CIRCULATING POWER" in ck["text"] and "meter cannot see" in ck["text"] for ck in _c.checks())) check("the banner leads with it - above the sign failsafe, because a sign " "disagreement is visible on every dashboard in the house and this is " "visible nowhere at all", "CIRCULATING POWER" in _c.status_payload()["banner"]) _pub = {} _c.mqtt = type("P", (), {"enabled": True, "publish": lambda _s, v: _pub.update(v), "close": lambda _s: None})() _c.publish() print(f" published: circulation={_pub.get('circulation')} " f"grid={_pub.get('grid')} status={_pub.get('status')}") check("⚠️ AND IT IS PUBLISHED AS ITS OWN SENSOR BESIDE THE GRID FIGURE. " "`grid 0 W` and `circulating 6000 W` on the same dashboard is the whole " "phenomenon in two rows, and NO arithmetic on the grid figure alone can " "ever produce the second one", _pub.get("circulation") == 6000 and _pub.get("grid") == 0.0) check("...and the controller-status sensor reports the failsafe state", _pub.get("status") == "failsafe") # ⚠️ AND THE UNKNOWN CASE, WHICH A SWEEP FOUND UNGRADED. A per-unit reading that # did not arrive makes the site figure None, and `MqttPublisher.publish()` DROPS # None values - so the entity keeps its last figure and the dashboard shows a # stale number rather than a fresh zero. Publishing `round(x or 0.0)` instead # survived the first mutation battery: it turns "I do not know" into "0 W", i.e. # health manufactured from an absence, on the one surface an operator actually # watches. Same defect as the sensor's own None rule, one layer out. _c.inverters[1].batt = None _pub_unknown = {} _c.mqtt = type("P", (), {"enabled": True, "publish": lambda _s, v: _pub_unknown.update( {k: x for k, x in v.items() if x is not None}), "close": lambda _s: None})() _c.publish() print(f" with inverter-b's reading missing, publish() offers: " f"circulation={_pub_unknown.get('circulation', '')!r}") check("⚠️ WITH ONE UNIT'S READING MISSING THE SENSOR IS NOT PUBLISHED AT ALL, " "so the entity holds its last figure instead of being handed a 0 that " "would read as health. `publish()` drops None, and this is what makes " "that drop deliberate rather than incidental", "circulation" not in _pub_unknown) _c.inverters[1].batt = 3000.0 print() print("5. AC 5: at ONE inverter the sensor is identically zero and NOTHING changes") async def _n1(): hass = _CircHass(a_w=-3000.0, b_w=0.0) opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 5000, "gain": 0.6, "circulation_w": 700, "circulation_hold_s": 1, "auto_start": True} c = Controller(opts, hass, _MemStore(), _P(), stop=asyncio.Event()) c.auto = True c.dev_min, c.dev_max = -5000.0, 5000.0 for _ in range(14): await c.read_inputs() await c.cycle() await asyncio.sleep(0.2) return c _c1 = asyncio.run(_n1()) print(f" roster={_c1.units} measured batt={_c1.inverters[0].batt:+.0f} W " f"sensor={_c1.site_circulating_w}") check("the shipped roster of one, with the single inverter measured at " "-3000 W - a magnitude that would trip instantly at N=2 - reads exactly " "0.0 W", _c1.site_circulating_w == 0.0) check("...so the watchdog never trips, over 14 cycles at a 1 s window", not _c1.circ_wd.tripped) check("...no `circulation` claim is ever filed, so the arbitration is " "byte-identical to what shipped", "circulation" not in _c1.reason) check("...and the status page says n/a rather than claiming to watch something " "that cannot happen - an 'armed' row for an identically-zero quantity is " "a guard advertising work it is not doing", any("circulating power: n/a at one inverter" in ck["text"] for ck in _c1.checks())) _pub1 = {} _c1.mqtt = type("P", (), {"enabled": True, "publish": lambda _s, v: _pub1.update(v), "close": lambda _s: None})() _c1.publish() check("...and it publishes 0, not None: the value is KNOWN at N=1, and a " "dashboard row reading 0 W is the honest report of it", _pub1.get("circulation") == 0) print() print("6. SAFETY-34: the alarm survives a Home Assistant restart") # ⚠️ THE LATCH IS NOT THE ALARM, AND THE 2026-09-02 OUTAGE IS WHY THAT SENTENCE # IS HERE. `persistent_notification` is IN MEMORY: Home Assistant restarts and # the card is gone while the fault runs on. That outage went unseen for 6 h 25 m # with the detector correct throughout. So the durable half (the store) and the # visible half (the notification) have to be joined, and this grades the join. async def _circ_restart(): shared = _MemStore() h1 = _CircHass() a = _two_inverter(h1, shared) for _ in range(14): await a.read_inputs() await a.cycle() await asyncio.sleep(0.25) # THE RESTART. Same store, brand new Controller and Hass - which is exactly # what a Home Assistant restart does to this add-on. The house has stopped # circulating in the meantime, so nothing in the new process's own # measurements could re-derive the trip. h2 = _CircHass(a_w=0.0, b_w=0.0) b = _two_inverter(h2, shared) for _ in range(3): await b.read_inputs() await b.cycle() await asyncio.sleep(0.05) return a, b, shared, h1, h2 _a, _b, _shared, _h1, _h2 = asyncio.run(_circ_restart()) print(f" store after the trip: circ_tripped=" f"{_shared.data.get('circ_tripped')!r} " f"circ_reason={str(_shared.data.get('circ_reason'))[:48]!r}...") check("the trip is written to /data - the add-on's only durable volume - as a " "flag AND a reason", _shared.data.get("circ_tripped") is True and "CIRCULATING_POWER" in str(_shared.data.get("circ_reason"))) check("⚠️ THE REBUILT CONTROLLER COMES UP TRIPPED, and its own measurements say " "the site is quiet - 0 W on both units - so this can ONLY have come from " "the store. A latch that lives in memory comes back armed, into the " "fault, with no operator involved", _b.circ_wd.tripped and _b.site_circulating_w == 0.0) check("...carrying the reason with it, so the operator who has to decide " "whether to clear it can read WHY", "CIRCULATING_POWER" in _b.circ_wd.reason) check("...and it is still commanding 0 W on the far side of the restart", _b.target == 0.0 and "circulation" in _b.reason) _notes1 = [c for c in _h1.calls if c[0] == "persistent_notification"] _notes2 = [c for c in _h2.calls if c[0] == "persistent_notification"] print(f" notifications raised: {len(_notes1)} before the restart, " f"{len(_notes2)} after it") check("the alarm was raised in the first process", len(_notes1) >= 1) check("⚠️ AND RE-RAISED IN THE SECOND ONE. THIS IS THE CARD'S THIRD " "DELIVERABLE. `persistent_notification` does not survive a Home Assistant " "restart, so a latch alone would leave a stopped site with NOTHING in " "front of the operator - the detector correct, the state file correct, " "and the screen blank. That is how 2026-09-02 went unseen for 6 h 25 m", len(_notes2) >= 1) check("...under a fixed notification_id of its own, so a second episode " "REPLACES the first rather than stacking, and it is not confused with " "the frozen meter, the clamp or the saturated peak guard", all(c[2]["notification_id"] == "goodwe_circulating_power" for c in _notes1 + _notes2) and len({c[2]["notification_id"] for c in _notes1 + _notes2}) == 1) _msg = _notes2[0][2]["message"] check("...and the message tells the operator the three things only this alarm " "can: that the meter cannot see it, that control has been stopped, and " "that if the grid reading does not move afterwards one inverter is no " "longer listening and a human at the isolator is the last actuator", "METER CANNOT SEE" in _msg and "CONTROL HAS BEEN STOPPED" in _msg and "isolator" in _msg) check("...and it does not raise the same card on every cycle - a per-PROCESS " "flag, deliberately not persisted, is what makes 'has a human been shown " "this' a different question from 'is this site tripped'", len(_notes2) == 1) print() print(" the operator's press is the only way out, and it clears BOTH halves") async def _circ_clear(): shared = _MemStore() c = _two_inverter(_CircHass(), shared) for _ in range(14): await c.read_inputs() await c.cycle() await asyncio.sleep(0.25) was = c.circ_wd.tripped await c.handle_action("auto_toggle") # stop await c.handle_action("auto_toggle") # start - the deliberate press return c, shared, was _cc, _cstore, _was = asyncio.run(_circ_clear()) check("it really was tripped before the press", _was) check("a deliberate START clears the latch in memory and on disk together - " "the condition ALWAYS goes away on its own, because the trip commands " "0 W everywhere and two inverters at zero cannot oppose, so clearing on " "the condition would clear on nothing", not _cc.circ_wd.tripped and _cstore.data.get("circ_tripped") is False and _cstore.data.get("circ_reason") == "") check("⚠️ AND THE 'HAS A HUMAN SEEN THIS' FLAG IS CLEARED WITH IT. Left " "standing through a clear, the NEXT trip in this process would latch, " "stop the site, and raise no notification at all", _cc._circ_announced is False) print() print("=" * 72) print("SAFETY-30: at two inverters the sign check is TWO questions, not one") print("=" * 72) # ⚠️ READ THIS BEFORE ADDING A CHECK HERE. Half of this card was already built # and it is NOT the half below. DEV-02 shipped the PER-INSTANCE half: each # `Inverter` owns its own `SignWatchdog`, latched under its own namespaced store # keys, and a trip on B no longer latches A. test_inverter.py part 3 grades # that; nothing here re-tests it. # # What is graded here is THE COMPARISON ITSELF. Before this card every unit's # detector was handed `self.grid`. There is ONE P1, so at N>=2 that is a number # this controller put on ONE actuator, graded against a measurement that answers # for ALL of them - wrong in both directions: a FALSE trip when a sibling # accounts for the meter, and a MISSED one when a sibling MASKS a unit that # really is reversed. The fix is two comparisons, and they are two different # faults: each unit against ITS OWN measured power, and the SUM of the standing # commands against the meter. # # ⚠️ AND IT IS NOT SAFETY-31, WHICH IS THE SECTION ABOVE. `circulation.py` # compares the units AGAINST EACH OTHER and catches two inverters pushing # THROUGH each other, which nets to zero at the meter. This compares each unit # against ITS OWN COMMAND and catches a unit not doing what it was told. They # overlap in exactly one scenario - a unit reversing while its sibling pulls the # other way - and part 4 below asserts what each one concludes there, because # two guards that overlap without saying so is how a false-alarm generator gets # built. from app.control import Tuning as _Tuning # noqa: E402 from app.watchdog import ( # noqa: E402 AGAINST_METER, AGAINST_OWN_POWER, SIGN_OPTION_DEFAULTS, disagrees, ) print() print("1. ONE predicate, TWO second terms - and the trip sentence says which") # ⚠️ THE ARITHMETIC IS UNCHANGED AND THAT IS THE POINT: `disagrees()` was not # split in two. What moved is a PARAMETER NAME - `grid_w` became `against_w` - # because what the command is opposed TO is now the caller's decision. A second # copy of a safety predicate is one copy plus a future disagreement about what # the check means (see `loop_is_following()`), so there is still exactly one. _scfg = SignWatchdogConfig(min_w=300.0, hold_s=180.0) _ucfg = SignWatchdogConfig( min_w=float(SIGN_OPTION_DEFAULTS["unit_sign_disagree_w"]), hold_s=float(SIGN_OPTION_DEFAULTS["unit_sign_disagree_s"])) check("the same call that means 'commanded -1000 W into a +1500 W import' " "means 'commanded -1000 W while measuring +1500 W of discharge' when the " "caller hands it a unit's own power - one function, two questions", disagrees(-1000.0, 1500.0, True, _scfg) is True) check("...and every gate still binds on the second term whatever it is: a " "250 W reading does not qualify against a 300 W gate", disagrees(-1000.0, 250.0, True, _scfg) is False) check("...nor does a unit sitting at zero, which is a REFUSED command and not " "a reversed one - the distinction the module docstring makes about a " "battery declining a discharge, now read of the unit's own sensor", disagrees(1500.0, 0.0, True, _ucfg) is False) _wd_site = SignWatchdog(_scfg) _wd_unit = SignWatchdog(_ucfg, against=AGAINST_OWN_POWER) for _t in range(0, 400, 5): _wd_site.evaluate(-1000.0, 1500.0, True, T0 + timedelta(seconds=_t)) _wd_unit.evaluate(-1000.0, 1500.0, True, T0 + timedelta(seconds=_t)) print(f" site: {_wd_site.reason!r}") print(f" unit: {_wd_unit.reason!r}") check("⚠️ THE SITE INSTANCE'S SENTENCE IS BYTE-IDENTICAL TO THE ONE THAT " "SHIPPED. It is what an operator reads on the status page, what goes in " "the action log and what is persisted to /data/state.json, so it is " "asserted against the literal rather than against a format string", _wd_site.reason == ( "SIGN_DISAGREEMENT: commanded -1000 W against a meter reading " "+1500 W for 180 s continuously (limits: both over 300 W, held over " "180 s)")) check("...and the per-unit instance names what it ACTUALLY compared instead. " "'against a meter reading' there would be a false statement about which " "instrument produced the evidence, on the one line somebody woken by " "this alarm reads first", AGAINST_OWN_POWER in _wd_unit.reason and AGAINST_METER not in _wd_unit.reason and "+1500 W" in _wd_unit.reason) # ⚠️ `arm()` IS CONFIGURATION AND MUST NOT BE STATE. `Controller.cycle()` calls # it on EVERY cycle, so a version that cleared `disagree_since` there would # restart the window every cycle, never accumulate a hold, and leave the check # permanently green with nothing saying so. Driven rather than argued: open a # window, re-arm mid-episode, and see the window still standing. _rearm = SignWatchdog(_ucfg, against=AGAINST_OWN_POWER) _rearm.evaluate(-1000.0, 1500.0, True, T0) _rearm.evaluate(-1000.0, 1500.0, True, T0 + timedelta(seconds=5)) _rearm.arm(_ucfg, AGAINST_OWN_POWER) check("re-arming mid-episode leaves the window exactly where it was", _rearm.disagree_since == T0) for _t in range(10, 200, 5): _rearm.arm(_ucfg, AGAINST_OWN_POWER) _rearm.evaluate(-1000.0, 1500.0, True, T0 + timedelta(seconds=_t)) check("...so a detector re-armed on every one of 38 cycles still trips at its " "85 s hold, which is what says `arm()` did not quietly disarm it", _rearm.tripped) print() print("2. CFG-03: config.yaml is the source of truth and there is ONE copy") _sign_from_yaml = {k: int(v) for k, v in _re.findall(r"^ ((?:unit|site)_sign_disagree_\w+):\s*(\d+)\s*$", _cfg_opts, _re.M)} print(f" config.yaml options: {_sign_from_yaml}") print(f" SIGN_OPTION_DEFAULTS: {SIGN_OPTION_DEFAULTS}") check("all four new options are declared in config.yaml's options: block and " "the in-code mirror agrees key for key and value for value - the " "Dockerfile does not put config.yaml in the image, so the running add-on " "cannot read it and this is where the two can drift", _sign_from_yaml == SIGN_OPTION_DEFAULTS and len(_sign_from_yaml) == 4) _sign_bounds = {m[0]: (int(m[1]), int(m[2])) for m in _re.findall(r"^ ((?:unit|site)_sign_disagree_\w+):" r"\s*int\((\d+),(\d+)\)\s*$", _cfg_schema, _re.M)} print(f" schema bounds: {_sign_bounds}") check("...and ALL FOUR have a schema range, so the Supervisor refuses an " "out-of-range value rather than the add-on discovering it at runtime", set(_sign_bounds) == set(SIGN_OPTION_DEFAULTS)) check("the shipped defaults sit inside their own schema ranges", all(_sign_bounds[k][0] <= v <= _sign_bounds[k][1] for k, v in SIGN_OPTION_DEFAULTS.items())) check("⚠️ THE THRESHOLD IS `saturation_w / 2`, WHICH IS WHERE THE CONTROL LAW " "ITSELF ALREADY DRAWS THE LINE. This check gates on BOTH sides, and the " "two are OPPOSED, so the smallest command-versus-power gap it can ever " "see is 2 x min_w. At 250 W that is exactly the 500 W at which " "`compute()`'s own anti-windup test concludes the inverter is not " "following and freezes the integrator - so this watchdog cannot fire on " "a divergence the shipped loop calls ordinary tracking, and the two " "guards meet at a number instead of overlapping in a band nobody drew", SIGN_OPTION_DEFAULTS["unit_sign_disagree_w"] * 2 == _Tuning().saturation_w) check("...and the window clears its three bounds: above the firmware's own " "30 s link veto (`inverter.LINK_STALE_S` - inside it, a unit not " "executing its share is a LINK fault the add-on already answers by " "dropping it from the split), and UNDER HALF the 180 s the site check " "takes, because the per-unit conclusion needs nothing to net out before " "it is complete", SIGN_OPTION_DEFAULTS["unit_sign_disagree_s"] > LINK_STALE_S and SIGN_OPTION_DEFAULTS["unit_sign_disagree_s"] < 180 / 2) check("⚠️ AND NEITHER IS A ROUND NUMBER. 250 W is read off `saturation_w`; " "85 s is 8.5x the worst legitimate opposition 1890 two-inverter runs " "produce (10.0 s) and 2.8x the firmware veto. 60, 75, 90, 100, 120, 180 " "and 300 are all values a guess would have landed on", SIGN_OPTION_DEFAULTS["unit_sign_disagree_s"] not in (30, 60, 75, 90, 100, 120, 180, 300)) # ⚠️ AND THE SITE PAIR IS TWO DECISIONS, NOT ONE, WHICH IS WHY BOTH ARE PINNED. # The THRESHOLD is carried from the shipped pair because its argument is about # the meter's noise floor and the deadband - a property of ONE P1 and of # `Tuning.deadband_w`, neither of which the roster touches. The WINDOW is NOT # carried, because the transient it rejects is the loop crossing zero across a # rail that IS the roster sum: measured at 110 s with one inverter and 215 s # with two, on the same stimuli, against a 180 s hold. That is a carried number # justified and a carried number refused, in the same pair. check("the SITE threshold is the shipped one, carried - the noise argument " "behind 300 W is about the meter and the deadband, and a second inverter " "changes neither", SIGN_OPTION_DEFAULTS["site_sign_disagree_w"] == 300) check("⚠️ ...AND THE SITE WINDOW IS LONGER THAN THE SHIPPED 180 s, WHICH IS THE " "MEASUREMENT REFUSING TO BE CARRIED. At two inverters the worst " "legitimate opposition the rig produces is 215.0 s - reachable at " "`gain: 0.05`, `slew_w: 50` and a 2.2 kW load reversal, all inside what " "config.yaml accepts - so 180 s would have shipped a false trip on the " "check with the best record in the safety table. 325 s is 1.51x that", SIGN_OPTION_DEFAULTS["site_sign_disagree_s"] > 215.0 and SIGN_OPTION_DEFAULTS["site_sign_disagree_s"] < 215.0 * 2) check("...and it is still inside the 900 s quarter that bills the capacity " "tariff, and inside every episode the 2026-08-20 fault day contains - " "its four clusters run 16, 30, 77 and 77 minutes", SIGN_OPTION_DEFAULTS["site_sign_disagree_s"] < 900 and SIGN_OPTION_DEFAULTS["site_sign_disagree_s"] < 16 * 60) check("...and it is deliberately not 300 or 360, the two round numbers next to " "it", SIGN_OPTION_DEFAULTS["site_sign_disagree_s"] not in (300, 360, 600, 900)) print() print("3. AC 1: ONE unit drives the wrong way, and ONLY that unit trips") # ⚠️ THE PLANT IS MODELLED AND SAYS SO. `_SignHass` is two inverters that obey # INSTANTLY - no ramp, no lag - and one meter that is `load - sum(measured)`. # That is deliberately not a claim about transients: the worst legitimate # opposition a real ramp and lag can produce is measured on admin/goodwe's # two-master rig, over 1890 closed-loop runs, and recorded in # DECISIONS/SAFETY-30-per-unit-sign.md. What this fixture is for is the # ATTRIBUTION - which detector concludes what, from which instrument - and for # that an instant plant is the honest model, because it removes the one # mechanism that could make a unit look reversed without being reversed. class _SignHass(_Hass): """Two inverters, one meter, and a chosen subset that does the OPPOSITE. `reversed_units` are roster indices whose measured power is the NEGATION of what this controller told them - HW-01's wedged power stage, an inverted driver, a unit executing a stale share backwards. Everything else obeys. """ def __init__(self, load_w=2000.0, reversed_units=()): super().__init__() self.load_w = float(load_w) self.reversed_units = set(reversed_units) self.ctrl = None def measured(self): if self.ctrl is None: return [0.0, 0.0] out = [] for i, inv in enumerate(self.ctrl.inverters): w = inv.last_write if inv.last_write is not None else inv.target w = float(w or 0.0) out.append(-w if i in self.reversed_units else w) return out async def number(self, entity, invert=False): m = self.measured() + [0.0, 0.0] return {"sensor.grid": self.load_w - sum(m[:2]), "sensor.soc": 50.0, "sensor.soc_b": 50.0, "sensor.batt": m[0], "sensor.batt_b": m[1]}.get(entity) def _sign_site(hass, store=None, **over): """A two-inverter Controller, with SAFETY-31 deliberately given a long window. ⚠️ THE LONG `circulation_hold_s` IS ISOLATION, NOT A CLAIM THAT SAFETY-31 IS SILENT HERE. A unit reversing while its sibling pulls the other way IS circulating power and SAFETY-31 sees it - part 4 asserts exactly that, and what each guard concludes about it. What must not happen is the SITE being stopped by circulation while this section is trying to grade a PER-UNIT attribution, because then every "only B is parked" check below would be measuring the wrong guard. """ opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 2000, "gain": 0.6, "sign_disagree_w": 300, "sign_disagree_s": 1, "unit_sign_disagree_w": 250, "unit_sign_disagree_s": 1, "site_sign_disagree_w": 300, "site_sign_disagree_s": 1, "circulation_hold_s": 3600, "auto_start": True} opts.update(over) c = Controller(opts, hass, store if store is not None else _MemStore(), _P(), stop=asyncio.Event()) b = Inverter("inverter-b", setpoint_entity="number.sp_b", soc_entity="sensor.soc_b", batt_entity="sensor.batt_b", max_w=B_RATING_HYPOTHESIS, store=c.store) b.dev_min, b.dev_max = -B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS c.inverters.append(b) c.auto = True for inv in c.inverters: if inv.dev_min is None: inv.dev_min, inv.dev_max = -5000.0, 5000.0 hass.ctrl = c return c async def _drive(c, n=14, dt=0.2, watch=None): seen = [] for _ in range(n): await c.read_inputs() await c.cycle() if watch is not None: seen.append(watch(c)) await asyncio.sleep(dt) return seen async def _one_reversed(): hass = _SignHass(load_w=2000.0, reversed_units={1}) c = _sign_site(hass) seen = await _drive(c, watch=lambda c: dict( grid=c.grid, stood=[i.last_write if i.last_write is not None else i.target for i in c.inverters], meas=[i.batt for i in c.inverters], a=c.inverters[0].sign_wd.tripped, b=c.inverters[1].sign_wd.tripped, site=c.site_sign_wd.tripped)) return c, hass, seen _c30, _h30, _s30 = asyncio.run(_one_reversed()) _a30, _b30 = _c30.inverters _mid = _s30[3] print(f" cycle 3: stood {[round(x) for x in _mid['stood']]} " f"measured {[round(x) for x in _mid['meas']]} " f"meter {_mid['grid']:+.0f} W") check("PRECONDITION: the stimulus is not vacuous - inverter B really is doing " "the OPPOSITE of what it was told, and inverter A really is obeying. " "Without this line every claim below could be true of a fixture in which " "nothing was reversed", _mid["stood"][1] * _mid["meas"][1] < 0 and abs(_mid["meas"][1]) > 250 and _mid["stood"][0] * _mid["meas"][0] > 0) check("⚠️ AC 1. THE REVERSED UNIT TRIPS AND ITS SIBLING DOES NOT. The evidence " "is B's own 0x0518 power against B's own share, so it is a fact about B " "and nothing about A - which is what makes parking one of them the right " "remedy rather than an attribution nobody measured", _b30.sign_wd.tripped and not _a30.sign_wd.tripped) check("...and the reason names the instrument, so the operator is sent to the " "right one", AGAINST_OWN_POWER in _b30.sign_wd.reason) print(f" {_b30.sign_wd.reason!r}") # ⚠️ READ OFF THE ALLOCATION RECORD, NOT OFF `Controller.reason`. At a roster # of two `reason` is the SITE resolution's line - "loop -> 2000 W" - because # B's failsafe is a UNIT claim resolved on B's own actuator and never competes # for the site winner. That is the DEV-02 split working, and a check that # grepped the site line for "failsafe" would go green only while the roster was # one. The per-unit record is where the pin is stated, and it names its source. _bplan = [pl for pl in (_c30.allocation.plans if _c30.allocation else ()) if pl.unit == "inverter-b"] print(f" allocation: {_c30.allocation.explain()}") print(f" inverter-b plan: {_bplan[0].why if _bplan else '(none)'}") check("...B is commanded 0 W by its OWN `failsafe` claim, named in the " "allocation record, rather than by some other guard happening to stop it", _b30.target == 0.0 and _bplan and "failsafe" in _bplan[0].why) check("...and A IS STILL COMMANDING. One inverter in trouble stops that " "inverter; the site goes on running on the one that works, which is what " "a UNIT-scoped claim buys and what a site-wide stop would have thrown " "away", _a30.target != 0.0) check("...the latch is under B's OWN namespaced key, and A's is untouched", _c30.store.data.get(unit_key("inverter-b", "sign_tripped")) is True and _c30.store.data.get(TRIPPED_KEY) is False) print() print("4. AC 2: and it is caught EVEN THOUGH THE METER SAYS EVERYTHING IS FINE") # ⚠️ THIS IS THE CHECK THE WHOLE CARD IS FOR, so it is built the way MUTATION.md # says to build one: the claim is "the new comparison sees what the old one # structurally cannot", and the OLD comparison is RUN, on the same samples, and # its blindness is asserted as its own precondition and PRINTED - not argued. _old_view = [disagrees(s["stood"][1], s["grid"], True, _scfg) for s in _s30] _site_view = [disagrees(sum(s["stood"]), s["grid"], True, _scfg) for s in _s30] print(f" the SHIPPED comparison (B's command vs the meter) was true on " f"{sum(_old_view)} of {len(_old_view)} cycles") print(f" the SITE comparison (sum of commands vs the meter) on " f"{sum(_site_view)} of {len(_site_view)}") check("⚠️ THE COMPARISON THAT SHIPPED IS FALSE ON EVERY SINGLE CYCLE HERE. B is " "commanded to discharge and the house is importing, so B's command and " "the meter AGREE all the way through - the meter cannot tell that the " "watts arriving are inverter A's while B does the opposite. This is not " "a threshold set too high: the summation destroys the evidence and no " "tuning recovers it", not any(_old_view)) check("...and the SITE comparison is blind for the same reason, which is why " "it is kept as a SEPARATE question rather than offered as the fix", not any(_site_view)) check("...while the shipped SITE watchdog stayed correctly silent, so nothing " "here is a site-level trip wearing a per-unit label", not _c30.site_sign_wd.tripped and _c30.store.data.get("site_sign_tripped") is not True) check("⚠️ AND SAFETY-31 SEES SOMETHING TOO, WHICH IS NOT A DUPLICATE AND IS " "SAID HERE RATHER THAN LEFT FOR SOMEBODY TO DISCOVER. A unit reversing " "while its sibling pulls the other way IS power going round a loop, so " "`circulating_w()` is over its threshold on this fixture - it is only " "the long window this fixture gives it that keeps it from stopping the " "site. The two guards CONCLUDE different things: circulating power is " "defined over the roster and names no offender, while this one names B " "and parks B. Neither is the other with a different threshold", max(circulating_w(s["meas"]) for s in _s30) > _c30.circ_wd.cfg.min_w and not _c30.circ_wd.tripped and _c30.site_circulating_w == 0.0) print(f" circulating power over the episode: " f"{[round(circulating_w(s['meas'])) for s in _s30]}") check("...and it falls back to ZERO the moment B is parked, which is the " "remedy working rather than the sensor going quiet: a unit held at 0 W " "cannot be half of a loop. It is asserted on the EPISODE rather than on " "the end state for exactly that reason", _s30[-1]["meas"][1] == 0.0) print() print("5. AC 3: the SITE question survives, separately, and catches whole-site") print(" reversal - which no per-unit check can see") # ⚠️ THE FAULT HERE IS THE ONE THE 2026-08-20 CAPTURE CONTAINS, and it is # INVISIBLE to every per-unit comparison by construction: both inverters are # doing EXACTLY what they were told. What is wrong is the number they were told. # A negative gain is the minimal faithful model of it - an inverted polarity, a # wound-up integrator, a strategy on the wrong side of the meter - and it is not # a reachable configuration (`gain` is float(0.05,1.0)), which is the point: it # stands in for a code regression, which is the case a check can hold the line on. async def _site_reversed(): hass = _SignHass(load_w=2000.0, reversed_units=()) c = _sign_site(hass, gain=-0.6) seen = await _drive(c, watch=lambda c: dict( grid=c.grid, site=c.site_sign_wd.tripped, stood=[i.last_write if i.last_write is not None else i.target for i in c.inverters], meas=[i.batt for i in c.inverters])) return c, hass, seen _c31, _h31, _s31 = asyncio.run(_site_reversed()) # ⚠️ THE LAST SAMPLE BEFORE THE TRIP, NOT A FIXED INDEX. After the trip every # command is 0 W by design, so a fixed index reads the REMEDY rather than the # fault and the precondition below would be asserting that a stopped site is # stopped. This is `Run.p2p_between`'s lesson from the rig, in a unit test: name # the window rather than inferring it. _pre31 = [x for x in _s31 if not x["site"]] _mid31 = _pre31[-1] print(f" last cycle before the trip: stood " f"{[round(x) for x in _mid31['stood']]} " f"measured {[round(x) for x in _mid31['meas']]} " f"meter {_mid31['grid']:+.0f} W") check("PRECONDITION: EVERY inverter is obeying - each unit's measured power " "agrees in sign with its own share - and the site is still charging into " "an importing house. Without this the site trip below could be a " "per-unit fault in disguise", all(s * m > 0 for s, m in zip(_mid31["stood"], _mid31["meas"])) and sum(_mid31["stood"]) * _mid31["grid"] < 0) check("⚠️ AC 3. THE SITE WATCHDOG TRIPS AND NOT ONE PER-UNIT WATCHDOG DOES. " "That asymmetry is the card: the fault is in the number the site was " "asking for, which no single inverter owns, and every unit is innocent " "of it", _c31.site_sign_wd.tripped and not any(i.sign_wd.tripped for i in _c31.inverters)) print(f" {_c31.site_sign_wd.reason!r}") check("...and it stops the WHOLE site, through the `sign_site` claim by name " "rather than by some other guard happening to stop it", _c31.target == 0.0 and "sign_site" in _c31.reason and all(i.target == 0.0 for i in _c31.inverters)) check("...the latch is FLAT - `site_sign_tripped`, beside SAFETY-31's pair - " "and NOT mirrored onto the pre-DEV-02 `sign_tripped`, which is the " "rollback view of a UNIT latch. A downgraded add-on drives exactly one " "actuator; handing it a latch taken on evidence about a roster it cannot " "have would be a trip it can never explain", _c31.store.data.get("site_sign_tripped") is True and "SIGN_DISAGREEMENT" in str(_c31.store.data.get("site_sign_reason")) and _c31.store.data.get("sign_tripped") is not True) check("the status page carries a SITE row of its own, and the banner says the " "site's TOTAL command opposed the meter rather than naming an inverter - " "naming one would send an operator to swap the wrong box", any("SITE SIGN-DISAGREEMENT FAILSAFE" in ck["text"] for ck in _c31.checks()) and "total command" in _c31.status_payload()["banner"]) _pub30 = {} _c31.mqtt = type("P", (), {"enabled": True, "publish": lambda _s, v: _pub30.update(v), "close": lambda _s: None})() _c31.publish() check("...and the controller-status sensor reports the failsafe state, so a " "site parked at 0 W cannot publish 'running'", _pub30.get("status") == "failsafe") # The latch, and the one way out of it. _was31 = _c31.site_sign_wd.tripped asyncio.run(_c31.handle_action("auto_toggle")) # stop check("pressing STOP clears nothing - the condition always goes away on its " "own, because the trip commands 0 W and a site at zero cannot oppose " "anything", _c31.site_sign_wd.tripped and _was31) asyncio.run(_c31.handle_action("auto_toggle")) # the deliberate start check("pressing START clears the latch in memory and on disk together, which " "is the same deliberate human act that clears every other latch here", not _c31.site_sign_wd.tripped and _c31.store.data.get("site_sign_tripped") is False and _c31.store.data.get("site_sign_reason") == "") async def _site_restart(store): hass = _SignHass(load_w=2000.0) c = _sign_site(hass, store=store, gain=-0.6) await _drive(c, n=3, dt=0.05) return c _store31 = _MemStore() _store31.data["site_sign_tripped"] = True _store31.data["site_sign_reason"] = "SIGN_DISAGREEMENT: (from before the restart)" _c31r = asyncio.run(_site_restart(_store31)) check("⚠️ AND IT SURVIVES A RESTART, which is the only thing that makes it a " "latch rather than a pause. 'Restart the add-on' is the first thing an " "operator staring at a FAILSAFE banner tries, and a site that came back " "armed would drive straight into the fault it had just stopped for", _c31r.site_sign_wd.tripped and _c31r.target == 0.0 and "sign_site" in _c31r.reason) print() print("6. AC 4: two inverters commanded OPPOSITE SIGNS by a legitimate") print(" allocation, and nothing trips") # ⚠️ THIS IS THE FALSE TRIP THE CARD OPENS WITH, AND IT IS DRIVEN THROUGH THE # SHIPPED PREDICATE RATHER THAN DESCRIBED. Inverter A charging 1500 W while # inverter B discharges 1500 W into a 1500 W import is an ordinary two-inverter # arrangement - a per-unit maintenance charge on A while the loop drives B, or # SAFETY-25's floor pinning one unit. Nothing is faulty. Before this card the # meter convicted A on it. _A_CMD, _B_CMD, _METER = -1500.0, 1500.0, 1500.0 _wd_a = SignWatchdog(_ucfg, against=AGAINST_OWN_POWER) _wd_b = SignWatchdog(_ucfg, against=AGAINST_OWN_POWER) _sitecfg = SignWatchdogConfig( min_w=float(SIGN_OPTION_DEFAULTS["site_sign_disagree_w"]), hold_s=float(SIGN_OPTION_DEFAULTS["site_sign_disagree_s"])) _wd_s = SignWatchdog(_sitecfg) for _t in range(0, 1200, 5): _wd_a.evaluate(_A_CMD, _A_CMD, True, T0 + timedelta(seconds=_t)) _wd_b.evaluate(_B_CMD, _B_CMD, True, T0 + timedelta(seconds=_t)) _wd_s.evaluate(_A_CMD + _B_CMD, _METER, True, T0 + timedelta(seconds=_t)) print(f" A commanded {_A_CMD:+.0f} W and measured {_A_CMD:+.0f} W; " f"B {_B_CMD:+.0f} / {_B_CMD:+.0f}; meter {_METER:+.0f} W") check("PRECONDITION: THE COMPARISON THAT SHIPPED CONVICTS INVERTER A HERE, on " "every one of these samples. A is commanded to CHARGE while the meter " "reads IMPORT - opposed, both far over 300 W, and it would have held for " "20 minutes. That is the false trip, and it is asserted rather than " "asserted about", disagrees(_A_CMD, _METER, True, _scfg) is True) check("⚠️ AC 4. NEITHER UNIT TRIPS UNDER THE NEW COMPARISON, over 20 minutes at " "an 85 s hold. Each one is doing exactly what it was told, and its own " "power sensor is the instrument that can say so", not _wd_a.tripped and not _wd_b.tripped) check("...and the SITE watchdog does not trip either, because the shares SUM " "to zero and a 0 W total is not a wrong-direction command - the " "magnitude gate that has always been there, now applied to the quantity " "it belongs on", not _wd_s.tripped and _wd_s.disagree_since is None) # ⚠️ AND THE COUNTERFACTUAL, BECAUSE "nothing tripped" IS ALSO WHAT A DISARMED # GUARD REPORTS. Same site, same meter, but the shares no longer cancel: a NET # -1600 W of charge into a +1500 W import is the site genuinely driving the # wrong way, and it must trip. Without this line the check above passes # identically against a site watchdog that was never wired to anything. _wd_s_net = SignWatchdog(_sitecfg) for _t in range(0, 600, 5): _wd_s_net.evaluate(-1600.0, _METER, True, T0 + timedelta(seconds=_t)) check("...and the site question is not thereby disarmed: the same site, " "commanded a NET -1600 W into the same import, DOES trip", _wd_s_net.tripped) print() print("7. AC 8: at ONE inverter this card changes NOTHING, and 'nothing' is") print(" asserted rather than assumed") async def _n1_sign(): hass = _SignHass(load_w=2000.0) opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "max_w": 2000, "gain": -0.6, "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True} c = Controller(opts, hass, _MemStore(), _P(), stop=asyncio.Event()) c.auto = True c.dev_min, c.dev_max = -5000.0, 5000.0 hass.ctrl = c # ⚠️ A CALL COUNTER, NOT AN OBSERVATION OF THE DETECTOR'S STATE, AND A # MUTATION IS WHY. The first version of the check below read # `site_sign_wd.disagree_since is None` - and arming the site detector at # N=1 as well left this suite GREEN, because the unit detector trips first # at its 1 s hold, parks the inverter, and the site's standing command is # then 0 W: the site predicate is false, so its window is never opened and # "never evaluated" reads identically to "evaluated and found nothing". That # is MUTATION.md's third failure mode - the assertion ran, the sentence was # true, and the data could not reach the failure it names. A counter can # only be zero for the one reason. _n1_calls = [] _orig_eval = c.site_sign_wd.evaluate def _spy(*a, **kw): _n1_calls.append(a) return _orig_eval(*a, **kw) c.site_sign_wd.evaluate = _spy await _drive(c, n=12, dt=0.2) return c, _n1_calls _c1s, _n1_site_calls = asyncio.run(_n1_sign()) _inv1 = _c1s.inverters[0] print(f" roster={_c1s.units} armed against {_inv1.sign_wd.against!r} " f"at {_inv1.sign_wd.cfg.min_w:.0f} W / {_inv1.sign_wd.cfg.hold_s:.0f} s") check("the single unit is graded against THE METER, at the SHIPPED thresholds " "- §6.2's identity, and the whole compatibility promise of this card. " "The unit's own power is not consulted, and `unit_sign_disagree_*` does " "nothing on a one-inverter install", _inv1.sign_wd.against == AGAINST_METER and (_inv1.sign_wd.cfg.min_w, _inv1.sign_wd.cfg.hold_s) == (300.0, 1.0)) check("...it trips, on the same evidence and through the same UNIT `failsafe` " "claim as before, commanding 0 W", _inv1.sign_wd.tripped and _c1s.target == 0.0 and "failsafe" in _c1s.reason and "sign_site" not in _c1s.reason) check("...the reason sentence still says 'a meter reading', word for word - it " "is what an operator reads, what the action log carries and what is " "persisted, so it is graded on the literal", "against a meter reading" in _inv1.sign_wd.reason) check("...the latch is under the same namespaced key with the same flat " "rollback mirror, and SAFETY-30's site pair is untouched at its default. " "A one-inverter install's state file is byte-identical to what it was", _c1s.store.data.get(TRIPPED_KEY) is True and _c1s.store.data.get("sign_tripped") is True and _c1s.store.data.get("site_sign_tripped") is False and _c1s.store.data.get("site_sign_reason") == "") print(f" site detector evaluated {len(_n1_site_calls)} times over " f"12 cycles") check("⚠️ ...AND THE SITE DETECTOR IS NEVER ARMED. Not 'did not trip', and not " "'its window stayed closed' - NEVER CALLED, counted on the object over " "twelve cycles. That is what says the second detector is ABSENT at a " "roster of one rather than merely quiet, and arming it would be two " "latches and two banners for one fault with a false-positive record " "measured on ONE comparison spent twice", _n1_site_calls == [] and not _c1s.site_sign_wd.tripped and _c1s.site_sign_wd.disagree_since is None) check("...the banner is the sentence that shipped", _c1s.status_payload()["banner"] == "FAILSAFE — the command opposed the meter. Inverter held at 0 W.") check("...the status page carries ONE sign row, with no unit tag and no site " "row - an extra row for an identically-duplicated question is a page " "advertising two guards where the install has one", len([ck for ck in _c1s.checks() if "sign watchdog" in ck["text"] or "SIGN-DISAGREEMENT" in ck["text"] ]) == 1 and not any("site sign" in ck["text"].lower() for ck in _c1s.checks())) check("...and no `sign_site` claim can be filed at a roster of one, so the " "arbitration is byte-identical to what shipped", "sign_site" not in _c1s.reason) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")