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 logging
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.inverter import UNIT_A
from app.main import Controller
from app.store import (
    DEFAULTS, MAINT_SCHEMA_KEY, MAINT_SCHEMA_VERSION, MAINT_UNIT_KEYS,
    SCHEMA_KEY, SCHEMA_VERSION, migrate_maint_keys, migrate_unit_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)")

# ⚠️ THE ROSTER IS ["garage", "shed"], AND THE FIRST NAME IS THE WHOLE POINT OF
# THIS SECTION'S FIX ROUND. Every check here used to name its packs
# "inverter-a" / "inverter-b" - and "inverter-a" is the string
# `inverter.UNIT_A` holds, so `self.inverters[0].name` and `UNIT_A` were the
# SAME VALUE in every fixture in this file. main.py's migration call site could
# be changed from the first to the second with all 19 suites still green:
# MUTATION.md's third defect exactly, a check whose DATA cannot reach the
# failure it names.
#
# A declared `inverters:` roster names its own entries (`build_roster()`), and
# the fallback for a NAMELESS entry is a POSITIONAL `inverter-1` - so
# `inverter-a` is not the default for a declared roster either. It is a name
# nothing on this site reads, which is what makes migrating into it silent.
ROSTER = ["garage", "shed"]

check("PRECONDITION, WRITTEN AS ITS OWN CHECK. The first pack on this "
      f"section's roster is NOT `UNIT_A`: {ROSTER[0]!r} != {UNIT_A!r}, and "
      f"{UNIT_A!r} is not on it at all. 'the first roster entry' and "
      "'inverter-a' are therefore DIFFERENT strings in every check below - and "
      "if a later edit makes them equal again this goes red rather than the "
      "claims going quietly vacuous",
      ROSTER[0] != UNIT_A and UNIT_A not in ROSTER)


# 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, ROSTER[0], ROSTER)
print(f"    a single-pack store growing to two: {_rep1.explain()}")
print(f"    keys after migration: "
      f"{sorted(k for k in _m1.data if k.startswith('unit/'))}")
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 not _rep1.misdirected
      and all(_m1.data.get(unit_key(ROSTER[0], k)) == _m1.data[k]
              for k in MAINT_UNIT_KEYS))
check("...and the namespace is THE ROSTER'S FIRST NAME, not `UNIT_A` - which is "
      "a claim this file could not make until its packs stopped being called "
      "`inverter-a`. Not one key was written under `unit/inverter-a/`, on a "
      "store where that is the name a wrong call site would have used",
      not any(k.startswith(unit_key(UNIT_A, "")) for k in _m1.data)
      and unit_key(ROSTER[0], "last_completed") in _m1.data)
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(ROSTER[0], "last_completed"))
      == "2026-08-07T12:00:00+00:00"
      and _m1.data.get(unit_key(ROSTER[0], "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(ROSTER[1], "")) for k in _m1.data)
      and Maintenance(cfg, _m1, unit=ROSTER[1]).due(_t))

# Rule 1, and it is the rule that makes re-running safe.
_m2 = flat_store(**{unit_key(ROSTER[0], "last_completed"):
                    "2026-09-01T09:00:00+00:00"})
_rep2 = migrate_maint_keys(_m2, ROSTER[0], ROSTER)
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(ROSTER[0], "last_completed"))
      == "2026-09-01T09:00:00+00:00"
      and unit_key(ROSTER[0], "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, ROSTER[0], ROSTER).ran is False)

# The store a multi-inverter version wrote. AC: not touched.
_m3 = MemStore({MAINT_SCHEMA_KEY: MAINT_SCHEMA_VERSION,
                unit_key(ROSTER[0], "last_completed"): "2026-09-01T09:00:00+00:00",
                unit_key(ROSTER[1], "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, ROSTER[0], ROSTER)
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, ROSTER[0], ROSTER)
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}),
                         ROSTER[0], ROSTER).ok is True)


# --------------------------------------------------------------------------- #
print()
print("9b. THE OTHER HALF OF THE POSTCONDITION: a migration into a namespace "
      "the machines do not read (AC5)")

# ⚠️ WHY THIS EXISTS, IN ONE SENTENCE: A POSTCONDITION THAT READS BACK THE
# NAMESPACE IT WAS HANDED CANNOT DETECT BEING HANDED THE WRONG ONE.
#
# `_verify_moves()` shipped reading back `rep.moved` - the keys it had just
# written - so a migration handed a `unit` that is on no roster wrote four keys,
# read all four back equal, set `verified = True`, and the caller stamped
# `maint_schema_version`. Every symptom of a clean migration, and an install's
# whole maintenance history parked under a name nothing will ever look up. The
# next start then sees the marker and SKIPS the migration, which is the one
# state store.py's header says must never happen: a failed migration a later
# version treats as done.
#
# It is not a hypothetical failure and it is not a hypothetical call site.
# main.py passes `self.inverters[0].name`; `inverter.UNIT_A` is one token away,
# reads correctly, and was indistinguishable from it in every fixture in this
# file until the roster above stopped being called `inverter-a`.
_m5 = flat_store()
_rep5 = migrate_maint_keys(_m5, UNIT_A, ROSTER)
print(f"    handed {UNIT_A!r} on a roster of {ROSTER}: {_rep5.explain()}")
print(f"    keys written: {sorted(k for k in _m5.data if k.startswith('unit/'))}"
      f"; marker {_m5.data.get(MAINT_SCHEMA_KEY, '<<absent>>')!r}")
check("⚠️ THE MIGRATION REFUSES: a namespace that is not on the roster is not a "
      "namespace anything reads, so the four keys landing perfectly is not the "
      "question. `ok` is False and `verified` is False even though every single "
      "read-back matched, which is the difference this postcondition adds",
      _rep5.ran and not _rep5.ok and not _rep5.verified and _rep5.misdirected
      and all(_m5.data.get(unit_key(UNIT_A, k)) is not None
              for k in MAINT_UNIT_KEYS))
check("...and it is reported as a DIFFERENT failure from a dropped write, "
      "because the remedy is different: `lost` is empty and `misdirected` is "
      "not. A store that dropped the write has a storage fault; this store did "
      "exactly as it was told, by a caller that was wrong",
      not _rep5.lost and len(_rep5.misdirected) == len(MAINT_UNIT_KEYS))
check("...and the refusal NAMES the namespace it used AND the roster it was "
      "graded against, rather than merely saying something went wrong - the "
      "operator's next question is 'which name, and which one should it have "
      "been', and a bare `not ok` answers neither",
      "WRONG NAMESPACE" in _rep5.explain()
      and unit_key(UNIT_A, "last_completed") in _rep5.explain()
      and repr(ROSTER) in _rep5.explain())
check("⚠️ AND THE MARKER IS NOT STAMPED, WHICH IS THE WHOLE REMEDY. A stamped "
      "marker makes the next start skip a migration that never reached a "
      "readable key; an unstamped one costs a repeated pass that rule 1 makes a "
      "no-op. Recoverable beats silent",
      MAINT_SCHEMA_KEY not in _m5.data)
_rep5b = migrate_maint_keys(_m5, ROSTER[0], ROSTER)
print(f"    the same store, re-run with the roster's own first name: "
      f"{_rep5b.explain()}")
check("...and 'recoverable' is measured, not asserted: the SAME store re-run "
      "with the roster's own first name migrates cleanly, and the pack reads "
      "back the month of history it started with. Rule 2 left the flat keys in "
      "the file, which is what there was to recover FROM",
      _rep5b.ran and _rep5b.ok
      and _m5.data.get(unit_key(ROSTER[0], "last_completed"))
      == "2026-08-07T12:00:00+00:00"
      and _m5.data[MAINT_SCHEMA_KEY] == MAINT_SCHEMA_VERSION)
check("...and the misdirected keys are LEFT WHERE THEY LANDED rather than "
      "deleted: this function moves history, and a migration that starts "
      "removing keys it did not put there is a second way to lose a month",
      _m5.data.get(unit_key(UNIT_A, "last_completed"))
      == "2026-08-07T12:00:00+00:00")

# The counterfactual, so this is not a check that passes on a validator that
# refuses everything: the SAME store, the SAME call, one argument different.
_m6 = flat_store()
_rep6 = migrate_maint_keys(_m6, ROSTER[0], ROSTER)
check("...and the counterfactual: one argument different - the roster's own "
      "first name instead of `UNIT_A` - and the identical call is accepted and "
      "stamped. So the refusal above is about the NAME, not about this fixture "
      "being refused whatever it is handed",
      _rep6.ran and _rep6.ok and not _rep6.misdirected
      and _m6.data[MAINT_SCHEMA_KEY] == MAINT_SCHEMA_VERSION)
check("...and a roster of ONE still returns before any of this, so the "
      "namespace postcondition cannot reach a single-pack install: nothing is "
      "moved, nothing is graded, nothing is stamped",
      migrate_maint_keys(MemStore({"phase": DRAIN}), UNIT_A, [UNIT_A]).ran
      is False)


# --------------------------------------------------------------------------- #
print()
print("9c. THE REFUSAL HAS TO SURVIVE THE NEXT START: the same wrong call "
      "site, twice, over ONE store")

# ⚠️ WHY 9b WAS NOT ENOUGH, AND IT IS THIS FILE'S OWN RECURRING DEFECT ONE
# LEVEL UP. 9b's recovery check re-runs the migration with `ROSTER[0]` - the
# CORRECTED name - so it measures recovery along the one path where the
# operator has ALREADY fixed the call site. The real failure is a call site that
# is STILL WRONG on the next start, and until this section nothing in the 19
# files called `migrate_maint_keys()` twice with the SAME wrong `unit`. 9b was
# true and passed for the wrong reason.
#
# WHAT THAT HID. Measured at Controller level over one persisted store, roster
# ["garage", "shed"], flat `last_completed` 2026-08-07, call site handing
# `inverter-a` on both starts:
#
#   construction 1: ok=False misdirected=4 moved=4 kept=0  marker=<absent>
#   construction 2: ok=True  misdirected=0 moved=0 kept=4  marker=1
#
# `_verify_moves()` graded only `rep.moved`. On start 2 the wrong call site
# finds the keys it wrote already present, rule 1 routes all four to `rep.kept`,
# `rep.moved` is EMPTY, the roster loop iterates over nothing, `misdirected` is
# empty and `ok` is True - so the marker is stamped and start 3 takes the
# `from_version >= MAINT_SCHEMA_VERSION` early return FOREVER, including after
# the call site is fixed. Loud-and-recoverable became silent-and-permanent after
# ONE restart. The fix grades `set(rep.moved) | set(rep.kept)`.
_m7 = flat_store()
_rep7a = migrate_maint_keys(_m7, UNIT_A, ROSTER)
_rep7b = migrate_maint_keys(_m7, UNIT_A, ROSTER)
print(f"    start 1: ok={_rep7a.ok} misdirected={len(_rep7a.misdirected)} "
      f"moved={len(_rep7a.moved)} kept={len(_rep7a.kept)} marker="
      f"{_m7.data.get(MAINT_SCHEMA_KEY, '<<absent>>')!r}")
print(f"    start 2: ok={_rep7b.ok} misdirected={len(_rep7b.misdirected)} "
      f"moved={len(_rep7b.moved)} kept={len(_rep7b.kept)} marker="
      f"{_m7.data.get(MAINT_SCHEMA_KEY, '<<absent>>')!r}")
check("PRECONDITION, SO THIS SECTION CANNOT PASS FOR THE WRONG REASON: start 2 "
      "goes down the KEPT path, which is the path no other check in this file "
      "reaches. Not one key is written a second time - all four are already "
      "there from start 1, so `moved` is EMPTY and `kept` holds every one of "
      "them. If a later edit makes start 2 re-WRITE them, this goes red rather "
      "than the claim below quietly reverting to 9b's",
      _rep7b.ran and _rep7b.moved == {}
      and sorted(_rep7b.kept) == sorted(unit_key(UNIT_A, k)
                                        for k in MAINT_UNIT_KEYS))
check("⚠️ THE MARKER IS STILL WITHHELD ON THE SECOND START. This is the whole "
      "section: a refusal that lasts exactly one start is not a refusal, it is "
      "a delay - the next start stamps the marker on a migration that has "
      "reached no readable key, and every start after that skips the migration "
      "entirely",
      MAINT_SCHEMA_KEY not in _m7.data)
check("...and it is refused for the SAME reason as the first time, graded "
      "against the same roster: four keys under a namespace nothing on this "
      "site reads. `ok` False, `misdirected` full, `lost` empty - the caller "
      "was wrong, the store was not",
      not _rep7b.ok and not _rep7b.verified and not _rep7b.lost
      and len(_rep7b.misdirected) == len(MAINT_UNIT_KEYS)
      and repr(ROSTER) in _rep7b.explain())
check("...and the history is still THERE to recover: rule 2 left every flat key "
      "in the file across both starts, and the roster's first pack still reads "
      "nothing - which is what makes this loud rather than lost",
      all(_m7.data[k] is not None for k in MAINT_UNIT_KEYS)
      and _m7.data["last_completed"] == "2026-08-07T12:00:00+00:00"
      and unit_key(ROSTER[0], "last_completed") not in _m7.data)
_rep7c = migrate_maint_keys(_m7, ROSTER[0], ROSTER)
print(f"    start 3, call site FIXED: {_rep7c.explain()}")
check("⚠️ AND BECAUSE THE MARKER WAS NEVER STAMPED, THE FIXED CALL SITE STILL "
      "GETS A RUN. A stamped marker would have made this an early return "
      "forever - the operator fixes the bug and the migration never happens "
      "again. Here the third start migrates the month into `garage` and stamps",
      _rep7c.ran and _rep7c.ok
      and _m7.data.get(unit_key(ROSTER[0], "last_completed"))
      == "2026-08-07T12:00:00+00:00"
      and _m7.data[MAINT_SCHEMA_KEY] == MAINT_SCHEMA_VERSION)

# The counterfactual, so this is not a check satisfied by a validator that
# refuses every second call: the same two calls with the RIGHT name.
_m7ok = flat_store()
_rep7ok_a = migrate_maint_keys(_m7ok, ROSTER[0], ROSTER)
_rep7ok_b = migrate_maint_keys(_m7ok, ROSTER[0], ROSTER)
check("...and the counterfactual: the SAME two calls with the roster's own "
      "first name are accepted - stamped on the first, and the second is the "
      "`already at schema` early return, not a refusal. So 9c is about the "
      "NAME surviving into the second start, not about second calls",
      _rep7ok_a.ok and _m7ok.data[MAINT_SCHEMA_KEY] == MAINT_SCHEMA_VERSION
      and _rep7ok_b.ran is False and "already at schema" in _rep7ok_b.explain())


# --------------------------------------------------------------------------- #
print()
print("9d. THE OTHER GUARD ON THE SAME LINE: a migration that LOST STATE is "
      "not stamped either, even when some keys were already in place")

# ⚠️ WHAT THIS PINS, AND IT IS A SURVIVOR THE SECTIONS ABOVE CANNOT SEE.
# `migrate_maint_keys()` gates the marker on `rep.ok`, which is BOTH guards:
# `not lost and not misdirected`. Every check in this file that reaches the
# marker reaches it through `misdirected` (9b, 9c) or through a store where
# NOTHING was already namespaced (part 9's `LosesTheHistory`, whose `kept` is
# empty). So the mutation
#
#     if rep.ok:            ->    if rep.ok or rep.kept:
#
# stamped the marker on a migration that lost a month of history - defeating the
# OLDER, settled `lost` guard as well as the new `misdirected` one - and all 19
# suites stayed green on it. Measured differential on this exact fixture:
# shipped -> marker absent; mutant -> marker 1, both reporting ok=False lost=2.
#
# The fixture is the one shape that separates them: TWO keys already namespaced
# (so `kept` is non-empty and the mutant's `or` fires) and TWO keys whose write
# the store drops (so `lost` is non-empty and `ok` is False). `misdirected` is
# empty throughout - the unit handed in IS the roster's first name - so this
# section cannot pass on the `misdirected` guard by accident.
_m8 = LosesTheHistory(dict(flat_store(**{
    unit_key(ROSTER[0], "phase"): CHARGE,
    unit_key(ROSTER[0], "phase_started"): "2026-09-03T22:00:00+00:00"}).data))
_rep8 = migrate_maint_keys(_m8, ROSTER[0], ROSTER)
print(f"    two keys already namespaced, two dropped on write: {_rep8.explain()}")
print(f"    ok={_rep8.ok} lost={len(_rep8.lost)} kept={len(_rep8.kept)} "
      f"misdirected={len(_rep8.misdirected)}; marker "
      f"{_m8.data.get(MAINT_SCHEMA_KEY, '<<absent>>')!r}")
check("PRECONDITION: this fixture reaches the marker through the `lost` guard "
      "and ONLY that one - two keys in `kept`, two in `lost`, `misdirected` "
      "empty. If it ever stops having a non-empty `kept` the mutation below "
      "goes invisible again, so the shape is asserted rather than assumed",
      _rep8.ran and len(_rep8.kept) == 2 and len(_rep8.lost) == 2
      and not _rep8.misdirected and not _rep8.ok)
check("⚠️ THE MARKER IS NOT STAMPED ON A MIGRATION THAT LOST A MONTH OF "
      "HISTORY, and the two keys that DID land do not buy it one. `rep.ok` is "
      "the gate and nothing weaker: stamping here would tell the next start a "
      "migration finished that dropped half of what it was moving",
      MAINT_SCHEMA_KEY not in _m8.data)
check("...and the counterfactual, one property different: the SAME fixture on a "
      "store that does not drop writes IS stamped. So the refusal above is "
      "about the state that was lost, not about `kept` being non-empty",
      migrate_maint_keys(
          MemStore(dict(flat_store(**{
              unit_key(ROSTER[0], "phase"): CHARGE,
              unit_key(ROSTER[0], "phase_started"):
                  "2026-09-03T22:00:00+00:00"}).data)),
          ROSTER[0], ROSTER).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}), UNIT_A, [UNIT_A]).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


# ⚠️ SPIKE-04a (2026-09-04) ESTABLISHED B'S REAL RATING FROM ITS DATASHEET:
# a GoodWe GW5048D-ES, serial `55048ESU22AW0992`, 4600 W on the battery path
# (goodwe/NOTES.md §1.1b). THIS FIXTURE DELIBERATELY DOES NOT USE IT. 3000 W is
# kept because the only property these checks consume is that B's number is
# UNMISTAKABLY not A's, and the real pair is 5000/4600 - a 400 W gap, narrower
# than `slew_w` and inside the range where an off-by-one-unit bug and ordinary
# control arithmetic produce indistinguishable numbers. Collapsing the fixture
# onto the real values would cost these checks their separating power to buy
# realism they do not use. So: ILLUSTRATIVE, NOT B'S RATING, and not a
# placeholder for it either - the placeholder era ended on 2026-09-04.
B_RATING_HYPOTHESIS = 3000.0
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: the first pack's history is namespaced AND mirrored flat -
# ⚠️ THE ROSTER HERE IS ["garage", "shed"] AND THAT IS THE FIX ROUND. Run on
# `TWO_OPTS`, whose entries are named `inverter-a` / `inverter-b`, this whole
# block is satisfied by a Controller that migrates into `UNIT_A` instead of into
# `self.inverters[0].name` - the two strings are equal there, so the check could
# not reach the failure it names (MUTATION.md, third defect). On an operator's
# real roster they are not equal, and `inverter-a` is not even what a nameless
# entry falls back to (`build_roster()` numbers those positionally). The names
# below are the reviewer's: a house with a garage pack and a shed pack.
NAMED_OPTS = {k: v for k, v in TWO_OPTS.items() if k != "inverters"}
NAMED_OPTS["inverters"] = [
    dict(name="garage", setpoint_entity="number.sp_a",
         soc_entity="sensor.soc_a", batt_entity="sensor.batt_a",
         max_w=A_RATING),
    # ⚠️ 3000 W IS THE MARKED HYPOTHESIS AGAIN - inverter B does not exist.
    dict(name="shed", setpoint_entity="number.sp_b",
         soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
         max_w=B_RATING_HYPOTHESIS),
]
_C2_NAMES = [e["name"] for e in NAMED_OPTS["inverters"]]
check("PRECONDITION for the migration checks below: this roster's first entry "
      f"is {_C2_NAMES[0]!r}, which is NOT `UNIT_A` ({UNIT_A!r}), and `UNIT_A` "
      "is not on the roster at all. Every claim under it about 'the first "
      "roster entry's namespace' is therefore a claim a wrong constant would "
      "fail",
      _C2_NAMES[0] != UNIT_A and UNIT_A not in _C2_NAMES)

# ⚠️ THE DATE IS RELATIVE TO `_now`, NOT A LITERAL. A pack's history is only
# observable through `due()` if the interval has not run out on its own: an
# absolute date drifts towards `interval_days` as this file ages and the day it
# crosses, "history preserved" and "history lost" stop being different answers -
# which is the same degenerate-fixture defect one level up.
_C2_LAST = _now - timedelta(days=3)
_c2store = MemStore({"schema_version": 1,
                     "last_completed": _C2_LAST.isoformat()})
_c2, _ = controller(NAMED_OPTS, store=_c2store)
print(f"    roster {_C2_NAMES} -> namespaces "
      f"{[m.unit for m in _c2.maints]}, mirrors "
      f"{[m.store.mirror_flat for m in _c2.maints]}")
print(f"    flat last_completed {_C2_LAST:%Y-%m-%d} (3 d ago, interval "
      f"{cfg.interval_days} d) -> migration: {_c2.maint_migration.explain()}")
print(f"    namespaced keys now: "
      f"{sorted(k for k in _c2store.data if k.startswith('unit/') and not k.endswith(('sign_tripped', 'sign_reason')))}")
check("a two-inverter Controller builds one machine per roster entry, each in "
      "its own namespace, and migrates the flat history into the FIRST ROSTER "
      "ENTRY'S - `garage`, the name the config gave it, not a constant",
      [m.unit for m in _c2.maints] == _C2_NAMES
      and _c2.maints[0].store.get_time("last_completed") == _C2_LAST
      and _c2.maints[1].store.get_time("last_completed") is None)
check("⚠️ AND NOT ONE MAINTENANCE KEY IS WRITTEN UNDER `unit/inverter-a/`. That "
      "is the namespace a call site handing the migration `UNIT_A` would use, "
      "and on this roster nothing reads it - so a store carrying one is an "
      "install whose whole maintenance history has been parked out of sight",
      not any(k.startswith(unit_key(UNIT_A, "")) for k in _c2store.data)
      and _c2store.data.get(unit_key(_C2_NAMES[0], "last_completed"))
      == _C2_LAST.isoformat())
check("⚠️ AND THE PACK IS NOT DUE, WHICH IS WHAT 'NO LOST HISTORY' MEANS TO AN "
      "OPERATOR (AC5). It recalibrated 3 days ago and the interval is "
      f"{cfg.interval_days} days. A migration into a namespace this pack does "
      "not read leaves it reading NOTHING - absent means never recalibrated, "
      "means due now - so this one line separates a preserved month from a lost "
      "one",
      _c2.maints[0].store.get_time("last_completed") is not None
      and not _c2.maints[0].due(_now))
check("...and the migration REPORTS clean and stamps the marker on this roster, "
      "so 'not due' above is a migration that landed rather than one that was "
      "refused",
      _c2.maint_migration.ok and _c2.maint_migration.verified
      and not _c2.maint_migration.misdirected
      and _c2store.data.get(MAINT_SCHEMA_KEY) == MAINT_SCHEMA_VERSION)
check("...and ONLY the first pack mirrors back to the flat keys, because the "
      "flat spelling means the first roster entry on every install that has "
      "one - a mirror from the second would overwrite its rollback view",
      [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: the first pack'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(_C2_NAMES[0], "phase")) == DRAIN
      and _c2store.data["phase"] == DRAIN)
check("...while the second pack's write reaches only its own namespace",
      _c2store.data.get(unit_key(_C2_NAMES[1], "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("13. FIXED (SAFETY-44): SAFETY-02's migration takes the roster too, and "
      "the guard is no longer defeated by the key the defect wrote")

# ⚠️ THIS SECTION USED TO BE THE SPECIFICATION OF A DEFECT. It was written while
# fixing MAINT-05's own instance, as a RUNNING check rather than a comment, with
# the instruction that "the card that fixes it should expect them to go red and
# should replace them". SAFETY-44 is that card and this is that replacement.
#
# WHAT IT WAS. `Controller.__init__` called `migrate_unit_keys(store, UNIT_A)` -
# a hardcoded constant, where the maintenance migration already passed the
# roster's own first name. On a DECLARED roster the pre-DEV-02 flat
# `sign_tripped: true` landed under `unit/inverter-a/sign_tripped`, which no
# unit on the site reads, and `_verify_moves()` read it back under the name it
# had been handed and was satisfied.
#
# AND THE GUARD DID NOT CATCH IT, WHICH WAS THE PART WORTH PINNING.
# `_grade_store_migration()` re-arms when the flat key says tripped and NOTHING
# NAMESPACED IS LATCHED - and the misdirected write IS namespaced and IS
# latched. The guard was defeated by the very key the defect wrote.
#
# ⚠️ THE REASON IT WAS LEFT - "there is no roster to name yet, because each
# `Inverter` restores its latch in its constructor" - WAS THE HALF-TRUTH THAT
# KEPT IT ALIVE. The keys must be written before the `Inverter`s read them; the
# NAMES do not come from the `Inverter`s. `roster_names(opts)` reads them off
# the options before anything is constructed, which is what it already existed
# to do for `MqttPublisher`. The factoring this section said the fix would need
# had been done a card earlier.
#
# ⚠️ NO `schema_version` IN THIS STORE, AND THAT IS THE PRE-DEV-02 SHAPE
# ITSELF: an install that has never run a version with namespaced keys, which is
# what makes `migrate_unit_keys()` run at all.
_lat_store = MemStore({"sign_tripped": True,
                       "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured"})
_lat, _ = controller(NAMED_OPTS, store=_lat_store)
print(f"    pre-DEV-02 flat latch + declared roster {_C2_NAMES} -> "
      f"{_lat.store_migration.explain()}")
print(f"    latch keys in the store: "
      f"{sorted(k for k in _lat_store.data if k.endswith('sign_tripped'))}")
print(f"    roster came up tripped: "
      f"{[(i.name, i.sign_wd.tripped) for i in _lat.inverters]}")
check("⭐ the flat pre-DEV-02 latch is migrated into the ROSTER'S OWN first "
      "namespace and NOT into `unit/inverter-a/` - the same wrong-namespace "
      "write `migrate_maint_keys()` refuses, now refused in the migration that "
      "runs before the `Inverter` objects exist",
      _lat_store.data.get(unit_key(_C2_NAMES[0], "sign_tripped")) is True
      and unit_key(UNIT_A, "sign_tripped") not in _lat_store.data)
check("⭐ AND AN INVERTER ON THE ROSTER COMES UP LATCHED, which is the whole "
      "card: a store whose file reads `sign_tripped: true` no longer produces "
      "a site that drives",
      _lat.inverters[0].sign_wd.tripped
      and _lat_store.data["sign_tripped"] is True)
check("...and the latch that came back carries the REASON, not just the "
      "boolean - an operator meeting a FAILSAFE banner with no sentence under "
      "it has been told the site stopped and not why",
      "SIGN_DISAGREEMENT" in _lat.inverters[0].sign_wd.reason)
check("⚠️ ...and the OTHER units on the roster are NOT latched by it. The flat "
      "key is one site-level boolean and re-arming every unit off it would be "
      "DEV-02's defect in the opposite direction - a healthy pack parked "
      "because its sibling once disagreed",
      not any(i.sign_wd.tripped for i in _lat.inverters[1:]))

# ⚠️ AND THE OTHER PATH, WHICH THE MIGRATION FIX ALONE DOES NOT REACH: a store
# ALREADY STAMPED by the shipped version. `migrate_unit_keys()` returns at its
# `from_version >= SCHEMA_VERSION` check and never moves anything, so the
# stranded `unit/inverter-a/sign_tripped` just sits there. This is the STAGED
# shape - "upgrade now, declare the roster later" - and it is the one an
# operator actually produces. Only the GUARD repairs it.
_staged_store = MemStore({
    "sign_tripped": True,
    "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    unit_key(UNIT_A, "sign_tripped"): True,
    unit_key(UNIT_A, "sign_reason"): "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    SCHEMA_KEY: SCHEMA_VERSION})
_staged, _ = controller(NAMED_OPTS, store=_staged_store)
print(f"    already-stamped store + declared roster -> ran="
      f"{_staged.store_migration.ran}, came up tripped: "
      f"{[(i.name, i.sign_wd.tripped) for i in _staged.inverters]}")
check("⚠️ PRECONDITION: the migration really does NOT run on this store - the "
      "marker is already there - so anything below is the guard's doing and "
      "not the migration's",
      not _staged.store_migration.ran)
check("⭐ and the roster's first unit STILL comes up latched: the guard no "
      "longer counts a latch stranded under `UNIT_A`, which is not on this "
      "roster, as protection an operator has",
      _staged.inverters[0].sign_wd.tripped)
check("...and the re-armed latch is written where a machine reads it, so the "
      "NEXT start finds it without needing the guard again",
      _staged_store.data.get(unit_key(_C2_NAMES[0], "sign_tripped")) is True)

# ⚠️ THE NEGATIVE CONTROL FOR THAT NARROWING, because "off the roster" on its
# own was too blunt and test_inverter.py's DEV-02 restart case caught it: a
# latch stranded under a name that is NOT `UNIT_A` is a real unit out of
# today's config, and it MUST still count as protection.
_other_store = MemStore({
    "sign_tripped": True,
    "sign_reason": "SIGN_DISAGREEMENT: B disagreed, mirrored flat",
    unit_key("inverter-b", "sign_tripped"): True,
    unit_key("inverter-b", "sign_reason"): "SIGN_DISAGREEMENT: B disagreed",
    SCHEMA_KEY: SCHEMA_VERSION})
_other, _ = controller(NAMED_OPTS, store=_other_store)
print(f"    stranded latch under `inverter-b` -> came up tripped: "
      f"{[(i.name, i.sign_wd.tripped) for i in _other.inverters]}")
check("⚠️ a latch stranded under a name that is not `UNIT_A` is NOT re-armed "
      "onto the roster: it belongs to a unit the operator has taken out of "
      "the config, the flat key is its own mirror, and re-arming a healthy "
      "sibling off it is the defect DEV-02 removed",
      not any(i.sign_wd.tripped for i in _other.inverters))

# ⭐ AND THE ARRANGEMENT NEITHER FIXTURE ABOVE HOLDS: **BOTH** KEYS AT ONCE.
# ⚠️ THIS IS INSTANCE #17 OF HABIT 4 AND IT WAS FOUND IN THIS CARD'S OWN FIX.
# The first version of the guard classified each key and then wrote ONE shared
# `namespaced_latched` flag for both "on the roster" and "off-roster but not
# UNIT_A". `_staged_store` above holds only an `inverter-a` latch and
# `_other_store` below holds only an `inverter-b` one, so the flag was never
# written twice with opposite intents in a single store - and a per-key check
# whose keys have never disagreed WITHIN ONE STORE is a check over whichever
# key happened to come last. Measured on that build, one added key:
#
#   sign_tripped + unit/inverter-a/sign_tripped   -> garage latched, repaired
#   ...AND unit/old-shed/sign_tripped             -> ⚠️ NOTHING latched
#
# A site DRIVING on a state file that reads `sign_tripped: true` - the sentence
# this card exists to make false - reached by an operator taking one unit out of
# `inverters:` after it had tripped. "A roster is configuration and it changes"
# is this module's own phrase, and the second inverter goes in this month.
_both_store = MemStore({
    "sign_tripped": True,
    "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    unit_key(UNIT_A, "sign_tripped"): True,
    unit_key(UNIT_A, "sign_reason"): "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    unit_key("old-shed", "sign_tripped"): True,
    unit_key("old-shed", "sign_reason"): "SIGN_DISAGREEMENT: old-shed, since removed",
    SCHEMA_KEY: SCHEMA_VERSION})
_both, _ = controller(NAMED_OPTS, store=_both_store)
print(f"    stranded `inverter-a` AND an unrelated off-roster `old-shed` -> "
      f"{[(i.name, i.sign_wd.tripped) for i in _both.inverters]}")
check("⭐ A STRANDED `inverter-a` LATCH IS STILL REPAIRED WHEN AN UNRELATED "
      "OFF-ROSTER LATCH SITS BESIDE IT. One shared flag let `old-shed` veto the "
      "repair, and the site came up driving on a file that says it is latched",
      _both.inverters[0].sign_wd.tripped)
check("...and the repair still lands on the roster's first unit, not on the "
      "removed one",
      _both_store.data.get(unit_key(_C2_NAMES[0], "sign_tripped")) is True)
check("⚠️ ...and the reason carried across is the FLAT one - the misdirected "
      "latch's own - and not `old-shed`'s, which belongs to a different "
      "physical inverter",
      "old-shed" not in _both.inverters[0].sign_wd.reason)

# ⚠️ AND THE OPERATOR MESSAGE IS ASSERTED, because review deleted it whole
# (`if stranded:` -> `if False:`) and 28/28 stayed green. `stranded` has no
# other consumer, so the list existed only to print a line nothing checked -
# and the line is the entire operator-facing half of this repair: it names the
# key that was stranded and the roster it was measured against. A repair an
# operator cannot see is a repair they cannot confirm.
_log_store = MemStore({
    "sign_tripped": True,
    "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    unit_key(UNIT_A, "sign_tripped"): True,
    unit_key(UNIT_A, "sign_reason"): "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured",
    SCHEMA_KEY: SCHEMA_VERSION})
_records = []
_h = logging.Handler()
_h.emit = lambda r: _records.append(r.getMessage())
logging.getLogger("goodwe").addHandler(_h)
try:
    _logged, _ = controller(NAMED_OPTS, store=_log_store)
finally:
    logging.getLogger("goodwe").removeHandler(_h)
_s44_lines = [m for m in _records if "SAFETY-44" in m]
print(f"    SAFETY-44 operator lines logged: {len(_s44_lines)}")
check("⭐ the repair SAYS SO, at ERROR, naming SAFETY-44 - deleting this line "
      "left the whole suite green, and it is how an operator learns their "
      "latch was re-armed rather than never lost",
      len(_s44_lines) == 1)
check("...and it names the STRANDED KEY, so the operator can find it in the "
      "state file rather than being told something happened",
      any(unit_key(UNIT_A, "sign_tripped") in m for m in _s44_lines))
check("...and the ROSTER it was measured against, which is the half that says "
      "WHY no machine read it",
      any("garage" in m and "shed" in m for m in _s44_lines))

# ⭐ AND `misdirected` IS MEASURED AT THE FUNCTION'S OWN LEVEL, BECAUSE IT IS
# UNREACHABLE FROM THE CALL SITE. `Controller.__init__` passes
# `roster_names(opts)[0]`, which `build_roster()` also names its first entry
# from, so `unit` is a member of `roster` by construction and the refusal can
# never fire there. Review measured that: dropping the `roster` argument from
# `_verify_moves()` inside `migrate_unit_keys()` survives 28/28. The argument is
# kept for the reason `_verify_moves()` gives for its own `roster=None` branch -
# it is what a NEW caller gets by default - and a guard nothing exercises is a
# guard nobody knows works, so it is exercised HERE rather than claimed.
_mis_store = MemStore({"sign_tripped": True,
                       "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured"})
_mis = migrate_unit_keys(_mis_store, UNIT_A, ["garage", "shed"])
print(f"    migrate_unit_keys(store, 'inverter-a', ['garage','shed']) -> "
      f"ok={_mis.ok} misdirected={sorted(_mis.misdirected)}")
check("⭐ a WRONG `unit` - one that is not on the roster it was handed - comes "
      "back `misdirected`, which is the refusal `migrate_maint_keys()` already "
      "had and this one did not",
      bool(_mis.misdirected) and not _mis.ok)
check("...and the schema marker is NOT stamped, so the migration runs again on "
      "the next start rather than being recorded as finished - the one thing "
      "store.py's header says must never happen",
      _mis_store.data.get(SCHEMA_KEY) != SCHEMA_VERSION)
check("⚠️ ...while the SAME call with a unit that IS on the roster stamps the "
      "marker - so the refusal is caused by the name and not by the fixture",
      migrate_unit_keys(
          MemStore({"sign_tripped": True, "sign_reason": "x"}),
          "garage", ["garage", "shed"]).ok)

# The counterfactual, and it is the accepted baseline habit 7 asks for: the
# same store, the same migration, one thing different - the roster shape.
_flat_lat_store = MemStore({"sign_tripped": True,
                            "sign_reason": "SIGN_DISAGREEMENT: SYNTHETIC FIXTURE, not measured"})
_flat_lat, _ = controller(FLAT_OPTS, store=_flat_lat_store)
print(f"    the same store on the FLAT config -> roster "
      f"{[(i.name, i.sign_wd.tripped) for i in _flat_lat.inverters]}")
_flat_lat_ok = (all(i.sign_wd.tripped for i in _flat_lat.inverters)
                and _flat_lat_store.data.get(unit_key(UNIT_A, "sign_tripped"))
                is True)
check("...while the SAME store on the FLAT config - the shape where `UNIT_A` "
      "really is the roster's only name - migrates and re-arms correctly. So "
      "this is a defect of the DECLARED roster, not of the migration as such, "
      "and that is the counterfactual that says which half is broken",
      _flat_lat_ok)

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