goodwe-addon/goodwe_controller/test_rehearse.py

"""Runnable check for the roster rehearsal.  `python3 test_rehearse.py`

ARCH-02. `tools/rehearse_roster.py` is the instrument that decides whether the
flat -> roster switch on 2026-09-12 loses a safety latch, a month of maintenance
history, or a Home Assistant entity that an automation is watching. This file
grades the instrument.

⚠️ AN INSTRUMENT THAT HAS NEVER BEEN SEEN TO GO RED IS NOT EVIDENCE. This
repository's most-repeated defect, `MUTATION.md` habit 4 and now FIFTEEN
recorded instances, is "a check whose DATA cannot reach the failure it names" -
and a rehearsal that reports "migration OK" because it compared a structure to
itself is the purest example of it there is. So this file does two things in
order, and the first one is the one that makes the second mean anything:

  Part 2  TWO CLEAN CONTROLS. Two (state file, roster) pairs on which the
          shipped code is CORRECT and the harness must report NOTHING. A
          harness that is red on every input has measured nothing, and every
          red below it would be free.

  Part 3  ⭐ SIX DELIBERATE MUTATIONS OF THE SHIPPED MIGRATION AND OF THE
          ENTITY-ID COMPOSITION, each applied to a private copy of the tree,
          each required to flip its control from green to red, AND each
          required to produce the SPECIFIC finding it should - not merely a
          non-zero exit. A mutation that goes red for the wrong reason is a
          harness that would go red on the real defect for the wrong reason
          too.

  Part 4  TWO MUTATIONS THAT DELIBERATELY SURVIVE, with the reason written
          down. A battery in which everything dies is a battery whose author
          stopped as soon as the number looked good - and the `object_id` one
          is what BOUNDS the four `app/mqtt.py` mutations above it: without a
          change to that file that must NOT be caught, "the reference notices
          any edit to mqtt.py" and "the reference notices what HA keys on" are
          the same green.

  Part 5b THE PER-LEG CHECK'S OWN POSITIVE CONTROL: a build in which exactly
          ONE of the two legs is clean. ⚠️ A per-thing check that has never
          seen the things DISAGREE is a total with extra steps, and this file
          shipped that defect twice - see `legs()`.

  Part 5  SAFETY-44's exact path, pinned as a finding, by name, in both
          directions.

⚠️ THE MUTATIONS ARE SOURCE EDITS IN A COPIED TREE, NOT MONKEYPATCHES. A
monkeypatch grades the name a module was asked for; an edit grades the code that
runs. Each copy is made with `mkdtemp()`, so two runs of this file cannot
overwrite each other's work - habit 18 - and `__pycache__` is never copied, so
no mutant is ever graded through a stale `.pyc` - habit 1 / TEST-15.
"""

import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
FIX = HERE / "tools" / "rehearsal"

fails = []


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


# --------------------------------------------------------------------------- #
print("1. the harness's own preconditions")
# --------------------------------------------------------------------------- #
sys.path.insert(0, str(HERE / "tools"))
import rehearse_roster as RR  # noqa: E402

check("the tool refuses rather than passing when it cannot run: its three exit "
      "codes are distinct and 2 is not 0",
      (RR.EXIT_OK, RR.EXIT_FINDING, RR.EXIT_COULD_NOT_RUN) == (0, 1, 2))

# ⚠️ THE ANCHOR. `app/mqtt.py` records `sensor.goodwe_p1_sample_age` as VERIFIED
# against the live system on 2026-08-26, and the ESP32's ingest guard subscribes
# to it. The tool's entity-id model is written independently of `mqtt._slug` on
# purpose (habit 4), so this is the one external fact that says the independent
# model is right. If it fails, every orphan claim the tool makes is worthless
# and the tool exits 2 rather than reporting.
try:
    RR._check_anchor()
    anchored = True
except RR.Refused:
    anchored = False
check("the entity-id model reproduces the ONE id verified against the live "
      f"system ({RR.ANCHOR_ID})", anchored)

check("...and it is NOT built out of `mqtt._slug`, so a mutation in the "
      "shipped slugifier moves one side of the comparison and not both",
      "_slug" not in RR.compose.__code__.co_names)

_opts = RR.load_options(str(FIX / "roster.yaml"))
check("the built-in YAML reader parses the roster fixture the check image "
      "cannot parse with PyYAML (there is no `yaml` module in it)",
      [e["name"] for e in _opts["inverters"]] == ["garage", "shed"]
      and _opts["max_w"] == 5000 and _opts["batt_invert"] is False
      and _opts["meter_source"] == "off")
try:
    RR._mini_yaml("inverters: [a, b]\n")
    _refused = False
except RR.Refused:
    _refused = True
check("...and it REFUSES a construct it does not understand rather than "
      "guessing, because a config silently mis-parsed here rehearses a "
      "configuration nobody is going to deploy", _refused)

_flat = RR.load_options(str(FIX / "flat.yaml"))
check("⚠️ the two configs differ in the ONE way the transition is about: the "
      "BEFORE roster is empty and the AFTER roster is declared - if they did "
      "not, every diff below would be a structure compared to itself",
      _flat.get("inverters") == [] and len(_opts["inverters"]) == 2)
check("...and the AFTER roster's first entry is NOT `inverter-a`, which is "
      "habit 4's seventh instance in one line: with it, `UNIT_A` and "
      "`inverters[0].name` are the same string and a misdirected namespace "
      "cannot be seen",
      _opts["inverters"][0]["name"] != RR.UNIT_A)

_legacy = json.loads((FIX / "state-legacy.json").read_text())
check("⚠️ the synthetic state carries MAINTENANCE HISTORY - state built by the "
      "happy path cannot detect a migration that drops it",
      _legacy.get("last_completed") and _legacy.get("phase"))
check("⚠️ ...and a LEGACY-SHAPED KEY WHOSE LOSS WOULD BE SILENT: a LATCHED "
      "`sign_tripped` with no `schema_version` beside it, i.e. written by a "
      "controller that had never heard of unit namespaces",
      _legacy.get("sign_tripped") is True
      and "schema_version" not in _legacy
      and "maint_schema_version" not in _legacy)


# --------------------------------------------------------------------------- #
print()
print("2. THE CLEAN CONTROLS: three transitions on which the shipped code is "
      "right and the harness must say nothing")
# --------------------------------------------------------------------------- #
# ⚠️ THIS PART IS WHAT MAKES PART 3 EVIDENCE. Six reds prove nothing about an
# instrument that is red on its own fixtures. Each control below is a genuine
# flat -> roster switch that the shipped migration handles correctly, and the
# required answer is EXIT 0. CONTROL-C earned its place by failing: it found the
# harness reporting a defect on every fresh install.
#
#   CONTROL-A  a PRE-DEV-02 flat store, latch tripped, onto a roster whose
#              first entry IS `inverter-a`. This is the path SAFETY-44's
#              documented reasoning describes, and on it that reasoning holds.
#   CONTROL-B  a store that has already run flat under 0.7.0 (schema stamped,
#              no latch), onto a roster named `garage`/`shed`. This is the
#              MAINT-05 half, and the names are deliberately not `inverter-a`.
#   CONTROL-C  a FRESH INSTALL - an empty state file - onto `garage`/`shed`.
#              ⚠️ THIS ONE FOUND A DEFECT IN THE HARNESS. A fresh store's
#              migration writes `sign_tripped = False` and `sign_reason = ""`
#              under `inverter-a` before any roster exists, and the first draft
#              reported both as STRANDED - red on every fresh install. Losing a
#              DEFAULT loses nothing; that is `MigrationReport.lost`'s own
#              distinction, and an alarm that fires on the harmless case is one
#              an operator learns to dismiss before the real one arrives.
CONTROLS = {
    "CONTROL-A": ("state-legacy.json", "roster-inverter-a.yaml"),
    "CONTROL-B": ("state-flat-070.json", "roster.yaml"),
    "CONTROL-C": ("state-fresh.json", "roster.yaml"),
}


def run(tree: Path, state: str, after: str, *extra) -> "tuple[int, str]":
    """The harness, as the owner runs it, over `tree`.

    ⚠️ ALWAYS WITH `--live-entities`. That file is the only part of the
    entity-id comparison that does not move with the code, so a run without it
    is a run that cannot see a renamed device.
    """
    fixtures = tree / "tools" / "rehearsal"
    proc = subprocess.run(
        [sys.executable, str(tree / "tools" / "rehearse_roster.py"),
         "--state", str(fixtures / state),
         "--before", str(fixtures / "flat.yaml"),
         "--after", str(fixtures / after),
         "--live-entities", str(fixtures / "live-entities-recorded.txt"),
         "--reference", str(fixtures / "announced-reference.txt"),
         *extra],
        capture_output=True, text=True,
        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"})
    return proc.returncode, proc.stdout + proc.stderr


def mutant_tree(edits) -> str:
    """A private copy of the add-on with `edits` applied.

    ⚠️ EVERY EDIT IS ASSERTED TO HAVE MATCHED. Four defects in this session came
    from `str.replace` calls whose pattern did not match: the call silently
    returns the original string, the mutation never happens, and the harness
    reports a SURVIVOR on code it never changed. That is a fabricated result.
    """
    tmp = tempfile.mkdtemp(prefix="rehearse-mut-")
    dest = Path(tmp) / "tree"
    ignore = shutil.ignore_patterns("__pycache__", "*.pyc")
    (dest / "tools").mkdir(parents=True)
    shutil.copytree(HERE / "app", dest / "app", ignore=ignore)
    shutil.copytree(FIX, dest / "tools" / "rehearsal", ignore=ignore)
    shutil.copyfile(HERE / "tools" / "rehearse_roster.py",
                    dest / "tools" / "rehearse_roster.py")
    for rel, old, new in edits:
        path = dest / rel
        text = path.read_text(encoding="utf-8")
        hits = text.count(old)
        if hits != 1:
            raise AssertionError(
                f"mutation target appears {hits} times in {rel}, expected 1: "
                f"{old!r} - the mutation did NOT happen and any result from it "
                f"would be fabricated")
        path.write_text(text.replace(old, new), encoding="utf-8")
    return str(dest)


def legs(out: str) -> dict:
    """The report, split into its legs on the tool's `LEG <id>` markers.

    ⚠️ THIS FUNCTION IS THE FIX FOR A DEFECT IN THIS FILE, AND THE DEFECT IS
    HABIT 4 ONE LEVEL UP. Part 5's check that the disarmed latch is "NOT THE
    SAME-STEP PATH ALONE" was written as
    `out.count("LATCH CAME UP DISARMED") >= 2`. The tool prints every finding
    TWICE per red leg - once in that leg's own FINDINGS block and once in the
    closing summary - so ONE red leg yields exactly 2 and the check passes.
    Measured on three builds:

        shipped code, both legs red                            count 4
        fix at main.py:250 only        -> STAGED alone red     count 2  ok
        partial guard fix on rep.ran   -> SAME-STEP alone red  count 2  ok

    The check named "NOT THE SAME-STEP PATH ALONE" reported `ok` on the build
    whose remaining failure IS the same-step path alone - and, worse, applying
    the fix DOCS.md names as the cause left all five part-5 checks green. A
    count over a whole report cannot distinguish one leg from two; only reading
    the legs separately can.

    ⚠️ AND IT CLOSES ON `END LEG`, WHICH IS THE SAME DEFECT ONE LEVEL DOWN.
    The first version of this splitter opened on `LEG <id>` and never closed,
    so every line after the LAST marker went to that leg - including the
    closing summary, which repeats every finding from every leg. The last leg
    then read red whenever ANY leg was red. Measured on a build whose SAME-STEP
    leg is genuinely clean: it still reported SAME-STEP disarmed. Part 5b is
    the running control for exactly that.
    """
    out_legs, current, closed = {}, None, set()
    for line in out.splitlines():
        if line.startswith("  LEG "):
            # ⚠️ AND THIS ASSERTION IS THE ONE THAT KILLS THE CLASS. Review
            # attacked part 5b directly and got through in three edits: delete
            # `END LEG` (5b catches it), ALSO reorder the legs (5b goes blind -
            # it only ever saw them disagree with the clean leg LAST), and ALSO
            # reword the closing summary (the backstop is a transcription of the
            # tool's own sentence, so rewording blinds that too). Result: "all
            # checks passed" with a splitter that closed ZERO legs.
            #
            # Two of those three are ordinary refactors - reordering report legs
            # and rewording a summary line - neither of which reads as touching
            # a safety check. A fixture cannot cover every arrangement, so the
            # STRUCTURE is asserted instead: a leg may not open while one is
            # open, and every leg that opens must be explicitly closed. Under
            # all three attacks nothing is ever closed, so this fires on the
            # FIRST edit regardless of order or wording.
            assert current is None, (
                f"leg {line.strip()[4:].strip()!r} opened while {current!r} was "
                "still open - the tool stopped closing its legs, and a splitter "
                "that never closes swallows the closing summary into the last "
                "leg. That is the defect this function exists to fix.")
            current = line.strip()[4:].strip()
            out_legs[current] = []
        elif line.startswith("  END LEG "):
            assert current is not None, "END LEG without an open leg"
            closed.add(current)
            current = None
        elif current is not None:
            out_legs[current].append(line)
    assert current is None, f"leg {current!r} was never closed"
    assert closed == set(out_legs), (
        f"legs opened {sorted(out_legs)} but closed {sorted(closed)} - a leg "
        "that is never closed absorbs everything after it")
    return {k: "\n".join(v) for k, v in out_legs.items()}


for label, (state, after) in CONTROLS.items():
    code, out = run(HERE, state, after)
    check(f"{label}: the shipped code, {state} -> {after}, exits 0 with no "
          f"findings", code == RR.EXIT_OK and "NO FINDINGS" in out)

# ⚠️ AND THE SOFTENING IS NOT A HOLE. The SAME key stranded with a NON-default
# value is still a finding - that is the whole of part 5 - so the two cases are
# separated by the VALUE and not by the key name.
_fresh_code, _fresh_out = run(HERE, "state-fresh.json", "roster.yaml")
# ⚠️ THE NAMESPACE IS READ OUT OF THE TOOL'S OWN REPORT, NOT SPELLED HERE. A
# hardcoded `unit/inverter-a/...` makes this check fail for the RIGHT reason on
# the shipped code and for the WRONG reason on a build that fixes the migration
# to use the roster's own name - it would look like the default-softening broke
# when what changed is the namespace. What this check is about is the VALUE.
_stranded_default = set(re.findall(
    r"^\s+stranded\s+(unit/\S+/sign_tripped) = False$", _fresh_out, re.M))
check("a fresh install DOES strand a `.../sign_tripped` key - it is written "
      "and nothing on the roster reads it - and the report SAYS SO under its "
      f"own heading (found: {sorted(_stranded_default)})",
      _fresh_code == RR.EXIT_OK and len(_stranded_default) == 1)
check("...and does NOT call it a finding, because the value is the DEFAULT; "
      "the same key at True is the finding part 5 turns on, so the two cases "
      "are separated by the VALUE and not by the key name",
      not any(f"STRANDED STORE KEY: {k}" in _fresh_out
              for k in _stranded_default)
      and "STRANDED STORE KEY: unit/inverter-a/sign_tripped = True"
      in run(HERE, "state-legacy.json", "roster.yaml")[1])

# --------------------------------------------------------------------------- #
print()
print("2b. ⚠️ OMITTING THE INSTALL'S ENTITY LIST IS A REFUSAL, NOT A QUIETER "
      "RUN")
# --------------------------------------------------------------------------- #
# ⚠️ THIS PART EXISTS BECAUSE THE ASYMMETRY WAS BACKWARDS AND IT WAS MEASURED.
# A `--live-entities` file that could not be READ refused with exit 2; the flag
# simply not being PASSED produced "NO FINDINGS" and exit 0 - and
# `grep -ci "live-entities|recorded list|composition"` over that report returned
# ZERO, so nothing in the artifact an owner reads said half the check had not
# run. Measured on the two entity mutations at the time:
#
#     M5 device rename   with the flag: exit 1     without: exit 0
#     M6 sensor rename   with the flag: exit 1     without: exit 0
#
# The owner runs this ONCE, at commissioning, under time pressure. A forgotten
# flag must not hand back a clean bill of health for the half of the transition
# that orphans entities.
def raw(*args):
    """`(exit code, THE REPORT, the log)` - and the split matters.

    ⚠️ THE REPORT IS STDOUT AND THE CONTROLLER'S LOGGING IS STDERR. Grading
    "is the degraded banner near the end of the report" over the two
    CONCATENATED reads the last lines of the LOG - `SAFETY-02 was TRIPPED...`,
    `unparseable timestamp...` - and reports a missing banner that is in fact
    printed. That is not a hypothetical; it is how this check first failed.
    """
    fx = HERE / "tools" / "rehearsal"
    proc = subprocess.run(
        [sys.executable, str(HERE / "tools" / "rehearse_roster.py"),
         "--state", str(fx / "state-legacy.json"),
         "--before", str(fx / "flat.yaml"),
         "--after", str(fx / "roster-inverter-a.yaml"), *args],
        capture_output=True, text=True,
        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"})
    return proc.returncode, proc.stdout, proc.stderr


_no_code, _no_out, _no_err = raw()
check("omitting --live-entities REFUSES with exit 2 rather than reporting a "
      "clean transition it did not check",
      _no_code == RR.EXIT_COULD_NOT_RUN and "REFUSED" in _no_out + _no_err)

_deg_code, _deg_out, _ = raw("--no-live-entities")
check("...and the explicit opt-out RUNS, so the escape hatch exists",
      _deg_code in (RR.EXIT_OK, RR.EXIT_FINDING))
check("⚠️ ...carrying a standing banner in the report HEADER and again in "
      "the closing block, because a report is skimmed from the bottom",
      _deg_out.count("DEGRADED RUN") >= 2
      and "DEGRADED RUN" in "\n".join(
          _deg_out.strip().splitlines()[-4:]))
check("...and the degraded report is not silent about what it did not do - "
      "the words are in it, which is the grep that returned 0 before",
      sum(w in _deg_out.lower() for w in
          ("live-entities", "orphan check", "not graded")) == 3)
check("contradictory flags are refused rather than one silently winning",
      raw("--no-live-entities", "--live-entities",
          str(FIX / "live-entities-recorded.txt"))[0]
      == RR.EXIT_COULD_NOT_RUN)
check("...and the banner really is at the END of the report, which is the "
      "half of that claim a concatenated stdout+stderr could not have graded",
      "DEGRADED RUN" in "\n".join(_deg_out.strip().splitlines()[-4:]))

# ⚠️ AND THE COMPOSITION HALF IS NOT BEHIND THE FLAG AT ALL ANY MORE. The
# reference is loaded on EVERY run, so a degraded run still catches a renamed
# device - which is what makes `--no-live-entities` a genuinely bounded
# concession rather than "half the tool off".
_ref_missing = raw("--no-live-entities", "--reference", "/nonexistent.txt")
check("a missing or malformed reference is exit 2 on every run, because it is "
      "the only reference in the tool that does not move with the code",
      _ref_missing[0] == RR.EXIT_COULD_NOT_RUN)

# ⚠️ AND THE VANISHED CHECK CARRIES THE SAME DEFAULT/NON-DEFAULT RULE, BOUNDED
# THE SAME WAY. `M9` below drops a NON-default key and must still be a finding;
# without it, "vanished at the default is not a finding" could widen into
# "vanished is not a finding" and nobody would see it.
_m9 = mutant_tree([("app/store.py",
                    "        value = data.get(legacy, DEFAULTS.get(legacy))\n"
                    "        store.set(target, value)",
                    "        value = data.get(legacy, DEFAULTS.get(legacy))\n"
                    "        store.data.pop(legacy, None)\n"
                    "        store.set(target, value)")])
_m9_code, _m9_out = run(Path(_m9), "state-legacy.json", "roster-inverter-a.yaml")
_m9_vanished = set(re.findall(r"VANISHED STORE KEY: (\S+) =", _m9_out))
print(f"      deleting the flat key as it migrates -> exit {_m9_code}, "
      f"vanished {sorted(_m9_vanished)}")
check("⭐ a NON-default key that vanishes is STILL a finding - a migration "
      "made to DELETE the flat rollback view store.py rule 2 requires it to "
      "leave in place is caught, so the default-softening above is bounded by "
      "its own positive",
      _m9_code == RR.EXIT_FINDING
      and {"last_completed", "phase_started", "last_start_attempt"}
      <= _m9_vanished)
check("...and `phase` is NOT among them, because its value 'idle' IS the "
      "default - the same mutation, the same file, and the rule splitting the "
      "two on the VALUE rather than on the key",
      "phase" not in _m9_vanished)
shutil.rmtree(Path(_m9).parent, ignore_errors=True)

# ⚠️ AND `M12` BOUNDS THE **MOVED-NOT-LOST** RULE, WHICH NOTHING DID. The
# suppression added with SAFETY-44 cites its own bound in a comment, and the
# first version of that comment cited M3 - which drops `sign_reason` from
# `LEGACY_UNIT_KEYS` and expects `LATCH KEY NOT UNDER A ROSTER NAME` - and the
# commit message cited it too. It is the wrong mutant, and so is M9: every key
# M9 vanishes is FLAT, so neither can satisfy the `unit/` gate the rule sits
# behind. THE BRANCH HAD NO NEGATIVE CONTROL AT ALL, in a card whose whole
# subject is checks that cannot reach the failure they name.
#
# ⚠️ AND THE HOLE IT LEFT WAS REAL. `DEFAULTS["sign_tripped"] is False`, so the
# ONLY non-default value a latch can hold is `True` - which means, before the
# `vanished_unit not in after.units` clause, ANY surviving latched roster unit
# suppressed ANY other unit's vanished latch. Measured on a build that drops a
# roster unit's `sign_*` while a sibling's survives at the same value: exit 0,
# "nothing vanished, nothing stranded", on a deleted SAFETY-02 latch.
_M12_OLD = 'tripped=bool(data.get(self.key("sign_tripped"), False)),'
_M12_NEW = ('tripped=bool(data.get(self.key("sign_tripped"), False))' + chr(10)
            + '                if self.name != "shed" else False,')
_m12 = mutant_tree([("app/inverter.py", _M12_OLD, _M12_NEW)])
_m12_code, _m12_out = run(Path(_m12), "state-legacy.json", "roster.yaml")
print(f"      a roster unit that stops restoring its latch -> exit {_m12_code}")
check("⭐ M12: a ROSTER unit's latch that stops being read is still a finding - "
      "the moved-not-lost rule suppresses a RENAME (the key left a name this "
      "roster does not carry and arrived under one it does), never a LOSS",
      _m12_code == RR.EXIT_FINDING)
check("...and the suppression is not what swallowed it: a key whose own unit is "
      "ON the roster is never reported as migrated, whatever else carries the "
      "same value",
      "migrated  unit/shed/" not in _m12_out)
shutil.rmtree(Path(_m12).parent, ignore_errors=True)

# ⚠️ AND THE INPUT FILE IS NEVER WRITTEN. The owner runs this against a copy of
# a live state file; a tool that edited it would be a tool that had to be
# trusted rather than one that cannot do harm.
_before_bytes = (FIX / "state-legacy.json").read_bytes()
run(HERE, "state-legacy.json", "roster.yaml", "--reverse", "--round-trip")
check("the state file the harness was pointed at is BYTE-IDENTICAL afterwards, "
      "including after the reverse and round-trip legs, which write",
      (FIX / "state-legacy.json").read_bytes() == _before_bytes)


# --------------------------------------------------------------------------- #
print()
print("3. ⭐ THE MUTATIONS: each one breaks the shipped migration or the "
      "entity-id composition, and each must turn its control RED")
# --------------------------------------------------------------------------- #
MUTATIONS = [
    # (id, control, what it breaks, edits, the finding it must produce)
    ("M1", "CONTROL-B",
     "the maintenance migration is handed the hardcoded `UNIT_A` instead of "
     "the roster's own first name - MAINT-05's exact shipped regression, which "
     "left all 19 suites green",
     [("app/main.py",
       "store, self.inverters[0].name, [inv.name for inv in self.inverters])",
       "store, UNIT_A, [inv.name for inv in self.inverters])")],
     "MAINTENANCE HISTORY LOST"),

    ("M2", "CONTROL-B",
     "the maintenance namespace is DROPPED: the migration moves no keys at all, "
     "so a month of history stays under a flat spelling the new packs do not "
     "read",
     [("app/store.py",
       "_move_flat_keys(store, unit, MAINT_UNIT_KEYS, rep)",
       "_move_flat_keys(store, unit, (), rep)")],
     "MAINTENANCE HISTORY LOST"),

    ("M3", "CONTROL-A",
     "the SAFETY-02 pair loses HALF of itself: `sign_reason` drops out of "
     "`LEGACY_UNIT_KEYS`, so the latch migrates without the sentence that says "
     "why it tripped",
     [("app/store.py",
       'LEGACY_UNIT_KEYS = ("sign_tripped", "sign_reason")',
       'LEGACY_UNIT_KEYS = ("sign_tripped",)')],
     "LATCH KEY NOT UNDER A ROSTER NAME"),

    ("M4", "CONTROL-A",
     "the READER's key composition changes while the migration's does not - "
     "`Inverter.key()` upper-cases the unit - so the latch is written "
     "perfectly, read back perfectly by the migration, and looked for under a "
     "name that is not there",
     [("app/inverter.py",
       "return unit_key(self.name, name)",
       "return unit_key(self.name.upper(), name)")],
     "LATCH CAME UP DISARMED"),

    ("M5", "CONTROL-A",
     "the entity-id composition changes on the DEVICE half: the MQTT device is "
     "renamed, which is the 'tidy it up' change `app/mqtt.py` spends thirty "
     "lines in capitals refusing. ⚠️ THE BEFORE/AFTER DIFF IS BLIND TO THIS - "
     "both sides are composed by the mutant, so every id moves equally and "
     "nothing looks orphaned. Only the recorded list catches it",
     [("app/mqtt.py", '"name": "GoodWe",', '"name": "GoodWe Controller",')],
     "COMPOSITION DRIFT"),

    ("M6", "CONTROL-A",
     "the entity-id composition changes on the ENTITY half: one site sensor is "
     "renamed, which orphans exactly one id and leaves the other seven alone",
     [("app/mqtt.py",
       '("grid", "goodwe_grid_power", "Grid power", "W", "power", "measurement", None),',
       '("grid", "goodwe_grid_power", "Grid power W", "W", "power", "measurement", None),')],
     "COMPOSITION DRIFT: this build announces sensor.goodwe_grid_power_w"),

    # ⚠️ M7 AND M8 ARE THE TWO CHANGES THAT ORPHAN AN INSTALL WITHOUT MOVING A
    # SINGLE COMPOSED entity_id, so an entity_id-only reference cannot see them.
    # Both came back exit 0 before the reference carried three columns.
    ("M7", "CONTROL-A",
     "the `unique_id` composition changes and NOT the entity_id: `mqtt.BASE` "
     "is renamed, so every announced unique_id moves while every composed "
     "entity_id stays byte-identical. HA's MQTT entity registry keys on "
     "unique_id, so this is a NEW entity beside a dead one - and discovery is "
     "retained with no removal payload, so the dead one stays",
     [("app/mqtt.py", 'BASE = "goodwe_ctl"', 'BASE = "goodwe_controller"')],
     "UNIQUE_ID DRIFT"),

    ("M8", "CONTROL-A",
     "the DEVICE IDENTIFIER changes: HA's device registry keys on it, so the "
     "whole install moves onto a new device and the old one keeps its dead "
     "entities. Again no composed entity_id moves",
     [("app/mqtt.py", '"identifiers": ["goodwe_rs485_controller"],',
       '"identifiers": ["goodwe_rs485_ctl_v2"],')],
     "DEVICE IDENTIFIER DRIFT"),
]

for mid, control, what, edits, must_say in MUTATIONS:
    state, after = CONTROLS[control]
    tree = mutant_tree(edits)
    code, out = run(Path(tree), state, after)
    red = code == RR.EXIT_FINDING
    named = must_say in out
    print()
    print(f"  --- {mid} ({control}) {what}")
    print(f"      exit {code} (0=clean 1=finding 2=could-not-run)")
    # ⚠️ THE FINDINGS, DE-DUPLICATED AND IN ORDER. The tool prints each finding
    # once per transition leg and once more in its summary, and its own header
    # is full of ⚠️ paragraphs - printing every line that starts with one would
    # bury the evidence this part exists to show under the tool's docstring.
    seen = []
    for line in out.splitlines():
        text = line.strip()
        # ⚠️ `_` IS IN THE CLASS. Without it `UNIQUE_ID DRIFT:` does not match
        # its own finding kind, and M7 - the mutation that renames `mqtt.BASE`
        # - printed its verdict with NO evidence under it. A filter that
        # silently drops the very finding the mutation exists to produce is a
        # small version of the same defect this file is about.
        if text.startswith("⚠️ ") and re.match(r"⚠️ [A-Z][A-Z _-]+[A-Z()]:",
                                                 text):
            if text not in seen:
                seen.append(text)
    for text in seen:
        print(f"      {text[:160]}")
    check(f"{mid} turns {control} RED", red)
    check(f"{mid} says WHY, and it is the right why: {must_say!r}", named)
    shutil.rmtree(Path(tree).parent, ignore_errors=True)


# --------------------------------------------------------------------------- #
print()
print("4. THE MUTATION THAT SURVIVES, ON PURPOSE, WITH THE REASON WRITTEN DOWN")
# --------------------------------------------------------------------------- #
# ⚠️ A BATTERY IN WHICH EVERYTHING DIES IS A BATTERY SOMEBODY STOPPED WRITING
# AS SOON AS THE NUMBER LOOKED GOOD. `store.unit_key()` is the ONE spelling of
# the namespaced key, read by `Inverter.key()`, by `UnitHistory.key()` and by
# both migrations - that is store.py's stated design, "one spelling, in one
# place". Changing it therefore moves the writer AND every reader together, and
# the store comes out internally consistent under a different name.
#
# The harness is BLIND to it and cannot be otherwise: nothing inside one version
# of this add-on can see that the key shape stopped matching what an OLDER
# version reads. That is a ROLLBACK contract, and grading it needs two versions
# of the code, which is a different instrument (`tools/predeploy_check.py`'s
# territory). Stated here rather than left as an unexplained green.
_tree = mutant_tree([("app/store.py",
                      'return f"unit/{unit}/{key}"',
                      'return f"ns.{unit}.{key}"')])
_code, _out = run(Path(_tree), *CONTROLS["CONTROL-A"])
print(f"      renaming the ONE key spelling -> exit {_code}")
check("renaming `unit_key()` survives, because writer and readers move "
      "together - a fact about the design, recorded, not a gap being hidden",
      _code == RR.EXIT_OK)
check("...and the store really did change shape underneath it, so this is a "
      "survivor of a mutation that HAPPENED rather than of one that did not",
      "ns.inverter-a.sign_tripped" in _out and "unit/inverter-a/" not in _out)
shutil.rmtree(Path(_tree).parent, ignore_errors=True)

# ⚠️ THE SECOND DECLARED SURVIVOR, AND IT IS THE ONE THAT BOUNDS M5-M8. Four
# mutations of `app/mqtt.py` are now required to go red; without a mutation of
# the same file that must NOT, "the reference notices any edit to mqtt.py"
# would be indistinguishable from "the reference notices what HA keys on".
# `object_id` is the case: `app/mqtt.py` lines 36-42 and 61-76 record that HA
# 2026.8 does NOT honour it in a discovery payload - the entity_id is composed
# from device name plus entity name - so renaming one changes nothing an
# install can see, and a tool that refused it would refuse a free change.
_tree2 = mutant_tree([("app/mqtt.py",
                       '("grid", "goodwe_grid_power", "Grid power",',
                       '("grid", "goodwe_grid_pwr", "Grid power",')])
_code2, _out2 = run(Path(_tree2), *CONTROLS["CONTROL-A"])
print(f"      renaming an object_id (HA does not honour it) -> exit {_code2}")
check("renaming an `object_id` survives, because HA composes the entity_id "
      "from the device and entity NAMES and ignores it - so the four mqtt.py "
      "mutations above are red for what HA keys on, not for being edits to "
      "mqtt.py",
      _code2 == RR.EXIT_OK)
shutil.rmtree(Path(_tree2).parent, ignore_errors=True)


# --------------------------------------------------------------------------- #
print()
print("5. SAFETY-44: a pre-DEV-02 flat store meeting a DECLARED ROSTER - "
      "FIXED, with each half of the fix as its own negative control")
# --------------------------------------------------------------------------- #
# The question on the card: `migrate_unit_keys(store, UNIT_A)` took a hardcoded
# constant and had no `misdirected` guard, while `migrate_maint_keys(store,
# unit, roster)` takes the roster and does. That asymmetry was NOT safe, and
# this part used to assert the defect. It now asserts the FIX.
#
# ⚠️ THE FIX IS TWO EDITS AND THEY DO DIFFERENT JOBS, WHICH IS THE WHOLE REASON
# THIS PART IS SHAPED THE WAY IT IS. Measured, per leg, by reverting each edit
# on its own (5b and 5c below):
#
#                              SAME-STEP leg            STAGED leg
#   both edits (shipped)       clean                    latch armed, litter
#   guard reverted             latch armed              ⚠️ DISARMED
#   call site reverted         latch armed, litter      latch armed, litter
#
# So the GUARD carries the safety and the CALL SITE carries the hygiene. Saying
# that plainly matters more than it looks: a reader who assumes the roster-aware
# migration is what re-arms the latch would delete the guard as redundant during
# a tidy-up and take the STAGED column with it - and STAGED is the path
# "upgrade now, declare the roster later" produces, which is the path an
# operator actually takes.
code44, out44 = run(HERE, "state-legacy.json", "roster.yaml")
_L = legs(out44)
print(f"      flat store + roster ['garage','shed'] -> exit {code44}")
print(f"      legs in the report: {sorted(_L)}")
for _leg in ("STAGED", "SAME-STEP"):
    print(f"      LEG {_leg}: "
          f"{'DISARMED' if 'LATCH CAME UP DISARMED' in _L.get(_leg, '') else 'clean'}")
check("the report is split into legs this check can grade SEPARATELY - the "
      "tool prints a `LEG <id>` marker, because counting a finding's text over "
      "the whole report cannot tell one red leg from two",
      set(_L) == {"STAGED", "SAME-STEP"})
check("⭐ SAFETY-44 IS FIXED ON BOTH PATHS: a store that ran pre-DEV-02 flat "
      "and declares a multi-inverter roster comes up LATCHED - neither leg "
      "reports a disarmed guard",
      "LATCH CAME UP DISARMED" not in out44)
check("...and the latch is under a name THE ROSTER READS, which is the whole "
      "of the claim - `garage` is on the roster and `inverter-a` is not",
      all("after : which units came up latched  : ['garage']" in v
          for v in _L.values()))
check("⚠️ the SAME-STEP leg is now CLEAN OUTRIGHT: the migration writes the "
      "latch straight into the roster's first namespace, so there is no "
      "stranded copy left behind and nothing for an operator to read past",
      "none in LEG SAME-STEP" in _L.get("SAME-STEP", ""))
check("⚠️ ...while the STAGED leg still reports the two STRANDED keys, and "
      "that is REPORTED not hidden: its flat start stamped `schema_version` "
      "before the roster existed, so the migration takes its early return and "
      "the old copies stay in the file with nothing reading them. The latch is "
      "armed anyway - by the guard - and the litter is true",
      "STRANDED STORE KEY" in _L.get("STAGED", "")
      and "unit/inverter-a/sign_tripped" in _L.get("STAGED", ""))
check("...and the run as a whole is a FINDING rather than silence, because "
      "keys nothing reads are still keys nothing reads",
      code44 == RR.EXIT_FINDING)
check("...and no leg's text has swallowed the CLOSING SUMMARY, which repeats "
      "every finding from every leg - the failure that made the last leg read "
      "red whenever any leg was",
      not any("FINDING(S) - this transition is not clean" in v
              for v in _L.values()))
check("⚠️ and the key that MOVED is not reported as lost: the before-snapshot "
      "holds `unit/inverter-a/sign_tripped = True` and the after-snapshot "
      "holds it under `garage`, three lines apart in the same report. A "
      "VANISHED finding there is the harmless-case alarm CONTROL-C exists to "
      "prevent",
      "migrated  unit/inverter-a/sign_tripped = True -> "
      "unit/garage/sign_tripped" in out44
      and "VANISHED STORE KEY" not in out44)

# --------------------------------------------------------------------------- #
print()
print("5b. NEGATIVE CONTROL, AND THE SPLITTER'S POSITIVE ONE: revert THE "
      "GUARD, and exactly one leg goes red")
# --------------------------------------------------------------------------- #
# ⚠️ A PER-LEG CHECK THAT HAS NEVER SEEN THE LEGS DISAGREE IS A COUNT WITH
# EXTRA STEPS, and MUTATION.md's extension says one that has seen them disagree
# in a single arrangement is a total over that arrangement. This build is the
# one that makes them disagree, and it is a negative control for part 5 at the
# same time: put back the clause that counted ANY `unit/*/sign_tripped` as
# protection, including one under a name no unit answers to.
# ⚠️ WHAT THIS CONTROL PINS IS THE SHIPPED SPELLING, NOT THE RULE, and review
# is owed that distinction. `mutant_tree()` asserts its target exists - rightly,
# because a mutation that did not happen yields a fabricated result - but the
# consequence is that EDITING the guard's line makes this file red at the
# ANCHOR before any behaviour is measured. Demonstrated: rewriting the same
# semantics as `on_roster = set()` leaves the anchor intact and IS caught, by
# M4's control; rewriting the line itself is caught by the anchor, which looks
# identical from the exit code and proves much less. Read 5b and 5c as controls
# over this text, and the behavioural coverage as living in test_maintenance.py
# section 13 and test_inverter.py, which construct stores rather than strings.
_no_guard = mutant_tree([(
    "app/main.py",
    "            if unit in on_roster:",
    "            if True:  # MUTANT: any namespaced latch is protection")])
_ng_code, _ng_out = run(Path(_no_guard), "state-legacy.json", "roster.yaml")
_NG = legs(_ng_out)
for _leg in ("STAGED", "SAME-STEP"):
    print(f"      LEG {_leg}: "
          f"{'DISARMED' if 'LATCH CAME UP DISARMED' in _NG.get(_leg, '') else 'clean'}")
check("⭐ with the off-roster clause reverted the STAGED leg comes up "
      "DISARMED - so the guard's roster test is load-bearing, and it is the "
      "half of the fix that repairs an install already stamped by the shipped "
      "version",
      "LATCH CAME UP DISARMED" in _NG.get("STAGED", ""))
check("⚠️ ...while the SAME-STEP leg stays ARMED on the very same build - the "
      "legs DISAGREE, so the splitter really does read them separately and a "
      "check that only counted findings would have missed this",
      "LATCH CAME UP DISARMED" not in _NG.get("SAME-STEP", ""))
check("...and the run is a finding, so the partial build is never read as "
      "clean",
      _ng_code == RR.EXIT_FINDING)
shutil.rmtree(Path(_no_guard).parent, ignore_errors=True)

# --------------------------------------------------------------------------- #
print()
print("5c. NEGATIVE CONTROL: revert THE CALL SITE, and the litter comes back "
      "without the latch going out")
# --------------------------------------------------------------------------- #
# ⚠️ THIS IS THE CONTROL THAT KEEPS 5b HONEST ABOUT WHAT IT PROVED. 5b shows
# the guard carries the latch; on its own that is an argument for deleting the
# roster-aware migration as decoration. This build is the measurement instead:
# hand `migrate_unit_keys()` the hardcoded `UNIT_A` again, with the guard left
# fixed. The latch stays armed - the guard catches it - and the SAME-STEP leg
# gains the two stranded keys it does not have on the shipped code. The
# migration fix buys a store with nothing dead in it, and buys the
# `misdirected` refusal that stops a WRONG marker being stamped; it does not
# buy the latch. Both edits stay.
_CALL_NOW = ("        self.store_migration = migrate_unit_keys(" + chr(10) +
             "            store, _roster_names[0], _roster_names)")
_old_call = mutant_tree([(
    "app/main.py", _CALL_NOW,
    "        self.store_migration = migrate_unit_keys(store, UNIT_A)")])
_oc_code, _oc_out = run(Path(_old_call), "state-legacy.json", "roster.yaml")
_OC = legs(_oc_out)
for _leg in ("STAGED", "SAME-STEP"):
    print(f"      LEG {_leg}: "
          f"{'DISARMED' if 'LATCH CAME UP DISARMED' in _OC.get(_leg, '') else 'clean'}")
check("⚠️ with `UNIT_A` handed back to the migration the latch is STILL "
      "armed on both legs - the guard is what re-arms it, and saying so is "
      "the difference between this fix and a story about it",
      "LATCH CAME UP DISARMED" not in _oc_out)
check("⭐ ...but the SAME-STEP leg is no longer clean: the two SAFETY-02 keys "
      "are written under `inverter-a` on a roster of ['garage', 'shed'] and "
      "nothing reads them - which is the state the shipped code does not "
      "produce",
      "STRANDED STORE KEY" in _OC.get("SAME-STEP", "")
      and "none in LEG SAME-STEP" not in _OC.get("SAME-STEP", ""))
shutil.rmtree(Path(_old_call).parent, ignore_errors=True)

# --------------------------------------------------------------------------- #
print()
print("5d. THE CONTRAST: the same file onto a roster that keeps the name")
# --------------------------------------------------------------------------- #
# ⚠️ AND THE CONTRAST, WHICH IS WHAT STOPS EVERY CHECK ABOVE BEING VACUOUS. The
# SAME state file onto a roster whose first entry IS `inverter-a` comes up
# LATCHED AND CLEAN. The two runs differ in one thing - the first roster name -
# so a finding above is attributable to it and to nothing else.
code_ok, out_ok = run(HERE, "state-legacy.json", "roster-inverter-a.yaml")
check("the SAME state file onto a roster whose first entry IS `inverter-a` "
      "comes up latched and clean - so anything part 5 reports is caused by "
      "the roster NAME and not by the fixture being tripped",
      code_ok == RR.EXIT_OK and "LATCH CAME UP DISARMED" not in out_ok)


# --------------------------------------------------------------------------- #
print()
print("6. THE REVERSE DIRECTION: the retreat, and the return from it")
# --------------------------------------------------------------------------- #
code_r, out_r = run(HERE, "state-flat-070.json", "roster.yaml",
                    "--reverse", "--round-trip")
_RL = legs(out_r)
print(f"      legs in the reverse run: {sorted(_RL)}")
_orphaned = set(re.findall(r"ORPHANED ENTITY: (sensor\.\S+)",
                           _RL.get("RETREAT", "")))
print(f"      the RETREAT leg orphans: {sorted(_orphaned)}")
check("the RETREAT is rehearsed as its own leg, and THAT LEG reports the "
      "entities it orphans - the eight per-unit rows, which is the cost of "
      "backing out",
      {"RETREAT", "ROUND-TRIP"} <= set(_RL) and _orphaned == {
          f"sensor.goodwe_{u}_{k}"
          for u in ("garage", "shed")
          for k in ("setpoint", "battery_power", "battery_soc", "sign_latch")})
check("...and NOT one site entity: the eight ids the install has today survive "
      "the retreat, which is what makes the eight above attributable to the "
      "per-unit rows and not to a blanket rename",
      not any(o in _orphaned for o in
              ("sensor.goodwe_grid_power", "sensor.goodwe_setpoint",
               "sensor.goodwe_battery_power", "sensor.goodwe_p1_sample_age")))
check("...and the first pack's maintenance history SURVIVES the retreat, "
      "because `UnitHistory.mirror_flat` keeps the flat keys live - so the "
      "retreat itself does not lose state. ⚠️ Read from the RETREAT LEG "
      "alone, so the round trip's own loss cannot be mistaken for it",
      "MAINTENANCE HISTORY LOST" not in _RL["RETREAT"])
check("⚠️ ...but the ROUND TRIP does not: a maintenance cycle completed while "
      "the install is rolled back is DISCARDED when it re-advances, because "
      "`maint_schema_version` is already stamped and the migration skips",
      "RETREAT-WINDOW WORK DISCARDED" in _RL["ROUND-TRIP"])
check("...and that is a leg whose data can reach the failure it names: the "
      "retreat window actually contains a completion, written through the "
      "shipped `UnitHistory`, at a timestamp the starting fixture does not have",
      RR.RETREAT_WINDOW_COMPLETION in out_r
      and RR.RETREAT_WINDOW_COMPLETION
      not in (FIX / "state-flat-070.json").read_text())


print()
if fails:
    print(f"FAILED {len(fails)}:")
    for f in fails:
        print(f"  - {f}")
    sys.exit(1)
print("all checks passed")