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 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, which "
      "is the path nobody had established either way")
# --------------------------------------------------------------------------- #
# The question on the card: `migrate_unit_keys(store, UNIT_A)` takes a hardcoded
# constant and has no `misdirected` guard, while `migrate_maint_keys(store,
# unit, roster)` takes the roster and does. Is that asymmetry safe?
#
# ⚠️ MEASURED, NOT ARGUED. The two checks below are the answer, and they are
# written as REQUIREMENTS so that a fix to `migrate_unit_keys` turns them red
# and this file has to be updated deliberately rather than drifting.
code44, out44 = run(HERE, "state-legacy.json", "roster.yaml")
print(f"      flat store + roster ['garage','shed'] -> exit {code44}")
check("⚠️ SAFETY-44 IS A DEFECT AGAIN: a store that ran pre-DEV-02 flat and "
      "declares a multi-inverter roster comes up with its SAFETY-02 latch "
      "DISARMED - the file says `sign_tripped: true` and not one inverter is "
      "latched",
      code44 == RR.EXIT_FINDING and "LATCH CAME UP DISARMED" in out44)
check("...and the harness names the two latch keys SAFETY-44 asks about, by "
      "the namespace they were stranded in",
      "unit/inverter-a/sign_tripped" in out44
      and "unit/inverter-a/sign_reason" in out44)
check("...and the mechanism is visible in the report: the keys are stranded "
      "under `inverter-a`, which is not on the roster, so no machine reads them",
      "STRANDED STORE KEY" in out44 and "['garage', 'shed']" in out44)
# ⚠️ PER LEG, NOT A COUNT OVER THE WHOLE REPORT. See `legs()` for the three
# builds this distinction was measured on and for what the count missed.
_L = legs(out44)
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("⚠️ SAME-STEP is disarmed: the roster is declared in the SAME step as "
      "the upgrade, the migration runs and misdirects the latch to "
      "`inverter-a`",
      "LATCH CAME UP DISARMED" in _L.get("SAME-STEP", ""))
check("⚠️ AND IT IS NOT THE SAME-STEP PATH ALONE - the STAGED leg reaches the "
      "same disarmed latch by a DIFFERENT route: the flat start stamps "
      "`schema_version`, so the roster start's migration takes its early "
      "return and the latch never moves at all",
      "LATCH CAME UP DISARMED" in _L.get("STAGED", ""))

# ⚠️ AND THE CONTRAST, WHICH IS WHAT STOPS THE CHECK ABOVE BEING VACUOUS. The
# SAME state file onto a roster whose first entry IS `inverter-a` comes up
# LATCHED. The two runs differ in one thing - the first roster name - so the
# finding is attributable to it and to nothing else.
code_ok, out_ok = run(HERE, "state-legacy.json", "roster-inverter-a.yaml")
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()))

# --------------------------------------------------------------------------- #
print()
print("5b. THE SPLITTER'S POSITIVE CONTROL: a build in which exactly ONE leg "
      "is clean")
# --------------------------------------------------------------------------- #
# ⚠️ A PER-LEG CHECK THAT HAS NEVER SEEN THE LEGS DISAGREE IS A COUNT WITH
# EXTRA STEPS. Every fixture above has both legs red, so nothing so far
# separates "reads each leg" from "reads the whole report twice". This part
# builds the one case that does: `migrate_unit_keys()` handed the roster's own
# first name instead of the hardcoded `UNIT_A`.
#
# ⚠️ AND WHAT IT MEASURES IS ALSO THE ANSWER TO "IS THAT THE FIX". It is NOT.
# It repairs the SAME-STEP path only. The STAGED path is untouched, because
# there the flat start stamps `schema_version = 1` and the roster start's
# migration takes its `from_version >= SCHEMA_VERSION` early return before the
# name it would be handed matters at all. A fix that stops at this line would
# close the half an operator is least likely to take and leave the half that
# "upgrade first, declare the roster later" produces.
_one_leg = mutant_tree([(
    "app/main.py",
    "        self.store_migration = migrate_unit_keys(store, UNIT_A)",
    "        self.store_migration = migrate_unit_keys(store, "
    "roster_names(opts)[0])")])
_ol_code, _ol_out = run(Path(_one_leg), "state-legacy.json", "roster.yaml")
_OL = legs(_ol_out)
for _leg in ("STAGED", "SAME-STEP"):
    print(f"      LEG {_leg}: "
          f"{'DISARMED' if 'LATCH CAME UP DISARMED' in _OL.get(_leg, '') else 'clean'}")
check("⭐ with `migrate_unit_keys()` handed the roster's own first name, the "
      "SAME-STEP leg comes up CLEAN - so the splitter really does read the "
      "legs separately, and a check that only counted would have missed this",
      "LATCH CAME UP DISARMED" not in _OL["SAME-STEP"])
check("⚠️ ...while the STAGED leg is STILL DISARMED on the same build: that "
      "line is not the whole fix, because the staged path never reaches it - "
      "the first start stamps `schema_version` and the second start's "
      "migration returns early",
      "LATCH CAME UP DISARMED" in _OL["STAGED"])
check("...and the run as a whole is still a finding, so nobody reads this "
      "partial fix as done",
      _ol_code == RR.EXIT_FINDING)
shutil.rmtree(Path(_one_leg).parent, ignore_errors=True)

print()
print("5c. back to the shipped code")
check("the SAME state file onto a roster whose first entry IS `inverter-a` "
      "comes up latched and clean - so the finding above 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")