goodwe-addon/goodwe_controller/test_maintenance.py

"""Runnable check for the maintenance state machine.  `python3 test_maintenance.py`

⚠️ SINCE MAINT-05 THIS FILE IMPORTS `app.main`, so it needs the add-on image's
`aiohttp` like test_inverter.py and test_clamp.py do. Parts 8-10 grade behaviour
that only exists once a Controller has built a roster - which pack a claim is
filed on, which pack's SoC a machine is ticked with, what a one-inverter store
looks like afterwards - and none of that is observable from the state machine
alone.

Walks a whole cycle in simulated time, which is the test the YAML version never
had - and it is exactly the phase transitions that a real cycle takes ten hours
to exercise once a month.

⚠️ 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 asyncio
import dataclasses
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

from app.maintenance import (
    CHARGE, DRAIN, HOLD, IDLE, MAINT_OPTION_DEFAULTS, POWER_MOVING, MaintConfig,
    Maintenance, UnitHistory, arbitrate,
)
from app.main import Controller
from app.store import (
    DEFAULTS, MAINT_SCHEMA_KEY, MAINT_SCHEMA_VERSION, MAINT_UNIT_KEYS,
    migrate_maint_keys, unit_key,
)

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


class FakeStore:
    """Same surface as Store, no disk."""

    def __init__(self):
        self.data = {"phase": IDLE, "phase_started": None,
                     "last_completed": None, "last_start_attempt": None}

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

    def set_time(self, k, when):
        self.data[k] = when

    def get_time(self, k):
        return self.data.get(k)

    def save(self):
        pass


cfg = MaintConfig(enabled=True, interval_days=28, start_hour=10,
                  discharge_w=2500, charge_w=2500,
                  soc_floor=11, soc_target=99, hold_min=120)

t0 = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)

print("scheduling")
m = Maintenance(cfg, FakeStore())
check("never run before is due", m.due(t0))
r = m.tick(t0, soc=80)
check("starts at the start hour when due", r.phase == DRAIN and r.owns_setpoint)
check("drain commands the discharge rate", r.setpoint_w == 2500)

m2 = Maintenance(cfg, FakeStore())
r = m2.tick(t0.replace(hour=11), soc=80)
check("does not start outside the start hour", r.phase == IDLE)

m3 = Maintenance(cfg, FakeStore())
m3.store.set_time("last_completed", t0 - timedelta(days=3))
check("not due three days after a cycle", not m3.due(t0))
check("due 29 days after a cycle",
      Maintenance(cfg, FakeStore()).due(t0) and
      (t0 - (t0 - timedelta(days=29))) >= timedelta(days=28))

print("ownership per phase")
m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0, soc=80)
check("drain: maintenance owns the setpoint", r.owns_setpoint and not r.charge_only)
r = m.tick(t0 + timedelta(hours=3), soc=10.5)
check("drain exits at the floor", r.phase == CHARGE)
check("charge: the LOOP owns the setpoint", not r.owns_setpoint and r.charge_only)
r = m.tick(t0 + timedelta(hours=20), soc=99)
check("charge exits at the target", r.phase == HOLD)
check("hold: maintenance owns and commands zero",
      r.owns_setpoint and r.setpoint_w == 0.0)
r = m.tick(t0 + timedelta(hours=21), soc=100)
check("hold keeps holding before hold_min", r.phase == HOLD)
r = m.tick(t0 + timedelta(hours=22, minutes=5), soc=100)
check("hold completes after hold_min", r.phase == IDLE)
check("completion is recorded", m.store.get_time("last_completed") is not None)
check("no longer due right after completing", not m.due(t0 + timedelta(hours=22)))

print("failure handling")
m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0 + timedelta(hours=13), soc=60)
check("drain aborts on timeout", r.phase == IDLE)
check("abort commands zero", r.setpoint_w == 0.0)

m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0 + timedelta(minutes=5), soc=None)
check("no SoC: holds the phase rather than guessing", r.phase == DRAIN)
check("no SoC: still commands the drain rate", r.setpoint_w == 2500)

m = Maintenance(cfg, FakeStore())
m.store.set("phase", "banana")
r = m.tick(t0, soc=50)
check("unknown phase recovers to idle at 0 W", r.phase == IDLE and r.setpoint_w == 0.0)

print("disabled schedule")
m = Maintenance(MaintConfig(enabled=False), FakeStore())
check("disabled never starts", m.tick(t0, soc=80).phase == IDLE)

print("CFG-03: config.yaml is the source, this module is the only mirror")
# ⚠️ WHY THIS IS PINNED BY READING THE FILE. Before CFG-03 each maintenance
# default was written down three times - config.yaml's `options:` block,
# `opts.get(key, <literal>)` in main.py, and MaintConfig's dataclass defaults -
# and MAINT-04 proved they had already drifted (2000 shipped, 2500 in code). The
# fix is one source and one mirror; the mirror is only worth anything if drift
# from it is a CI failure rather than a comment, and a transcription of
# config.yaml into this file would just be a fourth place to disagree. So the
# shipped values are READ out of the file the Supervisor reads.
#
# Regex rather than PyYAML on purpose, for test_validate.py's reason: the add-on
# image installs py3-aiohttp and py3-paho-mqtt only, and a check that skips on
# ImportError is not a check.
HERE = Path(__file__).resolve().parent
_CFG_TEXT = (HERE / "config.yaml").read_text(encoding="utf-8")
_SCHEMA_AT = _CFG_TEXT.index("\nschema:")


def shipped_default(key):
    """The value under `options:` - the FIRST match, before `schema:`."""
    m = re.search(rf"^  {key}: *(.+?) *$", _CFG_TEXT[:_SCHEMA_AT], re.M)
    return None if m is None else m.group(1)


def schema_range(key):
    """`int(lo,hi)` under `schema:` -> (lo, hi); None if the key is absent."""
    m = re.search(rf"^  {key}: *\w+\((-?[\d.]+),(-?[\d.]+)\) *$",
                  _CFG_TEXT[_SCHEMA_AT:], re.M)
    return None if m is None else (float(m.group(1)), float(m.group(2)))


def as_number(text):
    if text in ("true", "false"):
        return 1.0 if text == "true" else 0.0
    return float(text)


_ship = {k: shipped_default(k) for k in MAINT_OPTION_DEFAULTS}
print(f"    config.yaml ships {_ship}")
check("every mirrored key is actually shipped in config.yaml's options: block",
      all(v is not None for v in _ship.values()))
check("...and MAINT_OPTION_DEFAULTS reproduces each one exactly, so the mirror "
      "is a copy of the source and not an opinion about it",
      all(str(MAINT_OPTION_DEFAULTS[k]).lower() == _ship[k]
          for k in MAINT_OPTION_DEFAULTS))
# ⚠️ THE OTHER DIRECTION MATTERS TOO. Equality alone would still pass if the
# mirror quietly stopped covering a key someone added to config.yaml, and the
# key nobody mirrors is exactly the key that gets a literal typed for it.
_shipped_keys = set(re.findall(r"^  (maintenance_[a-z_]+):", _CFG_TEXT[:_SCHEMA_AT],
                               re.M))
check(f"...and it covers EVERY maintenance_* option the file ships, so a ninth "
      f"key cannot arrive unmirrored {sorted(_shipped_keys)}",
      _shipped_keys == set(MAINT_OPTION_DEFAULTS))

# ⚠️ WHERE THE RANGE LIVES, ASSERTED RATHER THAN COMMENTED. The PM proofread on
# this card named a FOURTH declaration site of a different kind:
# `maintenance_charge_w: int(500,5000)` in the schema block. It is a BOUND, not
# a default, and the decision this card records is that bounds live in the
# schema block ONLY - the Supervisor enforces them and reads only that file, so
# a Python copy would be a place to disagree and nothing else. What that
# decision needs from a check is (a) every mirrored key really does have a
# schema entry, or the Supervisor drops the option and the add-on silently runs
# on the mirror instead of the operator's value, and (b) the shipped default
# lies INSIDE its own range, because a default the schema rejects is a config
# that cannot be re-entered once it is edited.
_ranged = {k: schema_range(k) for k in MAINT_OPTION_DEFAULTS}
check("every mirrored key is in the schema: block too - `bool` has no range, so "
      "it is the one exception and it is named",
      all(v is not None for k, v in _ranged.items()
          if k != "maintenance_enabled")
      and _ranged["maintenance_enabled"] is None)
_outside = {k: (as_number(_ship[k]), rng) for k, rng in _ranged.items()
            if rng is not None and not rng[0] <= as_number(_ship[k]) <= rng[1]}
check(f"...and every shipped default sits inside its own schema range, so the "
      f"Supervisor accepts the file it ships with {_outside or 'all inside'}",
      not _outside)

# ⚠️ AND THE CONSUMERS MOVE WITH IT - checked as a RELATION, not against
# numbers. AC4 asks for the replacement to be broken and every consumer observed
# moving together. A number typed here would pass whether or not the derivation
# survived; `dataclasses.fields()` reads what the class actually defaulted to,
# so this can only pass while the class is still indexing the mirror.
_fields = {f.name: f.default for f in dataclasses.fields(MaintConfig)}
check(f"MaintConfig's field names are the option names minus the prefix, so a "
      f"new field cannot miss the mirror by being spelt differently {sorted(_fields)}",
      {f"maintenance_{n}" for n in _fields} == set(MAINT_OPTION_DEFAULTS))
check("...and every dataclass default IS the mirror's value - no literal "
      "survives in the class that used to hold the third opinion",
      all(float(d) == float(MAINT_OPTION_DEFAULTS[f"maintenance_{n}"])
          for n, d in _fields.items()))
# from_options() is the only reader, so it is the only thing that can put a
# different number in front of main.py.
_from_empty = MaintConfig.from_options({})
check("...and from_options({}) - the sim/inject.py and test-suite path, the one "
      "the Supervisor never takes and the only one the divergence was reachable "
      "from - equals MaintConfig() exactly",
      _from_empty == MaintConfig())
check("...while a supplied value still wins, or the mirror would be a ceiling "
      "rather than a default",
      MaintConfig.from_options({"maintenance_charge_w": 1234}).charge_w == 1234.0)
# ⚠️ `key in opts`, not truthiness: `false` and `0` are values an operator can
# set, and a truthy read would silently replace both with the default - this
# card's own bug in a different costume.
check("...and an explicit falsey value is honoured rather than defaulted away",
      MaintConfig.from_options({"maintenance_start_hour": 0}).start_hour == 0
      and MaintConfig.from_options({"maintenance_enabled": False}).enabled is False)



# =========================================================================== #
# MAINT-05. TWO PACKS, ONE GRID CONNECTION.
#
# ⚠️ THE FIXTURE RULE THIS REPOSITORY KEEPS PAYING FOR (MUTATION.md habit 4).
# Every two-pack fixture below gives the packs DIFFERENT phases and DIFFERENT
# `last_completed` times, because the bug being hunted is SHARED STATE and two
# equal values cannot show it. `phase("drain") == phase("drain")` is true whether
# the two machines have one history between them or two, and a fixture where
# both packs last completed on the same day cannot tell "the longest-waiting
# pack goes first" from "the first roster entry goes first". SAFETY-30 and
# CFG-04 were each amended for exactly this.
# =========================================================================== #


class MemStore:
    """`app.store.Store`'s surface, no disk, and it keeps keys VERBATIM.

    ⚠️ VERBATIM IS THE WHOLE POINT. A namespaced write has to land under a
    different key from a flat one, and a fake that normalised or ignored the
    prefix would make every check in parts 6-10 pass on a version that has no
    namespacing at all. Times go in and come out the way the real Store stores
    them - ISO-8601 with an offset - so the migration below moves the same kind
    of value a real state file holds.
    """

    def __init__(self, data=None):
        self.data = dict(DEFAULTS, **(data or {}))
        self.writes = []

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

    def get_time(self, k):
        raw = self.data.get(k)
        if not raw:
            return None
        if isinstance(raw, datetime):
            return raw
        parsed = datetime.fromisoformat(raw)
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)

    def set_time(self, k, when):
        self.data[k] = when.astimezone(timezone.utc).isoformat() if when else None
        self.writes.append(k)

    def save(self):
        pass


def two_packs(store=None, mirror=True):
    """Pack A and pack B on ONE store, each under its own namespace."""
    store = store if store is not None else MemStore()
    a = Maintenance(cfg, store, unit="inverter-a", name="inverter-a",
                    mirror_flat=mirror)
    b = Maintenance(cfg, store, unit="inverter-b", name="inverter-b",
                    mirror_flat=False)
    return store, a, b


def drive(machines, now, socs):
    """One site tick: arbitrate once, then tick every machine with ITS pack's
    SoC and the answer arbitration gave it. This is `Controller.cycle()`'s
    three lines, and calling `tick()` without the plan is the defect."""
    plan = arbitrate(machines, now)
    return [m.tick(now, soc, blocked_by=b)
            for m, soc, b in zip(machines, socs, plan)]


# --------------------------------------------------------------------------- #
print()
print("6. per-pack schedule, phase and history (AC1, AC2)")

_t = datetime(2026, 9, 4, 10, 0, tzinfo=timezone.utc)
_store, _a, _b = two_packs()
# ⚠️ DIFFERENT PHASES, DIFFERENT DATES. A is mid-charge and last ran 40 days
# ago; B is idle and last ran 33 days ago. Shared state shows up as either of
# those pairs collapsing into one value.
_a.store.set("phase", CHARGE)
_a.store.set_time("phase_started", _t - timedelta(hours=4))
_a.store.set_time("last_completed", _t - timedelta(days=40))
_b.store.set("phase", IDLE)
_b.store.set_time("last_completed", _t - timedelta(days=33))

print(f"    A: phase={_a.phase} last={_a.store.get_time('last_completed'):%Y-%m-%d}"
      f"   B: phase={_b.phase} last={_b.store.get_time('last_completed'):%Y-%m-%d}")
check("the two packs hold DIFFERENT phases at the same moment, which one FSM "
      "with one history cannot do - and which a fixture with two equal phases "
      "could not have shown",
      _a.phase == CHARGE and _b.phase == IDLE and _a.phase != _b.phase)
check("...and DIFFERENT last-completed dates, so 'A completed' and 'B "
      "completed' are distinguishable statements rather than the same one",
      _a.store.get_time("last_completed") != _b.store.get_time("last_completed")
      and _a.waited_days(_t) > _b.waited_days(_t))
check("...and they live under different Store keys, spelt by the same "
      "`unit_key()` the migration and the watchdog use",
      _store.data.get(unit_key("inverter-a", "phase")) == CHARGE
      and _store.data.get(unit_key("inverter-b", "phase")) == IDLE)

# AC: completing A's cycle must not mark B complete.
_b_before = _b.store.get_time("last_completed")
_a.store.set("phase", HOLD)
_a.store.set_time("phase_started", _t - timedelta(minutes=cfg.hold_min + 1))
_r = drive([_a, _b], _t, [100.0, 50.0])
print(f"    after A's hold expires: A={_r[0].phase} completed "
      f"{_a.store.get_time('last_completed'):%Y-%m-%d %H:%M}, B={_r[1].phase} "
      f"completed {_b.store.get_time('last_completed'):%Y-%m-%d %H:%M}")
check("A's cycle COMPLETES and stamps A's own history",
      _r[0].phase == IDLE
      and _a.store.get_time("last_completed") == _t)
check("⚠️ ...and B's history is untouched by it. This is the card's first "
      "acceptance criterion and the reason the two dates were seeded "
      "differently: with one shared `last_completed`, B would have just been "
      "marked recalibrated without a single watt moving through it",
      _b.store.get_time("last_completed") == _b_before
      and _b.store.get_time("last_completed") != _a.store.get_time("last_completed"))
check("...and B is still DUE, which is the consequence an operator would "
      "actually feel: a pack marked complete by its sibling stops being "
      "scheduled",
      _b.due(_t))


# --------------------------------------------------------------------------- #
print()
print("7. the interlock: one pack moves power at a time (AC3)")

# Both packs due in the same start hour - the case the card says two packs on
# one monthly schedule drift into. ⚠️ B IS THE MORE OVERDUE ONE, deliberately:
# roster order would pick A, so a check that passed with B waiting longer AND A
# winning would be a check that cannot see the arbitration at all.
_store2, _a2, _b2 = two_packs()
_a2.store.set_time("last_completed", _t - timedelta(days=30))
_b2.store.set_time("last_completed", _t - timedelta(days=45))
check("the fixture is discriminating: both packs want to start on this tick, "
      "and the one that has waited LONGER is the one roster order would NOT "
      "pick",
      _a2.wants_start(_t) and _b2.wants_start(_t)
      and _b2.waited_days(_t) > _a2.waited_days(_t))

_plan = arbitrate([_a2, _b2], _t)
_r2 = drive([_a2, _b2], _t, [80.0, 80.0])
print(f"    plan -> A: {_plan[0] or 'GO'} | B: {_plan[1] or 'GO'}")
print(f"    result -> A: {_r2[0].phase} owns={_r2[0].owns_setpoint} | "
      f"B: {_r2[1].phase} owns={_r2[1].owns_setpoint}")
check("⚠️ EXACTLY ONE PACK PROCEEDS. Two DRAINs at once is two power-moving "
      "phases on one grid connection - the circulating power SAFETY-31 exists "
      "to detect - and this is the line that makes it unreachable",
      sum(1 for r in _r2 if r.phase in POWER_MOVING) == 1
      and sum(1 for r in _r2 if r.owns_setpoint) == 1)
check("...and it is the LONGEST-WAITING pack, not the first in the roster. "
      "Roster order would recalibrate inverter-a every month forever and let "
      "inverter-b's coulomb counter drift - this card's own defect in a "
      "different costume",
      _r2[1].phase == DRAIN and _r2[0].phase == IDLE)
check("...and the deferred pack states a reason that NAMES the pack it is "
      "waiting for and the two waits it was compared on, so the arbitration is "
      "auditable rather than a silent skip",
      "inverter-b" in _r2[0].deferred and "45 d" in _r2[0].deferred
      and "30 d" in _r2[0].deferred)

# The mirror image, so the ordering is a comparison and not a constant.
_store3, _a3, _b3 = two_packs()
_a3.store.set_time("last_completed", _t - timedelta(days=45))
_b3.store.set_time("last_completed", _t - timedelta(days=30))
_r3 = drive([_a3, _b3], _t, [80.0, 80.0])
check("...and with the waits SWAPPED the winner swaps too, which is what "
      "distinguishes an arbitration from a hard-wired index",
      _r3[0].phase == DRAIN and _r3[1].phase == IDLE)

# A tie falls to roster order - stated, checked, and the only thing roster
# order is allowed to decide.
_store4, _a4, _b4 = two_packs()
_a4.store.set_time("last_completed", _t - timedelta(days=40))
_b4.store.set_time("last_completed", _t - timedelta(days=40))
_r4 = drive([_a4, _b4], _t, [80.0, 80.0])
check("a genuine numeric tie breaks on roster order, and only a tie does",
      _r4[0].phase == DRAIN and _r4[1].phase == IDLE)

# ⚠️ NEVER RECALIBRATED OUTRANKS ANY FINITE WAIT, and it is checked because
# `waited_days()` returns `inf` there rather than 0 - which is the same
# statement `due()` makes about an absent `last_completed`. A 0 would make a
# brand-new pack the LAST to be recalibrated on every contested month, which is
# the one pack whose coulomb counter has never been trued at all.
_store_new, _a_new, _b_new = two_packs()
_a_new.store.set_time("last_completed", _t - timedelta(days=100))
_r_new = drive([_a_new, _b_new], _t, [80.0, 80.0])
print(f"    A last ran 100 d ago, B has NEVER run -> A: {_r_new[0].phase}, "
      f"B: {_r_new[1].phase}")
check("a pack that has never been recalibrated outranks one that completed 100 "
      "days ago, because 'never' is not a small number - it is the absence of a "
      "coulomb count",
      _b_new.waited_days(_t) > _a_new.waited_days(_t)
      and _r_new[1].phase == DRAIN and _r_new[0].phase == IDLE)

# ⚠️ CHARGE HOLDS THE SLOT TOO. This is the check that bites if CHARGE is
# dropped from POWER_MOVING - the phase where the LOOP owns the setpoint and
# the pack is nonetheless being pulled full from the grid.
_store5, _a5, _b5 = two_packs()
_a5.store.set("phase", CHARGE)
_a5.store.set_time("phase_started", _t - timedelta(hours=2))
_a5.store.set_time("last_completed", _t - timedelta(days=40))
_b5.store.set_time("last_completed", _t - timedelta(days=60))
_r5 = drive([_a5, _b5], _t, [40.0, 80.0])
print(f"    A charging, B due and MORE overdue -> A: {_r5[0].phase}, "
      f"B: {_r5[1].phase} ({_r5[1].deferred[:60]}...)")
check("⚠️ A PACK IN **CHARGE** HOLDS THE SITE'S MAINTENANCE SLOT, even though "
      "the control loop owns the setpoint in that phase. A charging at 2000 W "
      "from the grid while B drains 2000 W to export is 4 kW going round in a "
      "circle with the meter reading zero and both packs believing they are "
      "behaving - which is why CHARGE is in POWER_MOVING",
      _r5[0].phase == CHARGE and _r5[1].phase == IDLE and _r5[1].deferred)
check("...and B is deferred even though it is the MORE overdue pack, because "
      "nothing preempts a cycle in flight: a recalibration stopped half way has "
      "spent the energy and bought no coulomb count",
      _b5.waited_days(_t) > _a5.waited_days(_t) and _r5[1].phase == IDLE)

# HOLD, for the same reason stated differently: it commands 0 W and moves
# nothing, but it owns the actuator and the pack is parked for the BMS.
_store6, _a6, _b6 = two_packs()
_a6.store.set("phase", HOLD)
_a6.store.set_time("phase_started", _t - timedelta(minutes=5))
_a6.store.set_time("last_completed", _t - timedelta(days=40))
_b6.store.set_time("last_completed", _t - timedelta(days=60))
_r6 = drive([_a6, _b6], _t, [100.0, 80.0])
check("a pack in HOLD holds the slot too - it commands 0 W and moves nothing, "
      "and it owns the actuator while the BMS balances",
      _r6[0].phase == HOLD and _r6[1].phase == IDLE and _r6[1].deferred)

# The recovery branch: a store that says BOTH packs are mid-cycle. Only
# reachable from a hand-edited state file or a version with no interlock, and
# the fail-safe direction is unambiguous.
_store7, _a7, _b7 = two_packs()
for _m, _ph in ((_a7, DRAIN), (_b7, DRAIN)):
    _m.store.set("phase", _ph)
    _m.store.set_time("phase_started", _t - timedelta(hours=1))
_r7 = drive([_a7, _b7], _t, [80.0, 80.0])
print(f"    a store that says BOTH are draining -> A: {_r7[0].phase}, "
      f"B: {_r7[1].phase} @ {_r7[1].setpoint_w} W")
check("a store carrying TWO packs mid-cycle is recovered from rather than "
      "obeyed: the first keeps its cycle, the second is stood down to 0 W and "
      "says why",
      _r7[0].phase == DRAIN and _r7[1].phase == IDLE
      and _r7[1].setpoint_w == 0.0
      and any("interlock" in e for e in _r7[1].events))

# The operator's button, and the API a future price-gated start goes through.
_store8, _a8, _b8 = two_packs()
_a8.store.set("phase", DRAIN)
_a8.store.set_time("phase_started", _t)
_ev = _b8.force_start(_t, blocked_by=arbitrate([_a8, _b8], _t)[1])
print(f"    forced start on B while A drains -> {_ev}")
check("a FORCED start is refused while another pack holds the slot, and the "
      "phase machine does not move. The calendar is a preference an operator "
      "may overrule; one pack at a time is a property of the grid connection "
      "and a button does not add a second one",
      _b8.phase == IDLE and len(_ev) == 1 and "REFUSED" in _ev[0]
      and "inverter-a" in _ev[0])
check("...and an unblocked forced start still works, so the refusal is a "
      "condition and not a disabled feature",
      Maintenance(cfg, MemStore(), unit="inverter-b", name="inverter-b")
      .force_start(_t) and True)


# --------------------------------------------------------------------------- #
print()
print("8. the deferred pack is not starved, and is not mistaken for a fault (AC4)")

_store9, _a9, _b9 = two_packs()
_a9.store.set_time("last_completed", _t - timedelta(days=45))
_b9.store.set_time("last_completed", _t - timedelta(days=30))
_r9 = drive([_a9, _b9], _t, [80.0, 80.0])
check("A takes the slot and B is deferred, which is the starting state for "
      "everything below",
      _r9[0].phase == DRAIN and _r9[1].phase == IDLE and _r9[1].deferred)

# ⚠️ THE STARVATION MECHANISM, ASSERTED DIRECTLY. `last_start_attempt` is the
# post-ABORT cooldown (90 min) and the start hour is 60 minutes long, so a
# deferral that stamped it would lock the deferred pack out of the very hour it
# is waiting for - and out of the moment the winner releases the slot.
check("⚠️ A DEFERRAL DOES NOT STAMP `last_start_attempt`. Nothing was attempted "
      "and nothing failed; stamping it would put the deferred pack under a "
      "90 min cooldown inside a 60 min start hour, which is starvation "
      "implemented by accident",
      _b9.store.get_time("last_start_attempt") is None)
check("...and the deferred pack's PHASE does not move either, so "
      "`phase_started` stays empty and neither the 12 h DRAIN nor the 36 h "
      "CHARGE timeout can be reached by waiting. §12's note is explicit that a "
      "phase timeout means 'the battery is not following commands - a FAULT'; "
      "a deferred pack must never raise it",
      _b9.phase == IDLE and _b9.store.get_time("phase_started") is None)
check("...and the words say WAITING, name the pack it waits for, and say it "
      "will run next - none of which any failure path in this module says",
      "WAITING" in _r9[1].events[0]
      and "inverter-a" in _r9[1].events[0]
      and "not a stalled phase" in _r9[1].events[0]
      and "ABORT" not in _r9[1].events[0] and "stuck" not in _r9[1].events[0])
# ⚠️ THE SECOND TICK SAYS SOMETHING NEW AND THE THIRD SAYS NOTHING. On the
# first tick both packs wanted to start and the reason was the comparison that
# settled it; on the second, A is mid-cycle and the reason is that instead - a
# genuinely different sentence, so it is logged. From there it is constant and
# the log falls silent, which is what edge-logging means.
_r9b = drive([_a9, _b9], _t + timedelta(minutes=1), [80.0, 80.0])
_r9c = drive([_a9, _b9], _t + timedelta(minutes=2), [80.0, 80.0])
check("...and it is said once per REASON, not once a second: a wait that lasts "
      "a whole cycle would otherwise write one line per second and flush the "
      "40-line action log of the very context that explains it",
      _r9b[1].deferred and _r9b[1].events and len(_r9b[1].events) == 1
      and _r9c[1].deferred == _r9b[1].deferred and _r9c[1].events == [])
check("...and the reason it changed to is the one that is true now - A is "
      "mid-cycle - rather than the comparison that was true when the slot was "
      "still open",
      "mid-cycle" in _r9c[1].deferred
      and "has waited longer" in _r9[1].deferred)

# Now let A finish, and prove B recalibrates on the NEXT opportunity. This is
# the acceptance criterion that asks for a test rather than a comment.
_when = _t
for _step, _soc_a in ((timedelta(hours=3), 10.0),        # A drains to its floor
                      (timedelta(hours=20), 99.0),       # A charges to target
                      (timedelta(hours=23), 100.0)):     # A's hold expires
    _when = _t + _step
    drive([_a9, _b9], _when, [_soc_a, 80.0])
check("A's cycle runs to completion while B waits",
      _a9.phase == IDLE and _a9.store.get_time("last_completed") is not None)
_next = (_t + timedelta(days=1)).replace(hour=cfg.start_hour)
_r9n = drive([_a9, _b9], _next, [80.0, 80.0])
print(f"    next start hour ({_next:%Y-%m-%d %H:%M}) -> A: {_r9n[0].phase}, "
      f"B: {_r9n[1].phase}")
check("⚠️ AND B RECALIBRATES AT THE NEXT START HOUR. Not a comment claiming it "
      "is not starved - the deferred pack driven through the winner's whole "
      "cycle and observed starting",
      _r9n[1].phase == DRAIN and _r9n[1].owns_setpoint)
check("...and A, which has just completed, does not contend for the slot "
      "again - so the two schedules come out of one contested month STAGGERED, "
      "which is the property the card asks the PR to state",
      _r9n[0].phase == IDLE and not _a9.due(_next))
check("...and B's deferral produced no claim on the tick it was deferred: a "
      "deferred pack owns no setpoint and shapes nothing, so it is simply "
      "under the loop like any idle pack",
      not _r9[1].owns_setpoint and not _r9[1].charge_only)



# --------------------------------------------------------------------------- #
print()
print("9. the store migration: no lost history, and the three rules hold (AC5)")

# A real pre-MAINT-05 single-pack store: a month of history under the flat keys,
# with SAFETY-02's own migration already done (schema_version 1).
def flat_store(**over):
    return MemStore(dict({
        "schema_version": 1,
        "phase": CHARGE,
        "phase_started": "2026-09-03T22:00:00+00:00",
        "last_completed": "2026-08-07T12:00:00+00:00",
        "last_start_attempt": "2026-09-03T10:00:00+00:00",
    }, **over))


_m1 = flat_store()
_rep1 = migrate_maint_keys(_m1, "inverter-a", 2)
print(f"    a single-pack store growing to two: {_rep1.explain()}")
check("every flat maintenance key lands in the FIRST pack's namespace, because "
      "the flat keys can only ever have belonged to the pack the one FSM was "
      "driving",
      _rep1.ran and _rep1.ok and _rep1.verified
      and all(_m1.data.get(unit_key("inverter-a", k)) == _m1.data[k]
              for k in MAINT_UNIT_KEYS))
check("...and the VALUES are the history, not defaults wearing its name - a "
      "migration that lands `last_completed: None` has lost the month it was "
      "moving and would recalibrate a pack that does not need it",
      _m1.data.get(unit_key("inverter-a", "last_completed"))
      == "2026-08-07T12:00:00+00:00"
      and _m1.data.get(unit_key("inverter-a", "phase")) == CHARGE)
check("RULE 2: the flat keys are LEFT IN THE FILE, so an operator who rolls the "
      "add-on back to a single-FSM version finds the history it had",
      all(k in _m1.data for k in MAINT_UNIT_KEYS)
      and _m1.data["last_completed"] == "2026-08-07T12:00:00+00:00")
check("RULE 3: the marker is written and the store is now at the per-pack "
      "schema",
      _m1.data[MAINT_SCHEMA_KEY] == MAINT_SCHEMA_VERSION)
check("...and the pack that has just arrived gets NOTHING, which is the right "
      "answer rather than an omission: absent means never recalibrated, means "
      "due, and a brand-new pack should be",
      not any(k.startswith(unit_key("inverter-b", "")) for k in _m1.data)
      and Maintenance(cfg, _m1, unit="inverter-b").due(_t))

# Rule 1, and it is the rule that makes re-running safe.
_m2 = flat_store(**{unit_key("inverter-a", "last_completed"):
                    "2026-09-01T09:00:00+00:00"})
_rep2 = migrate_maint_keys(_m2, "inverter-a", 2)
check("RULE 1: an EXISTING namespaced value is never overwritten by the flat "
      "one. A store with both spellings has been written by a version that knew "
      "about the namespace, and preferring the flat one would roll a pack's "
      "history back to whatever it was on migration day",
      _m2.data.get(unit_key("inverter-a", "last_completed"))
      == "2026-09-01T09:00:00+00:00"
      and unit_key("inverter-a", "last_completed") in _rep2.kept)
check("...so running the migration TWICE is a no-op, which is what makes a "
      "crash between the copies and the marker safe",
      migrate_maint_keys(_m1, "inverter-a", 2).ran is False)

# The store a multi-inverter version wrote. AC: not touched.
_m3 = MemStore({MAINT_SCHEMA_KEY: MAINT_SCHEMA_VERSION,
                unit_key("inverter-a", "last_completed"): "2026-09-01T09:00:00+00:00",
                unit_key("inverter-b", "last_completed"): "2026-08-20T09:00:00+00:00",
                "last_completed": "2026-09-01T09:00:00+00:00"})
_before3 = dict(_m3.data)
_rep3 = migrate_maint_keys(_m3, "inverter-a", 2)
print(f"    a store already written by a multi-inverter version: {_rep3.explain()}")
check("a store already at the per-pack schema is NOT TOUCHED - not one key "
      "written - and says so",
      not _rep3.ran and _m3.data == _before3 and _m3.writes == []
      and "not touched" in _rep3.explain())

# ⚠️ AND BREAK IT. The read-back postcondition is the whole difference between
# this and a two-line copy, and it is inherited from `migrate_unit_keys()`
# rather than re-implemented - so it has to be shown working HERE.
class LosesTheHistory(MemStore):
    """A store that accepts the namespaced write and drops it on the floor.

    What a partial write, a full disk and a silently-swallowed OSError all look
    like from here: `Store.save()` catches OSError and carries on by design.
    """

    def set(self, k, v):
        if k.startswith("unit/"):
            return
        super().set(k, v)


_m4 = LosesTheHistory(dict(flat_store().data))
_rep4 = migrate_maint_keys(_m4, "inverter-a", 2)
print(f"    a store that silently drops the migrated keys: {_rep4.explain()}")
check("⚠️ THE MIGRATION REPORTS THAT IT FAILED, because it READS BACK what it "
      "wrote. A month of coulomb-count history under a key nothing reads is a "
      "pack that has quietly stopped being maintained on an install whose state "
      "file says it has not",
      _rep4.ran and not _rep4.verified and not _rep4.ok and _rep4.lost)
check("...and it does NOT mark the store migrated, so the next start runs it "
      "again rather than skipping a migration that never happened",
      MAINT_SCHEMA_KEY not in _m4.data)
check("...while a store whose only failed write was a DEFAULT is not reported "
      "as lost state: a fresh install has no `last_completed` at all, and "
      "'absent' and 'never recalibrated' are the same statement - an alarm that "
      "fires on the harmless case is one an operator learns to dismiss",
      migrate_maint_keys(LosesTheHistory({"schema_version": 1}),
                         "inverter-a", 2).ok is True)


# --------------------------------------------------------------------------- #
print()
print("10. a roster of ONE is byte-identical (AC6)")

# ⚠️ THE STATE FILE IS THE ASSERTION, NOT A CLAIM ABOUT IT. A one-inverter
# install must come out of this card with the same keys it went in with: no
# namespace, no marker, no fifth key. `unit=None` is what makes that true and
# this is the check that bites if it stops being passed.
_solo_store = MemStore()
_solo = Maintenance(cfg, _solo_store)
_solo_events = []
_solo_events += _solo.force_start(t0)
for _dt, _soc in ((timedelta(0), 80.0), (timedelta(hours=3), 10.5),
                  (timedelta(hours=20), 99.0), (timedelta(hours=21), 100.0),
                  (timedelta(hours=22, minutes=5), 100.0)):
    _solo_events += _solo.tick(t0 + _dt, _soc).events
print(f"    one pack, one whole cycle -> store keys "
      f"{sorted(k for k in _solo_store.data if k not in DEFAULTS)}")
check("⚠️ A WHOLE CYCLE ON A ROSTER OF ONE WRITES NO KEY THAT DID NOT EXIST "
      "BEFORE THIS CARD. No `unit/...` namespace and no `maint_schema_version` "
      "- the state file is the one an install already has",
      not any(k.startswith("unit/") for k in _solo_store.data)
      and MAINT_SCHEMA_KEY not in _solo_store.data
      and set(_solo_store.data) == set(DEFAULTS))
check("...and the flat keys are the ones that carried the cycle, so 'no new "
      "keys' is not 'no history'",
      _solo_store.data["phase"] == IDLE
      and _solo_store.get_time("last_completed")
      == t0 + timedelta(hours=22, minutes=5))
check("...and `migrate_maint_keys()` refuses to run at all on a one-pack "
      "roster, which is why the layout above stays put",
      migrate_maint_keys(MemStore({"phase": DRAIN}), "inverter-a", 1).ran is False)

# ⚠️ THE REASON STRINGS, PINNED AS LITERALS. The card asks for byte-identical
# behaviour AND reason strings at N=1, and the only way a check can see a
# reworded event is to hold the words. Every one of these is the string the
# shipped module produced before this card; the interlock added no clause to any
# of them, because a single machine is never another machine's blocker.
check(f"the events of a whole one-pack cycle are the shipped strings, "
      f"unchanged {_solo_events}",
      _solo_events == [
          "maintenance: forced start -> drain",
          "maintenance: drain complete at 10 % after 180 min -> charge",
          "maintenance: charge complete at 99 % after 1020 min -> holding "
          "120 min to balance",
          "maintenance: cycle COMPLETE (125 min hold) - handing back to the loop",
      ])
check("...and nothing about a deferral, a slot or another pack appears in any "
      "of them - `arbitrate()` over one machine is the identity function and "
      "no reason string it can build is reachable",
      not any("slot" in e or "WAITING" in e or "REFUSED" in e
              for e in _solo_events)
      and arbitrate([_solo], t0) == [""])

# The abort path's words too, since it is the one an operator reads under
# pressure.
_solo2 = Maintenance(cfg, MemStore())
_solo2.force_start(t0)
check("...and so is the abort, which is the string an operator reads under "
      "pressure",
      _solo2.tick(t0 + timedelta(hours=13), 60.0).events
      == ["maintenance: ABORT (drain stuck at 60 % after 780 min)"])



# --------------------------------------------------------------------------- #
print()
print("11. the wiring: a real Controller, a real cycle, a real claim (AC3, "
      "point 3 of the card)")


class Hass:
    """Answers PER ENTITY, which is the property every check below needs.

    A stub that returned one number whatever it was asked cannot tell "pack B's
    machine was ticked with pack B's SoC" from "it was handed pack A's", and
    that is one of the two defects this part exists to catch.
    """

    def __init__(self, numbers, limits):
        self.numbers = dict(numbers)
        self.limits_by = dict(limits)
        self.writes = []

    async def number(self, entity, invert=False):
        v = self.numbers.get(entity)
        return None if v is None else (-v if invert else v)

    async def limits(self, entity):
        return self.limits_by.get(entity, (None, None, None))

    async def set_number(self, entity, value):
        self.writes.append((entity, value))
        return True

    async def call(self, domain, service, data):
        return True


class Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


# ⚠️ 3000 W FOR PACK B IS A HYPOTHESIS AND IS MARKED AS ONE EVERYWHERE IT
# APPEARS. Inverter B does not physically exist; the number is here only because
# it DIFFERS from pack A's 5000 W, which is the only property these checks use.
B_RATING_HYPOTHESIS = 3000.0
A_RATING = 5000.0
FLAT_OPTS = dict(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
                 batt_entity="sensor.batt_a", meter_source="off",
                 meter_entity="sensor.grid", max_w=A_RATING, gain=0.6,
                 # ⚠️ `slew_w` IS 6000 AND THE METER READS +5000 W FOR ONE
                 # REASON, STATED HERE RATHER THAN TUNED UNTIL GREEN: the
                 # allocator REFUSES a split whose shares have opposite signs,
                 # so a drain pinned at +2000 W beside a site target below
                 # +2000 W would park the whole site at 0 W and these checks
                 # would grade a failsafe instead of a claim. A house importing
                 # 5000 W and a slew that lets one cycle reach it puts the site
                 # target ABOVE the pin, which is the state a real site is in
                 # for most of a drain. The opposite state is a real defect and
                 # it is recorded below rather than tuned away.
                 slew_w=6000, deadband_w=15, target_grid_w=0, step_w=10,
                 heartbeat_s=10, stale_input_s=15,
                 maintenance_discharge_w=2000, maintenance_charge_w=2000)
TWO_OPTS = {k: v for k, v in FLAT_OPTS.items()
            if k not in ("setpoint_entity", "soc_entity", "batt_entity")}
TWO_OPTS["inverters"] = [
    dict(name="inverter-a", setpoint_entity="number.sp_a",
         soc_entity="sensor.soc_a", batt_entity="sensor.batt_a",
         max_w=A_RATING),
    dict(name="inverter-b", setpoint_entity="number.sp_b",
         soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
         max_w=B_RATING_HYPOTHESIS),
]


def controller(opts, store=None, soc_a=80.0, soc_b=80.0, grid=5000.0):
    """A real Controller against a stub HA. ⚠️ SIMULATED - no rig, no device."""
    hass = Hass({"sensor.grid": grid, "sensor.soc_a": soc_a, "sensor.batt_a": 0.0,
                 "sensor.soc_b": soc_b, "sensor.batt_b": 0.0},
                {"number.sp_a": (-A_RATING, A_RATING, 1.0),
                 "number.sp_b": (-B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS, 1.0)})
    c = Controller(dict(opts), hass, store if store is not None else MemStore(),
                   Pub(), stop=asyncio.Event())
    for inv in c.inverters:
        inv.dev_min, inv.dev_max = -inv.max_w, inv.max_w
    c.auto = True
    return c, hass


def one_cycle(c):
    async def go():
        await c.read_inputs()
        await c.cycle()
    asyncio.run(go())
    return c


_now = datetime.now(timezone.utc)

# --- the claim is filed on the pack whose machine produced it ---------------
# ⚠️ PACK **B** IS THE ONE RECALIBRATING, and that is the whole fixture. With
# the claim filed on `maint_unit` - `inverters[0]`, which is what shipped -
# pack A would be pinned at the drain rate while pack B, the pack actually being
# recalibrated, stayed under the loop: the site would move the right total and
# recalibrate nothing.
_cB, _hB = controller(TWO_OPTS)
_cB.maints[1].force_start(_now)
one_cycle(_cB)
print(f"    B recalibrating -> targets A={_cB.inverters[0].target:.0f} W, "
      f"B={_cB.inverters[1].target:.0f} W; wire saw {_hB.writes}")
check("⚠️ THE DRAIN LANDS ON PACK B, THE PACK WHOSE STATE MACHINE ASKED FOR IT",
      _cB.maints[1].phase == DRAIN
      and _cB.inverters[1].target == _cB.maint.cfg.discharge_w
      and ("number.sp_b", _cB.maint.cfg.discharge_w) in _hB.writes)
check("...and pack A is NOT commanded the drain rate - it carries the REST of "
      "the site target, as any unpinned unit does. This is the half that makes "
      "the check discriminating: filing B's procedure on A would move the same "
      "site total and recalibrate the wrong pack, and the site sum would look "
      "identical",
      _cB.inverters[0].target != _cB.maint.cfg.discharge_w
      and _cB.inverters[0].target + _cB.inverters[1].target == _cB.target)
check("...and the claim is a UNIT claim on B's own actuator, never a site one - "
      "ARCH-01's rule, and what structurally prevents the cross-inverter "
      "`lo > hi` contradiction",
      _cB.actuator_of("maintenance", _cB.inverters[1]) == "inverter-b"
      and _cB.actuator_of("maintenance", _cB.inverters[0]) == "inverter-a")

# The mirror image: pack A recalibrating must land on pack A.
_cA, _hA = controller(TWO_OPTS)
_cA.maints[0].force_start(_now)
one_cycle(_cA)
check("...and with pack A recalibrating instead, the drain lands on A and not "
      "on B - so the routing follows the machine rather than a constant",
      _cA.inverters[0].target == _cA.maint.cfg.discharge_w
      and _cA.inverters[1].target != _cA.maint.cfg.discharge_w)

# --- each machine is ticked with ITS OWN pack's SoC -------------------------
# ⚠️ THE TWO SoCs ARE DELIBERATELY ON OPPOSITE SIDES OF THE FLOOR. B is at
# 10.5 % (below `soc_floor` 11) and A is at 90 %, so "B's drain ends" and "A's
# drain continues" are different answers to the same tick - which a fixture with
# two equal SoCs could not have produced.
_cS, _ = controller(TWO_OPTS, soc_a=90.0, soc_b=10.5)
_cS.maints[1].force_start(_now)
one_cycle(_cS)
print(f"    A at 90 %, B at 10.5 %, floor {cfg.soc_floor} % -> B's phase "
      f"{_cS.maints[1].phase}")
check("⚠️ PACK B'S MACHINE IS GRADED ON PACK B'S OWN SoC: at 10.5 % against an "
      "11 % floor its drain is complete and it moves to CHARGE. Handed pack A's "
      "90 % it would still be draining a pack that is already empty",
      _cS.maints[1].phase == CHARGE)
_cS2, _ = controller(TWO_OPTS, soc_a=90.0, soc_b=10.5)
_cS2.maints[0].force_start(_now)
one_cycle(_cS2)
check("...and pack A's machine, on the same site with the same two readings, "
      "stays in DRAIN at 90 % - the same tick, the opposite answer, because the "
      "readings are not interchangeable",
      _cS2.maints[0].phase == DRAIN)

# --- the operator-facing surfaces ------------------------------------------
# ⚠️ TWO DIFFERENT DUE DATES, BECAUSE ONE ROW HAS TO CARRY BOTH SCHEDULES. A
# fixture with two equal dates cannot tell "the earliest across the packs" from
# "the first pack's", which is the whole content of the row.
_cD, _ = controller(TWO_OPTS)
_cD.maints[0].store.set_time("last_completed", _now - timedelta(days=2))
_cD.maints[1].store.set_time("last_completed", _now - timedelta(days=20))
_rows = dict(_cD.status_payload()["rows"])
print(f"    A last ran 2 d ago, B 20 d ago -> due row {_rows['Maintenance due']!r}")
check("the 'Maintenance due' row is the EARLIEST date across the packs, not the "
      "first pack's - one row on a site with two schedules has to be the one "
      "that comes first, or it says October while a pack recalibrates tomorrow",
      _rows["Maintenance due"]
      == (_now + timedelta(days=cfg.interval_days - 20)).strftime("%Y-%m-%d"))
_cD.maints[1].force_start(_now)
one_cycle(_cD)
_rows2 = dict(_cD.status_payload()["rows"])
print(f"    B recalibrating -> banner {_cD.status_payload()['banner']!r}, "
      f"phase row {_rows2['Maintenance phase']!r}")
check("...and at two packs the banner and the phase row NAME the pack, because "
      "the remedy is at one device and the site has two",
      "inverter-b" in _cD.status_payload()["banner"]
      and "inverter-b" in _rows2["Maintenance phase"])
_cD1, _ = controller(FLAT_OPTS)
_cD1.maint.force_start(_now)
one_cycle(_cD1)
_cD1rows = dict(_cD1.status_payload()["rows"])
print(f"    the same state at ONE pack -> banner "
      f"{_cD1.status_payload()['banner']!r}, phase row "
      f"{_cD1rows['Maintenance phase']!r}")
check("...while at a roster of one the banner and the phase row are the bare "
      "phase they have always been, with no pack name appended - which is what "
      "'byte-identical reason strings' means on the surfaces an operator reads",
      _cD1rows["Maintenance phase"] == _cD1.maint.phase
      and "(" not in _cD1rows["Maintenance phase"]
      and _cD1.status_payload()["banner"] == f"Maintenance: {_cD1.maint.phase}")

# --- and the Controller actually WIRES the interlock ------------------------
# ⚠️ THE INTERLOCK CAN BE PRESENT AND NOT CONNECTED, which is the failure this
# check exists for: `arbitrate()` computed and its answer never handed to
# `tick()` is two packs draining with a correct, unread plan sitting beside
# them. A store that comes up with BOTH packs mid-drain is the state that
# exercises it through a REAL cycle rather than through this file's own helper.
_cI, _ = controller(TWO_OPTS)
for _m in _cI.maints:
    _m.store.set("phase", DRAIN)
    _m.store.set_time("phase_started", _now - timedelta(hours=1))
one_cycle(_cI)
print(f"    both packs came up draining -> {[m.phase for m in _cI.maints]}")
check("⚠️ ONE REAL CYCLE LEAVES EXACTLY ONE PACK DRAINING. This is the check "
      "that bites if `Controller.cycle()` computes the plan and does not pass "
      "it to `tick()` - an interlock that is present, correct and not connected",
      [m.phase for m in _cI.maints] == [DRAIN, IDLE])
check("...and the pack that was stood down was commanded 0 W on the wire, not "
      "merely marked idle",
      _cI.inverters[1].target == 0.0)

# --- and at a roster of one, everything above is the shipped object ---------
_c1store = MemStore()
_c1, _ = controller(FLAT_OPTS, store=_c1store)
print(f"    one inverter -> {len(_c1.maints)} machine(s), namespace "
      f"{_c1.maints[0].unit!r}, store keys "
      f"{sorted(k for k in _c1store.data if k not in DEFAULTS)}")
check("a one-inverter Controller builds exactly ONE machine, on the FLAT keys, "
      "with no mirror to make - which is what keeps the state file the one the "
      "install already has",
      len(_c1.maints) == 1 and _c1.maints[0].unit is None
      and _c1.maints[0].store.mirror_flat is False
      and MAINT_SCHEMA_KEY not in _c1store.data)
check("...and `self.maint` is that machine on every tick, so `publish()`, the "
      "status banner and both operator buttons read exactly what they read "
      "before this card",
      _c1.maint is _c1.maints[0]
      and _c1.maint_unit is _c1.inverters[0]
      and _c1._maint_button(_now) is _c1.maints[0])
one_cycle(_c1)
print(f"    after a real one-inverter cycle: "
      f"{sorted(k for k in _c1store.data if k not in DEFAULTS)}")
# ⚠️ THE NAMESPACED KEYS THAT *ARE* THERE ARE DEV-02's SIGN LATCH AND ITS
# MARKER, WRITTEN BY THE MIGRATION THAT SHIPPED A CARD AGO. This card must add
# none of its own, which is the assertion: no maintenance key is namespaced and
# there is no `maint_schema_version` at all.
check("...and a real cycle on it writes no MAINTENANCE key that did not exist "
      "before this card: every one of the four is flat, none is namespaced, and "
      "the per-pack marker is absent",
      not any(k.endswith(tuple("/" + n for n in MAINT_UNIT_KEYS))
              for k in _c1store.data)
      and all(k in _c1store.data for k in MAINT_UNIT_KEYS)
      and MAINT_SCHEMA_KEY not in _c1store.data)

# --- two inverters: A's history is namespaced AND mirrored flat -------------
_c2store = MemStore({"schema_version": 1,
                     "last_completed": "2026-08-07T12:00:00+00:00"})
_c2, _ = controller(TWO_OPTS, store=_c2store)
print(f"    two inverters -> namespaces "
      f"{[m.unit for m in _c2.maints]}, mirrors "
      f"{[m.store.mirror_flat for m in _c2.maints]}")
check("a two-inverter Controller builds one machine per roster entry, each in "
      "its own namespace, and migrates the flat history into the FIRST pack's",
      [m.unit for m in _c2.maints] == ["inverter-a", "inverter-b"]
      and _c2.maints[0].store.get_time("last_completed")
      == datetime(2026, 8, 7, 12, 0, tzinfo=timezone.utc)
      and _c2.maints[1].store.get_time("last_completed") is None)
check("...and ONLY the first pack mirrors back to the flat keys, because the "
      "flat spelling means pack A on every install that has one - a mirror from "
      "B would overwrite A's rollback view with B's phase",
      [m.store.mirror_flat for m in _c2.maints] == [True, False])
_c2.maints[0].force_start(_now)
_c2.maints[1].store.set("phase", HOLD)
check("...and the mirror is LIVE: pack A's new phase reaches the flat key, so a "
      "rollback finds a current history rather than one dated migration day",
      _c2store.data.get(unit_key("inverter-a", "phase")) == DRAIN
      and _c2store.data["phase"] == DRAIN)
check("...while pack B's write reaches only pack B's namespace",
      _c2store.data.get(unit_key("inverter-b", "phase")) == HOLD
      and _c2store.data["phase"] == DRAIN)



# --------------------------------------------------------------------------- #
print()
print("12. RECORDED, NOT FIXED: the site loop still contradicts a drain at N>=2")

# ⚠️ THIS IS A DEFECT THIS CARD FOUND AND DELIBERATELY DID NOT FIX, AND IT IS
# WRITTEN DOWN AS A RUNNING CHECK RATHER THAN AS A COMMENT SO IT CANNOT BE
# FORGOTTEN OR QUIETLY "FIXED" BY A TUNING CHANGE.
#
# WHAT IT IS. The maintenance SET is a per-unit claim and PINS its pack. The
# loop's target is a SITE claim. When the site target is BELOW the pin - a
# drain of 2000 W while the house is importing less than that, which is most of
# a night - the only split that sums to the site target asks the sibling to
# CHARGE while the maintained pack DISCHARGES, and `allocator.plan()` refuses it
# by name as circulating power. The refusal is correct and fails safe: the whole
# site goes to 0 W with an ARBITER line in the action log.
#
# WHY IT IS NOT THIS CARD'S. It is PRE-EXISTING - measured on release/1.0 at
# 2d025ee, where the same drain filed on `inverters[0]` produces the identical
# refusal with the two unit names swapped - and fixing it means changing what
# the site loop's target MEANS while a unit is pinned (regulate the meter using
# only the unpinned packs), which is loop and allocator territory, not
# maintenance's. MAINT-05 moves which pack is pinned and nothing else about it.
#
# ⚠️ AND THE CONSEQUENCE AN OPERATOR WOULD SEE IS A MISDIAGNOSIS, WHICH IS WHY
# IT MATTERS ENOUGH TO PIN. The pack is held at 0 W while its FSM believes it is
# draining, so after 12 h the DRAIN timeout fires and reports "drain stuck" -
# §12's words for "the battery is not following commands, a fault". The battery
# is fine; the site refused to allocate. That is the same class of wrong story
# this card's own notes warn about for the deferral.
_cX, _ = controller(TWO_OPTS, grid=0.0)
_cX.maints[1].force_start(_now)
one_cycle(_cX)
print(f"    house at 0 W, B draining 2000 W -> reason {_cX.reason!r}; "
      f"{_cX.allocation.explain()}")
check("a maintenance drain BELOW the site target is refused by the allocator "
      "and the site parks at 0 W - recorded here so the next card can find it, "
      "not asserted as desirable",
      _cX.target == 0.0 and _cX.allocation.refused
      and "circulating power" in _cX.allocation.explain())
check("...and the pack that is supposedly draining is commanded 0 W, so its "
      "12 h DRAIN timeout would eventually report 'drain stuck' - a fault "
      "message about a healthy battery",
      _cX.maints[1].phase == DRAIN and _cX.inverters[1].target == 0.0)
check("...while the SAME site with the house importing more than the drain "
      "allocates cleanly, which is what makes this a CONDITION rather than "
      "'maintenance does not work at two inverters'",
      _cB.allocation.refused is None and _cB.inverters[1].target == 2000.0)

print()
print()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")