goodwe-addon/goodwe_controller/test_watchdog.py

"""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 ~8 days out (2026-09-04) and
# the bench half of SPIKE-04 measures it.
# ⚠️ SPIKE-04a (2026-09-04) ESTABLISHED B'S REAL RATING FROM ITS DATASHEET:
# a GoodWe GW5048D-ES, serial `55048ESU22AW0992`, 4600 W on the battery path
# (goodwe/NOTES.md §1.1b). THIS FIXTURE DELIBERATELY DOES NOT USE IT. 3000 W is
# kept because the only property these checks consume is that B's number is
# UNMISTAKABLY not A's, and the real pair is 5000/4600 - a 400 W gap, narrower
# than `slew_w` and inside the range where an off-by-one-unit bug and ordinary
# control arithmetic produce indistinguishable numbers. Collapsing the fixture
# onto the real values would cost these checks their separating power to buy
# realism they do not use. So: ILLUSTRATIVE, NOT B'S RATING, and not a
# placeholder for it either - the placeholder era ended on 2026-09-04.
B_RATING_HYPOTHESIS = 3000.0


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', '<omitted>')!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.

    ⚠️ THE THREE THRESHOLD PAIRS ARE PAIRWISE DISTINCT IN BOTH COORDINATES,
    AND THAT IS LOAD-BEARING RATHER THAN COSMETIC. This fixture is the only
    place in the suite where `_arm_sign_watchdogs()` runs at a roster of two, so
    it is the only place that can see WHICH pair reaches WHICH detector. The
    first version of it set every hold to 1 s and two of the three magnitude
    gates to 300 W; a detector armed with the shipped pair, or with the site
    pair, or with the right magnitude and the wrong hold, then compared EQUAL
    to the pair it was supposed to have, and the wiring check below would have
    passed on three of those four arrangements. That is MUTATION.md's third
    failure mode written into a fixture - a check comparing two things that
    happen to be equal cannot see the difference it names - and the
    precondition beside the check is what stops a later edit collapsing them
    again without a red line.

    Anyone changing a hold here: the PER-UNIT hold is what part 3's trip timing
    is read against, and the SITE hold must still expire inside `_drive()`'s
    window (14 cycles x 0.2 s), which part 5 depends on.
    """
    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": 3,
            "unit_sign_disagree_w": 250, "unit_sign_disagree_s": 1,
            "site_sign_disagree_w": 320, "site_sign_disagree_s": 2,
            "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

# ⚠️ THE WIRING PIN FOR AC 5, AND IT IS A DIFFERENT CLAIM FROM EVERY
# BEHAVIOURAL CHECK IN THIS SECTION. The rest of part 3 grades what the
# detectors CONCLUDED; this grades WHICH THRESHOLDS THEY WERE HANDED, and the
# gap between the two was live. Replacing
# `arm(self.sign_cfg if one else self.unit_sign_cfg, ...)` with
# `arm(self.sign_cfg, ...)` in `Controller._arm_sign_watchdogs()` throws the
# re-derived per-unit pair away and arms every unit with SAFETY-02's
# meter-facing one - and left ALL 19 FILES GREEN. Nothing at a roster of two
# read `inv.sign_wd.cfg` at all; the one place that does is part 7's N=1 arm,
# which that edit PRESERVES.
#
# WHY THE TRIPS BELOW CANNOT CATCH IT, said here because "the suite is green"
# and "nothing was wrong" are the two readings this comment exists to separate.
# B is reversed at ~2000 W - far over 250 W and far over 300 W - and every hold
# in this fixture is short enough to expire inside `_drive()`. A fault that is
# big enough and long enough to trip under BOTH pairs is evidence about
# neither. Only an equality on the pair itself can tell them apart.
#
# WHAT THE SURVIVOR COSTS IN THE FIELD, at config.yaml's own defaults: the
# per-unit check runs at `sign_disagree_s` 180 s instead of the re-derived 85 s
# - 2.1x slower to park a reversed inverter - and its gate moves from 250 W to
# 300 W, so a unit reversing a share between the two becomes invisible, while
# watchdog.SIGN_OPTION_DEFAULTS, config.yaml and DECISIONS section 3 all go on
# describing a pair the running code does not use.
_UNIT_PAIR = (250.0, 1.0)      # `unit_sign_disagree_w` / `_s`  - one per unit
_SITE_PAIR = (320.0, 2.0)      # `site_sign_disagree_w` / `_s`  - one per site
_SHIPPED_PAIR = (300.0, 3.0)   # `sign_disagree_w` / `_s` - N=1 only
_armed = {inv.name: (inv.sign_wd.cfg.min_w, inv.sign_wd.cfg.hold_s)
          for inv in _c30.inverters}
_armed_site = (_c30.site_sign_wd.cfg.min_w, _c30.site_sign_wd.cfg.hold_s)
print(f"       armed per unit: {_armed}")
print(f"       armed site:     {_armed_site}    "
      f"shipped pair, wanted by NOBODY here: {_SHIPPED_PAIR}")
check("PRECONDITION: the three pairs are pairwise distinct IN BOTH "
      "COORDINATES, so an equality below can fail on the magnitude alone, on "
      "the hold alone, or on both. This fixture originally gave all three the "
      "same 1 s hold and two of them the same 300 W gate, and a mis-wiring "
      "that delivered the right magnitude with the wrong hold would then have "
      "compared EQUAL to the pair it was supposed to have",
      len({_UNIT_PAIR, _SITE_PAIR, _SHIPPED_PAIR}) == 3
      and len({p[0] for p in (_UNIT_PAIR, _SITE_PAIR, _SHIPPED_PAIR)}) == 3
      and len({p[1] for p in (_UNIT_PAIR, _SITE_PAIR, _SHIPPED_PAIR)}) == 3)
check("PRECONDITION: and all three are pairs this Controller actually holds, "
      "so the literals below are the fixture's own numbers and not three "
      "constants that agree with nothing. In particular the shipped pair is "
      "still constructed and still a reachable mis-wiring target",
      (_c30.unit_sign_cfg.min_w, _c30.unit_sign_cfg.hold_s) == _UNIT_PAIR
      and (_c30.site_sign_cfg.min_w, _c30.site_sign_cfg.hold_s) == _SITE_PAIR
      and (_c30.sign_cfg.min_w, _c30.sign_cfg.hold_s) == _SHIPPED_PAIR)
check("⚠️ AC 5 WIRING. EVERY per-unit detector on this roster of two is armed "
      "with the PER-UNIT pair - 250 W / 1 s here, `unit_sign_disagree_w` and "
      "`unit_sign_disagree_s` in the field. Graded against LITERALS rather "
      "than against `_c30.unit_sign_cfg`, which cannot tell 'the per-unit "
      "pair' from 'whatever this Controller was handed'. This is the line that "
      "goes red when `_arm_sign_watchdogs()` stops choosing between the two "
      "pairs and hands every unit the site-facing one",
      len(_armed) == 2 and all(p == _UNIT_PAIR for p in _armed.values()))
check("...and the SITE detector is armed with the SITE pair - 320 W / 2 s "
      "here - which differs from the per-unit pair AND from the shipped pair "
      "in both coordinates, so this equality is not carried by a number the "
      "fixture happens to hold twice",
      _armed_site == _SITE_PAIR)
check("...and each detector is pointed at its own INSTRUMENT as well as at its "
      "own thresholds: the units at their own 0x0518 power, the site at the "
      "meter. They are two arguments of one call and a pin on one is not a pin "
      "on the other",
      all(inv.sign_wd.against == AGAINST_OWN_POWER for inv in _c30.inverters)
      and _c30.site_sign_wd.against == AGAINST_METER)

_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")