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 four exit "
      "codes are distinct and only 0 is a pass",
      (RR.EXIT_OK, RR.EXIT_FINDING, RR.EXIT_COULD_NOT_RUN,
       RR.EXIT_DEGRADED) == (0, 1, 2, 3))

# ⚠️ 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")
    # The input-shape gate (SAFETY-62) and the option names it checks against.
    shutil.copyfile(HERE / "tools" / "predeploy_check.py",
                    dest / "tools" / "predeploy_check.py")
    shutil.copyfile(HERE / "config.yaml", dest / "config.yaml")
    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 - but a "
      "clean degraded run is exit 3, never 0 (SAFETY-62)",
      _deg_code == RR.EXIT_DEGRADED and "NO FINDINGS" in _deg_out)
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)

# ⭐ `M12` IS BACK, AND THE CLAIM THAT KILLED IT WAS WRONG.
#
# M12 is SAFETY-44's end-to-end control for the off-roster narrowing: a mutant
# makes one ROSTER unit's latch key vanish while a sibling holds the same
# value, and it must go red when the narrowing is removed. INSTR-01's first
# draft DELETED it, on the claim that the provenance clause made it
# unrestorable and that "NO FIXTURE CAN RESTORE IT, measured both ways".
#
# ⚠️ TWO MEASUREMENTS ARE NOT AN IMPOSSIBILITY PROOF, AND THIS ONE WAS FALSE.
# Both arrangements I measured were SAME-STEP shaped, and the structural
# argument - `run()` builds both snapshots from one state file, so "already
# latched" and "made by this transition" are incompatible - is TRUE OF
# SAME-STEP AND FALSE OF STAGED. `snapshot_over()` starts the after-leg over
# the file the BEFORE-leg wrote, so on the staged leg a transition-made
# destination is available and the two questions stop being incompatible.
# Review found the third arrangement I had generalised past.
#
# HOW IT WORKS. The fixture is `state-legacy.json` plus a second roster unit
# already latched. The mutant makes `shed` POP its own latch key and come up
# untripped - a truer SAFETY-02 loss than the original, which left `tripped`
# True in memory. On the staged leg, losing shed's latch makes
# `_grade_store_migration()` fall through `on_roster_latched` and RE-ARM
# `unit/garage/sign_tripped`, which is absent from `before.data` - so the
# destination is transition-made and C7 lets it through. The off-roster
# `inverter-a` litter is present in BOTH snapshots on this leg, so it never
# vanishes and never claims the destination, and C6 stays out of the way.
#
# ⚠️ AND WHAT ACTUALLY BROKE WAS THE PRECONDITION, NOT THE CONTROL. M12's
# `exit == EXIT_OK` baseline does not survive a legacy-shaped fixture. That
# precondition is the one SAFETY-44's own commit called "belt-and-braces,
# load-bearing only if the exit-0 requirement is ever relaxed", saying that
# what closed the first draft's gap is `_m12_new` BEING A SET DIFFERENCE. A
# stable dirty baseline is fine for a set difference. So the control was
# deleted on the strength of a precondition the same commit had already
# characterised as not load-bearing - which is a worse mistake than the
# original inert check, because it removed coverage rather than failing to add
# it.
_M12_OLD = 'tripped=bool(data.get(self.key("sign_tripped"), False)),'
# ⚠️ THE MUTANT POPS **AND COMES UP UNTRIPPED**, which is the whole point and
# which the first draft got wrong: `data.pop(k, False)` RETURNS the value, so
# shed stayed latched in memory, `_grade_store_migration()` saw
# `on_roster_latched` and never re-armed garage - no transition-made
# destination, and the finding was reported whether or not C5 existed. The
# control looked restored and discriminated nothing. `and False` is the
# difference between "the key is gone" and "the LATCH is gone", and only the
# second reaches the re-arm this arrangement needs.
_M12_NEW = ('tripped=bool(data.pop(self.key("sign_tripped"), False) and False'
            + chr(10) + '                     if self.name == "shed" else'
            + chr(10) + '                     data.get(self.key("sign_tripped"), False)),')


def m12_findings(tree):
    """The finding SET of a staged-leg run over the legacy+shed fixture."""
    code, out = run(Path(tree), "state-legacy-plus-shed.json", "roster.yaml",
                    "--paths", "staged")
    return code, set(re.findall(r"⚠️ ([A-Z ]+): (\S+)", out)), out


_m12 = mutant_tree([("app/inverter.py", _M12_OLD, _M12_NEW)])
_m12_clean_code, _m12_clean, _ = m12_findings(HERE)
_m12_code, _m12_mut, _ = m12_findings(Path(_m12))
_m12_new = _m12_mut - _m12_clean
print(f"      staged leg, legacy+shed: shipped exit {_m12_clean_code}, "
      f"one roster unit's latch popped -> exit {_m12_code}")
# ⚠️ THE PRINTED SET IS LOSSIER THAN IT LOOKS: the regex captures the first
# TOKEN of each finding's text, so two findings of one kind collapse into one
# tuple. Harmless here - the difference is one-directional and the pair M12
# asserts on is exact - but do not read this line as an inventory.
print(f"      findings the mutation ADDS (first token only): {sorted(_m12_new)}")
check("⚠️ PRECONDITION: the baseline is STABLE and does not already carry the "
      "finding this mutation exists to produce. It is NOT clean - a legacy "
      "fixture never is - and the SET DIFFERENCE is what makes that fine, "
      "which is the sentence the deletion should have read before acting on it",
      not any(k == "unit/shed/sign_tripped" for _, k in _m12_clean))
# ⚠️ AND IT MATCHES THE **KIND** AS WELL AS THE KEY, which SAFETY-44's original
# did not. Review measured it: renaming the finding kind in `diff()` from
# "VANISHED STORE KEY" to "GONE STORE KEY" left M12 GREEN. A vanished latch
# quietly downgraded to a softer heading is exactly the shape part 3 requires
# every mutation to be graded against - "the SPECIFIC finding it should produce"
# - and this file was not holding itself to its own rule.
check("⭐ M12 RESTORED: a ROSTER unit's latch key that vanishes is a "
      "VANISHED STORE KEY finding even though a sibling holds True - and this "
      "is the STAGED leg, where `snapshot_over()` supplies a transition-made "
      "destination, so the provenance clause does not foreclose the narrowing",
      ("VANISHED STORE KEY", "unit/shed/sign_tripped") in _m12_new)
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()))
# ⚠️ THIS BLOCK USED TO SAY THOSE HOLES WERE KNOWN AND NOT FIXED, AND IT IS
# NOW STALE IN THE OTHER DIRECTION - INSTR-01 fixed them, and left the
# paragraph saying "BOTH ARE REVERTED OUT OF THIS CARD, because THIS SUITE
# CANNOT MEASURE EITHER ONE" describing a tree that no longer exists. Review
# found it. A comment that outlives its subject is the same defect as a
# citation that outlives its check.
#
# WHERE EACH ONE NOW LIVES, and every clause of `moved_to` has a row in 5e:
#
#   INJECTIVITY  (C6)  R5, and removing the clause flips exactly R5
#   PROVENANCE   (C7)  R3, likewise
#   SAME VALUE   (C3)  R4
#   SAME BARE KEY(C2)  R7 - which had NO row until review deleted the clause
#                      and the whole 28-file suite stayed green
#   MACHINE-READ (C1)  R6
#   OFF-ROSTER   (C5)  R2, and M12 end-to-end on the staged leg
#   PREFIX GATE  (C4)  NO 5e ROW, deliberately - `vanished_unit is not None`
#                      is reached end-to-end by part 3's M9, and a synthetic
#                      row would duplicate that rather than add separability.
#                      Listed because C5 lives in the same `if` and omitting
#                      one guard while naming the other reads as an oversight.
#
# ⚠️ ONE HOLE IS STILL OPEN AND IS **UNMEASURED**, stated so nobody reads
# `sorted()` as guarded: ATTRIBUTION. Which vanished key gets a destination is
# decided by `sorted()` order, so on a build renaming `inverter-b` -> `garage`
# while DELETING `inverter-a`, the deleted key sorts first and takes it - the
# report names the DELETED key as migrated and the one that really moved as
# VANISHED. Review measured that replacing `sorted(` with `list(` leaves the
# suite green: every 5e case has at most one candidate destination, so no row
# can see ordering. The run is still exit 1, so this is a WRONG SENTENCE rather
# than a hidden loss - an operator chasing the named key looks at the wrong
# inverter. Fixing it needs the destination's provenance rather than
# first-come, which is more than a rename diff currently knows.
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("5e. INSTR-01: `diff()`'s suppression rule, ONE FIXTURE PER CLAUSE")
# --------------------------------------------------------------------------- #
# ⚠️ THIS SECTION EXISTS BECAUSE SAFETY-44 PRODUCED THREE INERT CONTROLS FOR
# THIS ONE RULE, EACH INERT DIFFERENTLY:
#
#   1. a check over part 5's real destinations passed with the whole mechanism
#      deleted - its two destinations are `sign_reason` and `sign_tripped`,
#      DIFFERENT bare keys, which can never contend;
#   2. a dedicated end-to-end fixture discriminated nothing - both snapshots
#      read ONE state file, so nothing vanishes and there are no migrated lines
#      with the clause or without it;
#   3. rebuilding M12's fixture to reach a new clause DISARMED M12 for an older
#      one, and all three clause-removals went green at once.
#
# ⭐ (3) IS THE STRUCTURAL FINDING AND IT IS WHY THIS SECTION IS SHAPED THIS WAY.
# `moved_to` is ONE comprehension carrying FIVE conditions with five different
# reasons to exist, and an END-TO-END fixture cannot separate them: both
# snapshots come from one state file, so the arrangement that reaches clause N
# forecloses clause M. Rebuilding the fixture to reach one moves the others.
#
# So the rule is graded at the level it is written at - `diff()` on constructed
# Snapshots - with ONE CASE PER CLAUSE. Nothing here is a fiction: the inputs
# are the same three fields `diff()` reads (`data`, `machine_read`, `units`),
# and the end-to-end controls above - M12 on the STAGED leg, and part 5 -
# still grade the same rule through a real start. This is the separability
# those cannot have, not a replacement for them.
CLAUSE_CASES = {
    # name: (units, machine_read, before, after)
    "R1-rename-legit": (
        ["garage"], ["unit/garage/sign_tripped"],
        {"unit/old/sign_tripped": True}, {"unit/garage/sign_tripped": True}),
    "R2-narrowing": (
        ["garage", "shed"], ["unit/garage/sign_tripped", "unit/shed/sign_tripped"],
        {"unit/shed/sign_tripped": True}, {"unit/garage/sign_tripped": True}),
    "R3-provenance": (
        ["garage"], ["unit/garage/sign_tripped"],
        {"unit/old/sign_tripped": True, "unit/garage/sign_tripped": True},
        {"unit/garage/sign_tripped": True}),
    "R4-value": (
        ["garage"], ["unit/garage/sign_reason"],
        {"unit/old/sign_reason": "X-unique"},
        {"unit/garage/sign_reason": "Y-different"}),
    "R5-injectivity": (
        ["garage"], ["unit/garage/sign_tripped"],
        {"unit/old-a/sign_tripped": True, "unit/old-b/sign_tripped": True},
        {"unit/garage/sign_tripped": True}),
    "R6-machine-read": (
        ["garage"], [],
        {"unit/old/sign_tripped": True}, {"unit/garage/sign_tripped": True}),
    # ⚠️ R7 EXISTS BECAUSE THE BARE-KEY MATCH HAD NO ROW AT ALL, and review
    # deleted it and left the whole 28-file suite GREEN. The rule's own comment
    # says "THE TEST IS NOT 'the name appears somewhere'. It is: the SAME bare
    # key" - and nothing measured that sentence. It is load-bearing: a vanished
    # SAFETY-02 `sign_tripped` was suppressed by a SAFETY-42
    # `delivery_tripped` on a roster unit, i.e. by a DIFFERENT FAILSAFE's key.
    # `unit/*/delivery_tripped` is machine-read, so this is reachable on any
    # real two-latch store rather than a contrivance.
    "R7-bare-key": (
        ["garage"], ["unit/garage/delivery_tripped"],
        {"unit/old/sign_tripped": True},
        {"unit/garage/delivery_tripped": True}),
}
# What the SHIPPED rule must answer. `migrated` is the suppression firing.
CLAUSE_EXPECT = {
    "R1-rename-legit": ([], ["unit/old/sign_tripped"]),
    "R2-narrowing":    (["unit/shed/sign_tripped"], []),
    "R3-provenance":   (["unit/old/sign_tripped"], []),
    "R4-value":        (["unit/old/sign_reason"], []),
    "R5-injectivity":  (["unit/old-b/sign_tripped"], ["unit/old-a/sign_tripped"]),
    "R6-machine-read": (["unit/old/sign_tripped"], []),
    "R7-bare-key":     (["unit/old/sign_tripped"], []),
}
# ⚠️ THE DRIVER **IMPORTS** THE TREE UNDER TEST AND THE TABLE STAYS HERE, so
# expectation and implementation sit on opposite sides of the subprocess and
# a mutation to the tool cannot also move the answer it is graded against.
# (It is WRITTEN to a tempdir - see `clause_table()`. The first draft of this
# sentence said "written into the tree", which is the read-only-CI bug the
# same commit says it fixed, stated as if it were the design.)
CLAUSE_DRIVER = """
import json, sys
import rehearse_roster as RR
spec = json.load(sys.stdin)
out = {}
for name, case in spec.items():
    units, mread, before, after = case
    b = object.__new__(RR.Snapshot); a = object.__new__(RR.Snapshot)
    for snap, data in ((b, before), (a, after)):
        snap.label = name
        snap.data = dict(data)
        snap.units = list(units)
        snap.machine_read = set(mread)
        snap.entities = {}
        snap.latched = {u: False for u in units}
        snap.latch_reason = {u: "" for u in units}
        snap.flat_tripped = False
        snap.history = {}
        snap.read = set()
        snap.stranded_at_default = []
        snap.vanished_at_default = []
        snap.vanished_but_migrated = []
    f = RR.diff(b, a)
    out[name] = [
        sorted(x.text.split(" = ")[0] for x in f
               if x.kind == "VANISHED STORE KEY"),
        sorted(t.split(" = ")[0] for t in a.vanished_but_migrated),
    ]
print(json.dumps(out))
"""


def clause_table(tree) -> dict:
    """Run CLAUSE_CASES through `diff()` in `tree`, as the tool would see it."""
    # ⚠️ THE DRIVER GOES IN A TEMPDIR, NOT IN THE TREE. `checks / unit` mounts
    # the checkout READ-ONLY, so writing beside the tool works on a dev box and
    # is `OSError: Read-only file system` in CI - a difference that would have
    # turned this section red on every PR and green for whoever wrote it.
    tools = Path(tree) / "tools"
    tmp = tempfile.mkdtemp(prefix="clause-drv-")
    drv = Path(tmp) / "_clause_driver.py"
    drv.write_text(CLAUSE_DRIVER, encoding="utf-8")
    proc = subprocess.run(
        [sys.executable, str(drv)],
        input=json.dumps(CLAUSE_CASES), capture_output=True, text=True,
        cwd=tmp,
        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1",
             "PYTHONPATH": f"{tools}{os.pathsep}{Path(tree)}"})
    shutil.rmtree(tmp, ignore_errors=True)
    assert proc.returncode == 0, (
        "the clause driver did not run, so every row below would be "
        f"fabricated:\n{proc.stdout}\n{proc.stderr}")
    return {k: (v[0], v[1]) for k, v in json.loads(proc.stdout).items()}


_ct = clause_table(HERE)
for _name in sorted(CLAUSE_CASES):
    _got, _want = _ct[_name], CLAUSE_EXPECT[_name]
    print(f"      {_name:<18} vanished={_got[0]} migrated={_got[1]}")
    check(f"{_name}: the shipped rule answers as specified", list(_got) == [list(_want[0]), list(_want[1])])
check("⚠️ ...and the LEGITIMATE case is still suppressed - a rule that reports "
      "every rename is a rule an operator turns off, which is the failure "
      "CONTROL-C exists to prevent",
      _ct["R1-rename-legit"][1] == ["unit/old/sign_tripped"])

# ⭐ AND EACH CLAUSE REMOVAL MUST FLIP **EXACTLY ITS OWN ROW**. That second half
# is the acceptance criterion this card was filed with: a removal that flips
# nothing is an unmeasured clause, and a removal that flips someone else's row
# means the cases are not separable after all - which is (3) above, caught.
CLAUSE_MUTATIONS = [
    ("C2 same bare key", "R7-bare-key",
     '            if k.rsplit("/", 1)[-1] == bare and after.data.get(k) == value',
     "            if True and after.data.get(k) == value"),
    ("C1 machine_read", "R6-machine-read",
     "            k for k in after.machine_read",
     "            k for k in after.data"),
    ("C3 same value", "R4-value",
     " and after.data.get(k) == value\n", "\n"),
    ("C5 off-roster", "R2-narrowing",
     "        if vanished_unit is not None and vanished_unit not in after.units \\\n"
     "                and moved_to:",
     "        if vanished_unit is not None and moved_to:"),
    ("C6 injectivity", "R5-injectivity",
     " and k not in claimed)", ")"),
    ("C7 provenance", "R3-provenance",
     "            and before.data.get(k) != value", "            and True"),
]
for _label, _owns, _old, _new in CLAUSE_MUTATIONS:
    _t = mutant_tree([("tools/rehearse_roster.py", _old, _new)])
    _mt = clause_table(_t)
    _flipped = sorted(n for n in CLAUSE_CASES if _mt[n] != _ct[n])
    print(f"      {_label:<16} removed -> rows that changed: {_flipped}")
    check(f"⭐ removing {_label} flips {_owns} - the clause is load-bearing and "
          f"this row is what measures it",
          _flipped == [_owns])
    shutil.rmtree(Path(_t).parent, ignore_errors=True)


# --------------------------------------------------------------------------- #
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")