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."""

    def __init__(self):
        self.data = {"phase": "idle", "sign_tripped": False, "sign_reason": ""}

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, *a):
        return None


async def _restart():
    shared = _MemStore()
    hass1 = _Hass()
    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
            "meter_source": "off", "max_w": 2000, "gain": -0.6,
            "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True}
    a = Controller(opts, hass1, shared, _P(), stop=asyncio.Event())
    a.auto = True
    a.dev_min, a.dev_max = -5000.0, 5000.0
    for _ in range(12):
        await a.read_inputs()
        await a.cycle()
        await asyncio.sleep(0.25)

    # The restart. Same options, same store, brand new Controller and Hass.
    hass2 = _Hass()
    b = Controller(opts, hass2, shared, _P(), stop=asyncio.Event())
    b.auto = True
    b.dev_min, b.dev_max = -5000.0, 5000.0
    for _ in range(6):
        await b.read_inputs()
        await b.cycle()
        await asyncio.sleep(0.05)
    return a, b, shared, hass2


pre, post, store, hass2 = asyncio.run(_restart())
check("it tripped before the restart", pre.sign_wd.tripped)
check("the trip reached the store, UNDER THE TRIPPED INVERTER'S OWN KEY",
      store.data[TRIPPED_KEY] is True)
check("...with the reason beside it, so the operator is told why",
      "SIGN_DISAGREEMENT" in store.data[REASON_KEY])
check("⚠️ ...AND THE FLAT PRE-DEV-02 KEY IS MIRRORED, AS THE ROLLBACK VIEW. "
      "This assertion used to read the other way - that the flat key was NOT "
      "written - on the argument that writing it would latch every other "
      "inverter on the site. That argument was right about the READ and wrong "
      "about the WRITE: nothing at runtime reads the flat key, and the one "
      "thing that does (`_grade_store_migration()`, on every start) ignores it "
      "whenever ANY namespaced latch is set, which a sibling's trip always "
      "leaves. What the absence actually cost was the ROLLBACK: store.py's "
      "rule 2 keeps this key so a downgrade finds the latch, and a trip taken "
      "after the migration wrote nothing here - so an operator who downgraded "
      "came up DRIVING. `test_inverter.py` part 3 is where 'B's trip does not "
      "latch A across a restart' is graded, on a real roster of two",
      store.data.get("sign_tripped") is True
      and "SIGN_DISAGREEMENT" in str(store.data.get("sign_reason", "")))
check("the new Controller comes back TRIPPED", post.sign_wd.tripped)
check("...and says why, restored rather than re-detected",
      "SIGN_DISAGREEMENT" in post.sign_wd.reason)
check("...commands 0 W from its very first cycle", post.target == 0.0)
check("...and never commanded anything else after the restart",
      hass2.writes != [] and set(hass2.writes) == {0.0})
check("the banner still leads with the failsafe",
      "FAILSAFE" in post.status_payload()["banner"])
# ⚠️ TWO PRESSES, and the first one must NOT clear it. `auto` is True while
# tripped, so the first press stops control; only the transition back to True
# clears. That is the documented behaviour and it is worth an assertion, because
# a one-press clear would be reachable by an operator who never read the banner.
asyncio.run(post.handle_action("auto_toggle"))          # -> stopped
check("pressing stop does not clear the failsafe",
      post.sign_wd.tripped and store.data[TRIPPED_KEY] is True)
asyncio.run(post.handle_action("auto_toggle"))          # -> started
check("pressing start DOES clear it, and clears the store with it - BOTH "
      "spellings, which is what stops the every-start latch guard re-arming a "
      "failsafe the operator has already dealt with",
      not post.sign_wd.tripped and store.data[TRIPPED_KEY] is False
      and store.data[REASON_KEY] == ""
      and store.data.get("sign_tripped") is False)
# One more restart from the cleared store, to prove the clear persisted too.
fresh = Controller({"setpoint_entity": "number.sp", "meter_source": "off"},
                   _Hass(), store, _P(), stop=asyncio.Event())
check("...so the next restart comes back armed and untripped, as it should",
      not fresh.sign_wd.tripped)

# ⚠️ THE WRITE ORDERING IS AN INVARIANT, SO IT GETS AN ASSERTION. main.py writes
# `sign_reason` and THEN `sign_tripped`, each one its own `save()`, so that a
# process that dies between the two leaves a reason nothing reads - harmless -
# rather than a latch with no explanation for the operator who has to decide
# whether to clear it. Documented reasoning is not enforcement: swapping those two
# lines used to pass every check in this file.
#
# The death is modelled as the second write raising. In the field it is a power
# cut or a container kill (Store.save() swallows OSError rather than raising), so
# what is asserted here is the PROPERTY - what survives on disk - not the
# mechanism.


class _CrashOnFlag(_MemStore):
    def set(self, k, v):
        if k == TRIPPED_KEY and v:
            raise RuntimeError("power cut, mid-trip")
        super().set(k, v)


async def _crash_mid_write():
    store = _CrashOnFlag()
    hass = _Hass()
    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
            "meter_source": "off", "max_w": 2000, "gain": -0.6,
            "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True}
    c = Controller(opts, hass, store, _P(), stop=asyncio.Event())
    c.auto = True
    c.dev_min, c.dev_max = -5000.0, 5000.0
    died = False
    for _ in range(12):
        try:
            await c.read_inputs()
            await c.cycle()
        except RuntimeError:
            died = True
            break
        await asyncio.sleep(0.25)
    return store, died


crashed_store, died = asyncio.run(_crash_mid_write())
check("the second write is the one that sets the flag", died)
check("a crash between the two writes leaves the REASON on disk",
      "SIGN_DISAGREEMENT" in crashed_store.data.get(REASON_KEY, ""))
check("...and never a latch with nothing to explain it",
      not crashed_store.data.get(TRIPPED_KEY))


# =============================================================================
print()
print("=" * 72)
print("SAFETY-31: circulating power, which nets to zero at the meter")
print("=" * 72)
# ⚠️ READ THIS BEFORE ADDING A CHECK HERE. Half of this card was already built
# and it is NOT the half below. ALLOC-01 shipped the SIGN-COHERENCE GATE
# (`allocator.REFUSE_SIGN`, graded by test_alloc.py part 6): an allocation whose
# shares carry two signs is REFUSED, both units named, no shares returned. So
# the allocator can no longer COMMAND circulating power, and nothing here
# re-tests that.
#
# What is graded here is the MEASUREMENT half: circulating power the allocator
# NEVER COMMANDED. The two are complements, not duplicates - the gate refuses an
# allocation before it is written and cannot see one millisecond past that; this
# reads the hardware and does not care what was written.

from app.inverter import LINK_STALE_S  # noqa: E402
from app.circulation import (  # noqa: E402
    CIRC_OPTION_DEFAULTS, CirculationConfig, CirculationWatchdog,
    circulating_w, opposing,
)

print()
print("1. the sensor: sum(|Bi|) - |sum(Bi)| over MEASURED per-unit power")
check("both units discharging: the triangle equality holds and it is ZERO. "
      "This is the ordinary case and the sensor must be silent in it",
      circulating_w([1000.0, 2000.0]) == 0.0)
check("...both charging, likewise", circulating_w([-1000.0, -2000.0]) == 0.0)
check("...and a unit resting at exactly zero beside a working one is not "
      "opposition", circulating_w([0.0, 2500.0]) == 0.0)
# ⚠️ ASSERTED AGAINST A LITERAL, NOT AGAINST `2 * min(abs(a), abs(b))`.
# MUTATION.md's rule 2: an assertion written in terms of the expression under
# test cannot tell "the right formula" from "the same wrong formula twice".
check("A charging 3000 W while B discharges 3000 W reads 6000 W - the card's "
      "own scenario, and the figure is 2*min(|A|,|B|), the power going round "
      "the loop through BOTH conversion stages",
      circulating_w([-3000.0, 3000.0]) == 6000.0)
check("...unequal opposition is bounded by the smaller unit: -3000 against "
      "+1000 is 2000 W, not 4000",
      circulating_w([-3000.0, 1000.0]) == 2000.0)
print(f"       [-3000, +3000] -> {circulating_w([-3000.0, 3000.0]):.0f} W   "
      f"[-3000, +1000] -> {circulating_w([-3000.0, 1000.0]):.0f} W   "
      f"[+1000, +2000] -> {circulating_w([1000.0, 2000.0]):.0f} W")
check("three units, one out of step: it is defined over the whole roster, not "
      "over a pair", circulating_w([-2000.0, 1000.0, 1000.0]) == 4000.0)

print()
print("   AC 5: at N=1 it is IDENTICALLY zero, for every reading")
# ⚠️ SWEPT, NOT SAMPLED. "Identically zero" is a claim about EVERY value, and
# one call with one number cannot distinguish it from "zero for this number".
_n1 = [circulating_w([v]) for v in
       (0.0, 1.0, -1.0, 2500.0, -2500.0, 5000.0, -5000.0, 1e9, -1e9)]
check("nine readings from -1e9 to +1e9 W, every one of them 0.0",
      set(_n1) == {0.0})
check("...and an EMPTY roster is zero too, not a crash - `Controller.publish()` "
      "calls this before anything has been read",
      circulating_w([]) == 0.0)
check("⚠️ AND A MISSING READING AT N=1 IS STILL ZERO, WHICH IS THE OPPOSITE OF "
      "THE N>=2 RULE BELOW AND IS DELIBERATE. With one term the algebra has no "
      "room for an unknown: |B| - |B| is 0 for every B, INCLUDING one nobody "
      "read. Returning None here would park the shipped one-inverter install in "
      "a permanent 'cannot tell' on a quantity that cannot be anything else",
      circulating_w([None]) == 0.0)

print()
print("   a missing reading at N>=2 is UNKNOWN, and never zero")
check("⚠️ None, NOT 0.0. Zero would be this sensor reporting HEALTH FROM AN "
      "ABSENCE - which is the 2026-09-02 failure itself, where a subsystem "
      "stopped answering and its silence read as calm for 6 h 25 m",
      circulating_w([None, 3000.0]) is None
      and circulating_w([-3000.0, None]) is None
      and circulating_w([None, None]) is None)

print()
print("   it is NON-NEGATIVE, including where floating point would say otherwise")
_fp = [circulating_w([0.1, 0.2, 0.3]), circulating_w([1e-9, 2e-9]),
       circulating_w([-0.1, -0.2, -0.30000000000000004])]
print(f"       {_fp}")
check("a same-sign roster never publishes a negative number. ⚠️ THIS GRADES "
      "THE PROPERTY, NOT THE `max(0.0, ...)` GUARD THAT DEFENDS IT, and the "
      "difference is stated so a mutation sweep does not report a gap that is "
      "not one: for a same-sign roster the two sums accumulate the same "
      "magnitudes in the same order and are bit-identical, so the guard is an "
      "EQUIVALENT MUTANT - searched over 700 000 random rosters without a "
      "single negative. circulation.py says so at the line",
      all(v >= 0.0 for v in _fp))

print()
print("2. CFG-03: config.yaml is the source of truth and there is ONE copy")
# ⚠️ READ OFF THE FILE, NOT TRANSCRIBED. The Dockerfile does not put config.yaml
# in the image, so the running add-on cannot read it and `CIRC_OPTION_DEFAULTS`
# has to exist - which makes it a place to drift. This is the same pin
# test_maintenance.py puts on MAINT_OPTION_DEFAULTS, for the same reason.
import re as _re  # noqa: E402
from pathlib import Path as _Path  # noqa: E402

_cfg_txt = (_Path(__file__).resolve().parent / "config.yaml").read_text(encoding="utf-8")
# ⚠️ SPLIT ON THE TOP-LEVEL KEY, NOT ON THE WORD. "schema:" appears inside the
# CFG-03 comment above these very options ("options:, schema:, ..."), and
# splitting on the bare word cut the file in the middle of a comment - the
# options block then contained no options, the regex found none, and this pin
# compared an EMPTY dict against an empty expectation. That is MUTATION.md's
# third failure mode exactly: the assertion ran, the sentence was true, and the
# data could not reach the failure it names. Caught by the printed line.
_cfg_opts, _cfg_schema = _cfg_txt.split("\nschema:", 1)
_from_yaml = {k: int(v) for k, v in
              _re.findall(r"^  (circulation_\w+):\s*(\d+)\s*$", _cfg_opts, _re.M)}
print(f"       config.yaml options: {_from_yaml}")
print(f"       CIRC_OPTION_DEFAULTS: {CIRC_OPTION_DEFAULTS}")
check("both options are declared in config.yaml's options: block and the "
      "in-code mirror agrees with it key for key and value for value",
      _from_yaml == CIRC_OPTION_DEFAULTS and len(_from_yaml) == 2)
_bounds = dict(_re.findall(r"^  (circulation_\w+):\s*int\((\d+),(\d+)\)\s*$",
                           _cfg_schema, _re.M)
               and [(m[0], (int(m[1]), int(m[2]))) for m in
                    _re.findall(r"^  (circulation_\w+):\s*int\((\d+),(\d+)\)\s*$",
                                _cfg_schema, _re.M)])
print(f"       schema bounds: {_bounds}")
check("...and BOTH have a schema range, so the Supervisor refuses an "
      "out-of-range value rather than the add-on discovering it at runtime",
      set(_bounds) == set(CIRC_OPTION_DEFAULTS))
check("⚠️ AND THE THRESHOLD'S FLOOR IS THE MEASURED KNEE, NOT A ROUND NUMBER. "
      "On the TEST-19 rig the worst CONTINUOUS opposition a site where NOTHING "
      "is faulty can produce is 5.0 s at 400 W, 10.0 s at 300 W, 40.0 s at "
      "250 W and 505.0 s at 100 W. Below the knee, ordinary operation holds "
      "this sensor over the threshold for MINUTES and no window shorter than "
      "the 900 s billing quarter could tell that from a fault - so a "
      "configurable value under 400 W would be a false-alarm generator, and "
      "the schema does not offer one",
      _bounds["circulation_w"][0] == 400)
check("...and the window's floor is the firmware's own veto window, "
      "`inverter.LINK_STALE_S`, because below it a unit still executing a "
      "stale share is a LINK fault the add-on already handles by dropping it "
      "from the split - and its ceiling is the 900 s quarter that bills the "
      "capacity tariff, inside which this must be caught and acted on",
      _bounds["circulation_hold_s"] == (int(LINK_STALE_S), 900))
check("the shipped defaults sit inside their own schema ranges",
      all(_bounds[k][0] <= v <= _bounds[k][1]
          for k, v in CIRC_OPTION_DEFAULTS.items()))
check("⚠️ AND NEITHER IS A ROUND NUMBER, WHICH IS AC 3 READ LITERALLY. 700 W "
      "is 1.75x the measured knee - the headroom is bought for inverter B's "
      "ramp and lag being PLACEHOLDERS until SPIKE-04 measures them - and 75 s "
      "is 15x the worst legitimate episode and 2.5x the firmware veto. 60, 100, "
      "120, 180, 300, 500 and 1000 are all values a guess would have landed on",
      CIRC_OPTION_DEFAULTS["circulation_w"] not in (100, 500, 1000, 2000)
      and CIRC_OPTION_DEFAULTS["circulation_hold_s"] not in (30, 60, 120, 180, 300))

print()
print("3. the trip: sustained opposition fires, the slew transient does not")


def circ_drive(series, cfg=None, step=5.0):
    """Feed a list of circulating figures, one per control cycle."""
    wd = CirculationWatchdog(cfg or CirculationConfig())
    at = None
    for i, v in enumerate(series):
        if wd.evaluate(v, T0 + timedelta(seconds=i * step)) and at is None:
            at = i * step
    return wd, at


_CFG = CirculationConfig()
_sustained, _at = circ_drive([2000.0] * 40)
check("2000 W of measured opposition, held, trips", _sustained.tripped)
check(f"...and at {_CFG.hold_s:.0f} s, not before - the window is a window, "
      "not a delay somebody rounded", _at == _CFG.hold_s)
check("the reason gives the operator the figure, the peak and the duration, "
      "and says in words why nothing else reported it",
      "2000" in _sustained.reason and "75 s" in _sustained.reason
      and "nets to zero" in _sustained.reason.lower())
print(f"       e.g. {_sustained.reason!r}")

print()
print("   ⚠️ AND NOT ON THE TRANSIENT, WHICH IS THE FALSE-ALARM CASE")
# ⚠️ THE SHAPES BELOW ARE THE RIG'S, NOT INVENTED. Measured on TEST-19 over 924
# closed-loop runs in which every unit executed the share it was given: the
# worst continuous opposition above 700 W is 5.0 s - ONE control cycle - at a
# peak of 1700 W. That is the serialised write path: on a reversing cycle B's
# new number lands up to a whole cycle after A's, both slew through zero at
# 800 W/s behind a 10 s lag, and between the two crossings they genuinely
# oppose. A guard that trips on that is a false-alarm generator, and this
# project's doctrine on crying wolf is explicit (SAFETY-15, SAFETY-19).
_one_cycle = [0.0] * 5 + [1700.0] + [0.0] * 30
_t1, _ = circ_drive(_one_cycle)
check("the worst single transient the rig can produce - 1700 W for ONE control "
      "cycle - does not trip", not _t1.tripped)
_repeated = ([0.0] * 3 + [1700.0]) * 20
_t2, _ = circ_drive(_repeated)
check("...and neither does it REPEATED every fourth cycle for 400 s, which is "
      "a load flapping across zero all afternoon. The window clears between "
      "them, so nothing accumulates", not _t2.tripped)
_almost = [1500.0] * 14 + [0.0] + [1500.0] * 14
_t3, _ = circ_drive(_almost)
check("...and opposition that stands for 70 s, clears for one cycle, then "
      "stands again for 70 s does NOT trip: 140 s of trouble, but never 75 s "
      "CONTINUOUSLY, which is what the guard actually claims to measure",
      not _t3.tripped)
_below = [_CFG.min_w] * 60
_t4, _ = circ_drive(_below)
check(f"...and exactly {_CFG.min_w:.0f} W forever does not trip - the "
      "comparison is strictly greater, so the threshold is a threshold rather "
      "than an off-by-one", not _t4.tripped)
_just_over, _at5 = circ_drive([_CFG.min_w + 1.0] * 40)
check("...while one watt above it does, held long enough. The pair is what "
      "makes the boundary a measured line rather than a direction",
      _just_over.tripped and _at5 == _CFG.hold_s)

print()
print("   a MISSING reading clears the window rather than accumulating it")
_gap, _ = circ_drive([2000.0] * 10 + [None] + [2000.0] * 10)
check("⚠️ SAFETY-02 MAKES THE SAME DECISION FOR A STALE METER, AND FOR THE SAME "
      "REASON: a frozen figure held against a moving one manufactures a fault "
      "in about a window's worth of cycles. 50 s over, one unknown, 50 s over "
      "again is not 100 s of opposition", not _gap.tripped)

print()
print("   the latch: it does not re-evaluate, because the trip removes the evidence")
_latched, _ = circ_drive([2000.0] * 40 + [0.0] * 40)
check("still tripped after the opposition has gone - and the opposition ALWAYS "
      "goes, because the trip commands 0 W everywhere and two inverters at zero "
      "cannot oppose. A non-latching version would de-assert and start the "
      "whole thing again, three cycles at a time, forever", _latched.tripped)
_r = CirculationWatchdog(_CFG, tripped=True, reason="restored from the store")
check("...and one restored as tripped stays tripped without being re-driven",
      _r.evaluate(0.0, T0) and _r.reason == "restored from the store")
_r.reset()
check("...until it is reset, which is the operator's press and nothing else",
      not _r.tripped and _r.reason == "" and _r.held_s(T0) == 0.0)

print()
print("   `opposing()` names the two sides, for the sentence the operator reads")
check("indices of who is discharging and who is charging",
      opposing([-3000.0, 3000.0]) == ([1], [0]))
check("...and a unit idling below the floor is not named as a participant in "
      "somebody else's loop", opposing([-3000.0, 3000.0, 3.0], min_w=350.0)
      == ([1], [0]))

print()
print("4. END TO END, AND THIS IS AC 2 - the sensor is MEASURED, so it sees")
print("   circulation the allocator NEVER COMMANDED")
# ⚠️ THIS IS THE CHECK MUTATION.md's THIRD FAILURE MODE IS ABOUT, so it is
# built the way that section says to build one. The claim is "X differs from Y"
# - the MEASURED sensor differs from a sensor computed off the allocation
# record - and X and Y are ACTUALLY MADE DIFFERENT here, with the difference
# asserted as its own precondition and PRINTED, before anything is concluded
# from it. DEV-02 shipped a check comparing a number against itself; ALLOC-01
# shipped a generator that found 0 escapes where a wider one found 39 991.

from app.inverter import Inverter  # noqa: E402

# ⚠️ INVERTER B DOES NOT PHYSICALLY EXIST - it is ~10 days out and SPIKE-04
# measures it. Its rating here is a HYPOTHESIS, named as one, exactly as
# test_peakguard.py's helper names its own.
B_RATING_HYPOTHESIS = 3000.0


class _CircHass(_Hass):
    """A house at rest, and two inverters MEASURED to be fighting each other.

    ⚠️ THE MEETER READS ZERO AND THAT IS THE WHOLE POINT. grid = load - sum(B),
    and with A at -3000 W and B at +3000 W the sum is zero, so P1 reads exactly
    what an idle site reads. Every existing guard in this add-on is a consumer
    of that number.
    """

    def __init__(self, a_w=-3000.0, b_w=3000.0):
        super().__init__()
        self.a_w, self.b_w = a_w, b_w

    async def number(self, entity, invert=False):
        return {"sensor.grid": 0.0,
                "sensor.soc": 50.0, "sensor.soc_b": 50.0,
                "sensor.batt": self.a_w, "sensor.batt_b": self.b_w}.get(entity)


def _two_inverter(hass, store, *, hold_s=1, min_w=700, **over):
    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
            "meter_source": "off", "max_w": 5000, "gain": 0.6,
            "circulation_w": min_w, "circulation_hold_s": hold_s,
            "auto_start": True}
    opts.update(over)
    c = Controller(opts, hass, store, _P(), stop=asyncio.Event())
    b = Inverter("inverter-b", setpoint_entity="number.sp_b",
                 soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
                 max_w=B_RATING_HYPOTHESIS, store=store)
    b.dev_min, b.dev_max = -B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS
    c.inverters.append(b)
    c.auto = True
    for inv in c.inverters:
        inv.dev_min = inv.dev_min if inv.dev_min is not None else -5000.0
        inv.dev_max = inv.dev_max if inv.dev_max is not None else 5000.0
    return c


async def _circ_e2e():
    hass = _CircHass()
    c = _two_inverter(hass, _MemStore())
    seen = []
    for _ in range(14):
        await c.read_inputs()
        await c.cycle()
        seen.append(dict(circ=c.site_circulating_w, grid=c.grid,
                         target=c.target, alloc=c.allocation,
                         tripped=c.circ_wd.tripped))
        await asyncio.sleep(0.25)
    return c, hass, seen


_c, _chass, _seen = asyncio.run(_circ_e2e())
_first = _seen[0]
_alloc = _first["alloc"]
_shares = dict(_alloc.shares)
_meas = [inv.batt for inv in _c.inverters]
print(f"       MEASURED per-inverter power : {dict(zip(_c.units, _meas))}")
print(f"       ALLOCATION RECORD           : ok={_alloc.ok}  {_alloc.explain()}")
print(f"       meter (P1)                  : {_first['grid']:+.0f} W")
print(f"       sensor over MEASURED power  : {circulating_w(_meas):.0f} W")
print(f"       sensor over the RECORD      : "
      f"{circulating_w(list(_shares.values())):.0f} W")

# --- THE PRECONDITION, AS ITS OWN CHECK -------------------------------------
# ⚠️ MUTATION.md RULE 1. If a later edit makes the record and the measurement
# agree again, THIS goes red rather than the claim below going quietly vacuous.
check("PRECONDITION: the allocation record and the measured power are "
      "ACTUALLY DIFFERENT in this fixture. The record is coherent - it is `ok`, "
      "its shares carry one sign, and they sum to the site target - while the "
      "hardware is measured at -3000 W and +3000 W. Without this line the two "
      "claims below could both be true of a fixture in which nothing differed",
      _alloc.ok is True
      and len({(s > 0) - (s < 0) for s in _shares.values() if s}) <= 1
      and sorted(_meas) == [-3000.0, 3000.0]
      and set(_shares.values()) != set(_meas))
check("ALLOC-01's gate is SATISFIED here, and that is the point rather than an "
      "accident: the allocator refused nothing, because it commanded nothing "
      "wrong. `REFUSE_SIGN` cannot fire on a record whose shares agree in sign",
      _alloc.refused is None)
check("⚠️ SO A SENSOR COMPUTED FROM THE ALLOCATION RECORD READS EXACTLY 0 W. "
      "That is the cheap version of this card, and it is worth nothing: "
      "ALLOC-01's sign-coherence gate already guarantees the record is "
      "sign-coherent, so a record-derived sensor is zero BY CONSTRUCTION and "
      "can only ever confirm that the allocator agrees with itself",
      circulating_w(list(_shares.values())) == 0.0)
check("⚠️ AND THE SHIPPED SENSOR READS 6000 W ON THE SAME CYCLE, because it is "
      "derived from `Inverter.batt` - each unit's OWN measurement. This is the "
      "whole card: it detects circulation THE ALLOCATOR NEVER COMMANDED - a "
      "unit ignoring or lagging its share, a stale share still being executed, "
      "or HW-01's power stage running its last instruction for 6 h 25 m with "
      "the comms dead and the record impeccable throughout",
      _c.site_circulating_w == 6000.0)
check("...and the METER READS ZERO through all of it, which is why no existing "
      "guard fires. SAFETY-02 gates on abs(grid_w) > 300 W and that gate never "
      "opens; the peak guard and the capacity tariff both bill grid exchange",
      _first["grid"] == 0.0
      and not any(i.sign_wd.tripped for i in _c.inverters))

print()
print("   ...and the trip ACTS: a site-wide stop, not a notification")
check("the watchdog tripped", _c.circ_wd.tripped)
check("the site is commanded 0 W", _c.target == 0.0)
check("...and 0 W actually reached BOTH inverters, not just the one somebody "
      "decided was to blame",
      _chass.writes and _chass.writes[-1] == 0.0
      and all(i.target == 0.0 for i in _c.inverters))
check("...and it got there through the `circulation` claim, by name, rather "
      "than by some other guard happening to stop the site",
      "circulation" in _c.reason)
print(f"       status line: {_c.reason!r}")
check("the status page says so in words, and says the meter cannot see it",
      any("CIRCULATING POWER" in ck["text"] and "meter cannot see" in ck["text"]
          for ck in _c.checks()))
check("the banner leads with it - above the sign failsafe, because a sign "
      "disagreement is visible on every dashboard in the house and this is "
      "visible nowhere at all",
      "CIRCULATING POWER" in _c.status_payload()["banner"])
_pub = {}
_c.mqtt = type("P", (), {"enabled": True,
                         "publish": lambda _s, v: _pub.update(v),
                         "close": lambda _s: None})()
_c.publish()
print(f"       published: circulation={_pub.get('circulation')} "
      f"grid={_pub.get('grid')} status={_pub.get('status')}")
check("⚠️ AND IT IS PUBLISHED AS ITS OWN SENSOR BESIDE THE GRID FIGURE. "
      "`grid 0 W` and `circulating 6000 W` on the same dashboard is the whole "
      "phenomenon in two rows, and NO arithmetic on the grid figure alone can "
      "ever produce the second one",
      _pub.get("circulation") == 6000 and _pub.get("grid") == 0.0)
check("...and the controller-status sensor reports the failsafe state",
      _pub.get("status") == "failsafe")
# ⚠️ AND THE UNKNOWN CASE, WHICH A SWEEP FOUND UNGRADED. A per-unit reading that
# did not arrive makes the site figure None, and `MqttPublisher.publish()` DROPS
# None values - so the entity keeps its last figure and the dashboard shows a
# stale number rather than a fresh zero. Publishing `round(x or 0.0)` instead
# survived the first mutation battery: it turns "I do not know" into "0 W", i.e.
# health manufactured from an absence, on the one surface an operator actually
# watches. Same defect as the sensor's own None rule, one layer out.
_c.inverters[1].batt = None
_pub_unknown = {}
_c.mqtt = type("P", (), {"enabled": True,
                         "publish": lambda _s, v: _pub_unknown.update(
                             {k: x for k, x in v.items() if x is not None}),
                         "close": lambda _s: None})()
_c.publish()
print(f"       with inverter-b's reading missing, publish() offers: "
      f"circulation={_pub_unknown.get('circulation', '<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()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")