goodwe-addon/goodwe_controller/test_mqtt.py

"""Runnable check for TEL-07 - the SECOND MQTT device, and A's history.
`python3 test_mqtt.py`

WHAT THIS FILE IS FOR, in one sentence: to prove by DIFFING THE ANNOUNCED
PAYLOADS - not by reading the code - that giving inverter B its own MQTT device
changes nothing about inverter A's.

⚠️ WHY THAT IS THE WHOLE CARD. `app/mqtt.py`'s own note records the finding:
Home Assistant composes an entity_id from the DEVICE name plus the ENTITY name,
and `object_id` in the discovery payload did NOT override it (tested on HA
2026.8). So re-scoping the device renames every entity - and because HA keeps
the id it composed at FIRST discovery, what a rename actually produces is a NEW
entity beside a DEAD one. The recorded history stays behind on the old id, and
every dashboard, template and automation that names it - including the three
live watchdog automations `goodwe_master_watchdog_held_*` and
`goodwe_master_parked_at_0_w_*` - goes on pointing at an entity that has quietly
stopped updating. Nothing errors. That is the failure this file exists to make
impossible to ship.

⚠️ AND THE TEMPTING DESIGN IS THE DESTRUCTIVE ONE. "One device per inverter" -
A becomes `inverter-a`, B becomes `inverter-b` - is symmetrical and tidy and is
exactly the change described above. A KEEPS THE DEVICE IT ALREADY HAS; only B
gets a new one. Part 8 below writes that tidy version and shows it going red.

THE SHAPE OF THE EVIDENCE. Part 1 records what `release/1.0` actually publishes -
generated by running THAT commit's module, not transcribed - and every later
part is a diff against it, reported as `N topics, M missing, C changed, A added`.

  Part 1  The recorded baseline, and the diff at a roster of one and of none.
  Part 2  `_announce()` publishes EXACTLY `discovery_payloads()` and nothing
          else, so parts 1 and 3-8 are diffing what goes on the wire.
  Part 3  A roster of two: A's nine messages byte-identical, eight added.
  Part 4  Zero overlap between A's entities and B's - topics, unique_ids,
          object_ids, state topics, and devices.
  Part 5  Site-scoped sensors stay site-scoped, at every roster length 1-6.
  Part 6  The composed entity_ids, modelled INDEPENDENTLY of `app/mqtt.py`, and
          anchored on the one composed id that was verified against the live
          system (`sensor.goodwe_p1_sample_age`, mqtt.py's own note).
  Part 7  ⚠️ THE ANNOUNCEMENT AND THE PUBLICATION AGREE, on a real `Controller`,
          with the two units carrying DIFFERENT numbers - so a swap is visible.
  Part 8  ⚠️ VERIFIED BY BREAKING IT. Twelve mutations of the shipped module,
          re-executed from mutated source, each one named with the property it
          breaks. A property nothing breaks is a property nothing is watching.
  Part 9  What these checks CANNOT see, enumerated and sized.

⚠️ THE FIXTURE GIVES A AND B DIFFERENT NAMES, DIFFERENT RATINGS, DIFFERENT SoC,
DIFFERENT BATTERY POWER AND DIFFERENT LATCH STATES, ON PURPOSE. A fixture whose
two units are equal cannot observe a bug that swaps them - MUTATION.md habit 4,
and the defect this repository has now shipped six cards in a row.

⚠️ INVERTER B DOES NOT PHYSICALLY EXIST (~2026-09-12). Every number attributed
to it below is SIMULATED, from the datasheet and serial recorded in NOTES.md
§1.1b (GW5048D-ES, serial 55048ESU22AW0992, max_w 4600 W). Nothing here is
measured, and nothing here should ever be quoted as if it were.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ. Read MUTATION.md.
Part 8 sidesteps the whole problem by exec'ing mutated SOURCE into a fresh
namespace, so no `.pyc` is involved at any point.
"""

import ast
import asyncio
import hashlib
import json
import re
import sys
from pathlib import Path

from app import mqtt
from app.inverter import build_roster, roster_names
from app.main import Controller

HERE = Path(__file__).resolve().parent
MQTT_SRC = (HERE / "app" / "mqtt.py").read_text(encoding="utf-8")

fails = []


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


# ---------------------------------------------------------------------------
# The recorded baseline. ⚠️ GENERATED, NOT TRANSCRIBED. Reproduce it with:
#
#   git show release/1.0:goodwe_controller/app/mqtt.py > /tmp/mqtt_base.py
#   python3 - <<'EOF'
#   import sys; sys.path.insert(0, "/tmp"); import mqtt_base as M
#   class F:
#       def __init__(self): self.m = []
#       def publish(self, t, p=None, retain=False): self.m.append((t, p, retain))
#   p = M.MqttPublisher.__new__(M.MqttPublisher)
#   p.omit, p.client, p.enabled = set(), F(), True
#   p._announce()
#   for row in p.client.m: print(row)
#   EOF
#
# A transcription would be a check that grades this branch against MY MEMORY of
# release/1.0 rather than against release/1.0.
# ---------------------------------------------------------------------------
_DEV = ('{"identifiers": ["goodwe_rs485_controller"], "name": "GoodWe", '
        '"manufacturer": "GoodWe (via RS485 meter emulation)", '
        '"model": "ES/BP series"}')

BASELINE = [
    ('homeassistant/sensor/goodwe_ctl_setpoint/config',
     '{"name": "Setpoint", "object_id": "goodwe_setpoint", "unique_id": "goodwe_ctl_setpoint", '
     '"state_topic": "goodwe_ctl/setpoint", "availability_topic": "goodwe_ctl/availability", '
     '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", '
     '"state_class": "measurement"}', True),
    ('homeassistant/sensor/goodwe_ctl_grid/config',
     '{"name": "Grid power", "object_id": "goodwe_grid_power", "unique_id": "goodwe_ctl_grid", '
     '"state_topic": "goodwe_ctl/grid", "availability_topic": "goodwe_ctl/availability", '
     '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", '
     '"state_class": "measurement"}', True),
    ('homeassistant/sensor/goodwe_ctl_battery/config',
     '{"name": "Battery power", "object_id": "goodwe_battery_power", "unique_id": "goodwe_ctl_battery", '
     '"state_topic": "goodwe_ctl/battery", "availability_topic": "goodwe_ctl/availability", '
     '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", '
     '"state_class": "measurement"}', True),
    ('homeassistant/sensor/goodwe_ctl_soc/config',
     '{"name": "Battery SoC", "object_id": "goodwe_battery_soc", "unique_id": "goodwe_ctl_soc", '
     '"state_topic": "goodwe_ctl/soc", "availability_topic": "goodwe_ctl/availability", '
     '"device": ' + _DEV + ', "unit_of_measurement": "%", "device_class": "battery", '
     '"state_class": "measurement"}', True),
    ('homeassistant/sensor/goodwe_ctl_circulation/config',
     '{"name": "Circulating power", "object_id": "goodwe_circulating_power", '
     '"unique_id": "goodwe_ctl_circulation", "state_topic": "goodwe_ctl/circulation", '
     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
     '"unit_of_measurement": "W", "device_class": "power", "state_class": "measurement", '
     '"icon": "mdi:sync-alert"}', True),
    ('homeassistant/sensor/goodwe_ctl_phase/config',
     '{"name": "Maintenance phase", "object_id": "goodwe_maintenance_phase", '
     '"unique_id": "goodwe_ctl_phase", "state_topic": "goodwe_ctl/phase", '
     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
     '"icon": "mdi:battery-sync"}', True),
    ('homeassistant/sensor/goodwe_ctl_status/config',
     '{"name": "Controller status", "object_id": "goodwe_controller_status", '
     '"unique_id": "goodwe_ctl_status", "state_topic": "goodwe_ctl/status", '
     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
     '"icon": "mdi:heart-pulse"}', True),
    ('homeassistant/sensor/goodwe_ctl_p1_age/config',
     '{"name": "P1 sample age", "object_id": "p1_sample_age_s", "unique_id": "goodwe_ctl_p1_age", '
     '"state_topic": "goodwe_ctl/p1_age", "availability_topic": "goodwe_ctl/availability", '
     '"device": ' + _DEV + ', "unit_of_measurement": "s", "device_class": "duration", '
     '"state_class": "measurement"}', True),
    ('goodwe_ctl/availability', 'online', True),
]

# --- TEL-10's three, and they are DELIBERATELY NOT IN `BASELINE` -------------
#
# ⚠️ `BASELINE` IS WHAT release/1.0 PUBLISHES AND NOTHING MAY BE ADDED TO IT.
# Every strong claim in this file rests on that list being a record of another
# tree; append a row this branch invented and "byte-identical to release/1.0"
# quietly becomes "identical to itself", which is the defect MUTATION.md's
# habit 4 is about - a check whose data cannot reach the failure it names.
#
# So TEL-10's arrival is TWO claims and gets two checks: the nine release/1.0
# messages did not move, and EXACTLY these three arrived after them.
#
# ⚠️ AND THIS LIST IS A REGRESSION PIN, NOT EVIDENCE OF CORRECTNESS - the same
# status as part 3's sha256 over the per-unit payloads, for the same reason:
# there is no released version of these bytes to diff against, so all it can
# say is "these three payloads have not changed since the PR was reviewed".
_TEL10_NUM = ('"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV
              + ', "unit_of_measurement": "W", "device_class": "power", '
                '"state_class": "measurement"}')
TEL10 = [
    ('homeassistant/sensor/goodwe_ctl_loop_i/config',
     '{"name": "Loop integrator", "object_id": "goodwe_loop_integrator", '
     '"unique_id": "goodwe_ctl_loop_i", "state_topic": "goodwe_ctl/loop_i", '
     + _TEL10_NUM, True),
    ('homeassistant/sensor/goodwe_ctl_loop_lead/config',
     '{"name": "Loop integrator lead", "object_id": "goodwe_loop_integrator_lead", '
     '"unique_id": "goodwe_ctl_loop_lead", "state_topic": "goodwe_ctl/loop_lead", '
     + _TEL10_NUM, True),
    ('homeassistant/sensor/goodwe_ctl_loop_reason/config',
     '{"name": "Loop reason", "object_id": "goodwe_loop_reason", '
     '"unique_id": "goodwe_ctl_loop_reason", "state_topic": "goodwe_ctl/loop_reason", '
     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
     '"icon": "mdi:sigma"}', True),
]

# What THIS branch announces at a roster of one: release/1.0's eight sensor
# configs, then TEL-10's three, then the availability message - which stays
# LAST, because `_announce()` publishes it after every config and a broker
# replaying the retained set must see availability arrive after the entities it
# is the availability OF.
SITE_NOW = BASELINE[:-1] + TEL10 + BASELINE[-1:]


class FakeClient:
    """Captures what `_announce()` actually hands paho."""

    def __init__(self):
        self.msgs = []

    def publish(self, topic, payload=None, retain=False):
        self.msgs.append((topic, payload, retain))


def announce(units=(), omit=(), module=mqtt, include=()):
    """The messages `_announce()` publishes, through the SHIPPED method.

    ⚠️ Not `discovery_payloads()` directly. Part 2 proves the two agree; every
    other part goes through this one, so what is diffed is what paho is told.

    `include` is TEL-11's optional-row set and DEFAULTS TO EMPTY here for the
    same reason it does in the shipped constructor: every check in this file
    written before that key existed must go on grading the announcement an
    install that never set `pv_entity` receives, which is all of them.
    """
    pub = module.MqttPublisher.__new__(module.MqttPublisher)
    pub.omit = set(omit)
    pub.units = [str(u) for u in units]
    pub.include = [str(k) for k in include]
    pub.enabled = True
    pub.client = FakeClient()
    pub._announce()
    return pub.client.msgs


def diff(got, want=SITE_NOW):
    """`(missing, changed, added)` between two announcement lists, by topic."""
    g, w = dict((t, (p, r)) for t, p, r in got), dict((t, (p, r)) for t, p, r in want)
    missing = [t for t in w if t not in g]
    changed = [t for t in w if t in g and g[t] != w[t]]
    added = [t for t in g if t not in w]
    return sorted(missing), sorted(changed), sorted(added)


def report(label, got, want=SITE_NOW):
    m, c, a = diff(got, want)
    print(f"    {label}: {len(want)} topics, {len(m)} missing, {len(c)} changed, "
          f"{len(a)} added")
    for t in m:
        print(f"      MISSING {t}")
    for t in c:
        print(f"      CHANGED {t}")
    for t in a:
        print(f"      ADDED   {t}")
    return m, c, a


def cfgs(msgs):
    """The discovery payloads, parsed, without the availability message."""
    return [json.loads(p) for t, p, _ in msgs if t.startswith("homeassistant/")]


# The roster this card exists for. ⚠️ SIMULATED: inverter B does not physically
# exist. Its rating is the datasheet's (NOTES.md §1.1b, GW5048D-ES, serial
# 55048ESU22AW0992) and has never been measured.
OPTS2 = {
    "meter_source": "off",
    "inverters": [
        {"name": "inverter-a", "setpoint_entity": "number.sp_a",
         "soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
         "max_w": 2000, "bms_soc_floor": 10},
        {"name": "inverter-b", "setpoint_entity": "number.sp_b",
         "soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b",
         "max_w": 4600, "bms_soc_floor": 10},
    ],
}
NAMES2 = roster_names(OPTS2)


# ---------------------------------------------------------------------------
print("\n1. the recorded baseline, and a roster of one")
# ---------------------------------------------------------------------------
# AC 1 and AC 4. ⚠️ THE COMPARISON IS BYTE-FOR-BYTE ON THE PAYLOAD STRING, not
# on a parsed dict. `json.dumps` writes keys in insertion order, so a reordered
# payload is a different retained message on the broker even though the parsed
# dicts compare equal - and a reordered payload is a re-announcement of every
# entity on a live install.
m, c, a = report("roster of one (`inverters:` empty, i.e. every install today)",
                 announce(units=roster_names({})))
# ⚠️ THE HEADLINE SAYS EIGHT, NOT NINE, AND THE SLICE UNDER IT IS WHY.
# release/1.0 announces NINE messages; the ninth is `goodwe_ctl/availability`,
# which `_announce()` publishes LAST and which TEL-10 therefore MOVES - it is
# byte-identical but no longer at index 8. So what is asserted here is the eight
# entity CONFIGS, `[:len(BASELINE) - 1] == BASELINE[:-1]`, and the word "nine"
# above it named an assertion this file does not make. The ninth is not
# unguarded: the `== SITE_NOW` on the next check pins it byte-for-byte in its
# new last position. Two sentences, two checks.
check("⭐ AT THE SHIPPED ROSTER OF ONE EVERY ONE OF release/1.0's EIGHT ENTITY "
      "CONFIGS IS STILL BYTE-IDENTICAL AND STILL FIRST: same topics, same "
      "unique_ids, same payload bytes, same retain flags, in the same order - "
      "no second device, no renamed entity, nothing re-ordered. TEL-10 adds "
      "rows AFTER them; it may not move one of them. (release/1.0's NINTH "
      "message, `availability`, is byte-identical too and deliberately moves "
      "to LAST; the `== SITE_NOW` below is what pins it)",
      announce(units=roster_names({}))[:len(BASELINE) - 1] == BASELINE[:-1])

# ⚠️ AND THE ADDITION IS ITS OWN CLAIM, WITH A COUNT. "Nothing moved" and
# "exactly three arrived" are different sentences and a single `== SITE_NOW`
# would let a fourth row hide inside whichever of the two a reader had in mind.
_m1, _c1, _a1 = diff(announce(units=roster_names({})), BASELINE)
check("⭐ TEL-10 ADDS EXACTLY THREE TOPICS AND CHANGES NONE: the integrator, "
      "the lead and the loop's own reason string, all NEW, all on the SITE "
      "device - nothing of release/1.0's is missing and nothing of it changed",
      (_m1, _c1) == ([], [])
      and sorted(_a1) == sorted(t for t, _p, _r in TEL10)
      and announce(units=roster_names({})) == SITE_NOW)

check("...and the roster-of-one path is the one every install actually reads: "
      "an empty `inverters:` names exactly one unit",
      roster_names({}) == ["inverter-a"])

check("...and a publisher told NOTHING about the roster announces the same "
      "thing, so a caller that has not been taught about units cannot change "
      "what a live install sees",
      announce(units=()) == SITE_NOW)

m, c, a = report("roster of one, P1 off (`omit=('p1_age',)`)",
                 announce(units=roster_names({}), omit=("p1_age",)),
                 [row for row in SITE_NOW if "p1_age" not in row[0]])
check("...and the `omit` path is unchanged too: with P1 ingestion off the "
      "watchdog input is still not announced, and nothing else moved",
      (m, c, a) == ([], [], []))

# --- TEL-11 ------------------------------------------------------------------
# ⚠️ THE OPTIONAL ROW'S DEFAULT IS THE CLAIM, AND IT IS THE OPPOSITE DEFAULT
# FROM `omit`'s. `p1_age` is announced unless withdrawn; `pv` does not exist
# unless asked for. Every check above this line passes `include` nothing, so
# they are all grading the announcement an install that never set `pv_entity`
# receives - which is MOST installs, and is the acceptance criterion "unset
# behaves exactly as today".
check("⭐ TEL-11 ADDS NOTHING AT THE SHIPPED DEFAULT: a publisher that is told "
      "nothing about `pv_entity` announces the same list, in the same order, "
      "with the same bytes - so an install that never sets the option sees no "
      "new entity, no new topic and no re-announcement of an old one",
      announce(units=roster_names({})) == SITE_NOW)
_PV_TOPIC = "homeassistant/sensor/goodwe_ctl_pv/config"
_PV_PAYLOAD = ('{"name": "PV power", "object_id": "goodwe_pv_power", '
               '"unique_id": "goodwe_ctl_pv", "state_topic": "goodwe_ctl/pv", '
               '"availability_topic": "goodwe_ctl/availability", "device": '
               + _DEV + ', "unit_of_measurement": "W", "device_class": '
               '"power", "state_class": "measurement"}')
_pv_one = announce(units=roster_names({}), include=("pv",))
_m2, _c2, _a2 = diff(_pv_one, SITE_NOW)
check("⭐ AND WITH `pv_entity` SET IT IS EXACTLY ONE NEW TOPIC AND NOTHING "
      "ELSE MOVES: same eleven configs first, same bytes, same order - the "
      "additions this card makes are additions, not a re-composition",
      (_m2, _c2) == ([], []) and _a2 == [_PV_TOPIC]
      and _pv_one[:len(SITE_NOW) - 1] == SITE_NOW[:-1])
check("...and the payload is the site device, in watts, with "
      "`state_class: measurement` - which is what puts it in the recorder's "
      "statistics, and a week of statistics is the entire point of the card",
      (_PV_TOPIC, _PV_PAYLOAD, True) in _pv_one)
# ⚠️ AND IT GOES LAST, WHICH IS WHY A ROSTER OF TWO IS ASKED THE SAME QUESTION.
# Placed among the site rows it would shift every per-unit entity's position in
# the retained announcement - a zero-byte edit that re-announces inverter B.
_pv_two = announce(units=NAMES2, include=("pv",))
check("...and at a roster of two it is STILL the last config, after every "
      "per-unit row, so no existing entity's position in the retained "
      "announcement moves when the option is turned on",
      [t for t, _p, _r in _pv_two if t.startswith("homeassistant/")][-1]
      == _PV_TOPIC
      and _pv_two == announce(units=NAMES2)[:-1] + [
          (_PV_TOPIC, _PV_PAYLOAD, True)] + announce(units=NAMES2)[-1:])
check("...and `omit` still wins over `include`, so the two arguments cannot "
      "disagree about one key",
      [t for t, _p, _r in announce(units=roster_names({}), include=("pv",),
                                   omit=("pv",))
       if t == _PV_TOPIC] == [])

# ---------------------------------------------------------------------------
print("\n2. what is diffed is what goes on the wire")
# ---------------------------------------------------------------------------
# ⚠️ WITHOUT THIS PART, PARTS 1 AND 3-8 GRADE A FUNCTION THE ADD-ON MIGHT NOT
# USE. `_announce()` used to build its payloads inline; if it still did, every
# diff here would be a diff of a second implementation that happens to sit in
# the same file. This is the join.
_direct = [(t, json.dumps(cfg), True)
           for t, cfg in mqtt.discovery_payloads(NAMES2)] + \
          [(mqtt.AVAILABILITY, "online", True)]
check("`_announce()` publishes EXACTLY `discovery_payloads()`, in order, with "
      "retain set, plus the availability message - so every diff in this file "
      "is a diff of the bytes paho is handed",
      announce(units=NAMES2) == _direct)

# ---------------------------------------------------------------------------
print("\n3. a roster of two: A untouched, B added")
# ---------------------------------------------------------------------------
TWO = announce(units=NAMES2)
m, c, a = report("roster of two (inverter-a + inverter-b, SIMULATED)", TWO)
check("⭐ AC 1: AT A ROSTER OF TWO, EVERY ONE OF A's NINE MESSAGES IS STILL "
      "BYTE-IDENTICAL - nothing missing, nothing changed. This is the criterion "
      "the live install's recorded history depends on, and it is a diff of the "
      "payloads, not a reading of the code",
      (m, c) == ([], []))
check("...and the eight additions are all B's and A's own per-unit rows, and "
      "all of them are NEW topics", len(a) == 8)
check("...and A's eight entity configs are still the FIRST eight, in the same "
      "order, so a broker replaying retained messages sees them unchanged "
      "(the ninth, `availability`, is byte-identical and last - the check "
      "above is the one that covers it, as a diff rather than a prefix)",
      TWO[:len(BASELINE) - 1] == BASELINE[:-1])

# ⚠️ THE SITE SET IS `SITE_NOW`, NOT `BASELINE`. Read from `BASELINE` it would
# be short by TEL-10's three, and those three would then be counted as PER-UNIT
# rows by every line below that says "not a site id" - which silently changes
# what the sha256 pin, the uniqueness checks and part 7's fed/announced set are
# taken over.
site_ids = {c_["unique_id"] for c_ in cfgs(SITE_NOW)}
unit_cfgs = [c_ for c_ in cfgs(TWO) if c_["unique_id"] not in site_ids]
a_cfgs = [c_ for c_ in unit_cfgs if "inverter_a" in c_["unique_id"]]
b_cfgs = [c_ for c_ in unit_cfgs if "inverter_b" in c_["unique_id"]]
check("A's per-unit rows stay on THE DEVICE A ALREADY HAS - no new device is "
      "created for the first roster entry, which is the whole asymmetry",
      len(a_cfgs) == 4
      and all(c_["device"] == mqtt.DEVICE for c_ in a_cfgs))
check("⭐ AC 2: B's rows are on B's OWN device, with its own identifiers",
      len(b_cfgs) == 4
      and all(c_["device"]["identifiers"] == ["goodwe_rs485_controller_inverter_b"]
              for c_ in b_cfgs))
check("...and B's device hangs off the controller's via `via_device`, so HA "
      "shows one controller driving two inverters rather than two strangers",
      all(c_["device"].get("via_device") == mqtt.DEVICE["identifiers"][0]
          for c_ in b_cfgs))
check("...and B's device carries the SAME manufacturer and model as the site "
      "device - it is the same installation, not a second product",
      all(c_["device"]["manufacturer"] == mqtt.DEVICE["manufacturer"]
          and c_["device"]["model"] == mqtt.DEVICE["model"] for c_ in b_cfgs))

# A regression pin on the NEW payloads. ⚠️ THIS IS NOT EVIDENCE OF CORRECTNESS -
# it is generated from this branch, so it can only say "these bytes have not
# changed since the PR was reviewed". It exists because B's payloads have no
# released version to be diffed against, and a later edit that quietly changes
# a per-unit `device_class` or state topic should have to be deliberate.
_pin = hashlib.sha256(
    json.dumps(unit_cfgs, sort_keys=False).encode()).hexdigest()[:16]
print("    the eight per-unit payloads, in announcement order:")
for c_ in unit_cfgs:
    print(f"      {c_['unique_id']:<40} {c_['object_id']:<34} "
          f"{c_['state_topic']}")
print(f"    regression pin (sha256 of the eight payloads): {_pin}")
check("the per-unit payloads are the ones this PR was reviewed with "
      "(REGRESSION PIN, not a correctness claim - see the comment)",
      _pin == "eb5022b0455158d7")

# ---------------------------------------------------------------------------
print("\n4. zero overlap between A's entities and B's")
# ---------------------------------------------------------------------------
def field(rows, key):
    return [r[key] for r in rows]


for key in ("unique_id", "object_id", "state_topic"):
    av, bv = set(field(a_cfgs, key)), set(field(b_cfgs, key))
    sv = set(field(cfgs(SITE_NOW), key))
    check(f"⭐ AC 2: no `{key}` is shared between A's entities and B's, and "
          f"none of B's collides with a SITE entity that already exists",
          len(av) == 4 and len(bv) == 4 and not (av & bv) and not (bv & sv)
          and not (av & sv))

all_ids = field(cfgs(TWO), "unique_id")
check("...and across the WHOLE announcement every unique_id is distinct - an "
      "HA discovery collision is not an error, it DROPS the second entity, so "
      "one inverter's telemetry would silently stop existing",
      len(all_ids) == len(set(all_ids)) == 19)

all_topics = [t for t, _, _ in TWO]
check("...and so is every topic", len(all_topics) == len(set(all_topics)) == 20)

# ⚠️ THE SLUG COLLISION rule 9(c) DOES NOT REFUSE. `inverter-b` and `Inverter B`
# are two different names by the validator's test and one string once slugged.
COLLIDE = announce(units=["inverter-b", "Inverter B"])
col_ids = field(cfgs(COLLIDE), "unique_id")
print("    two names that slug alike ('inverter-b', 'Inverter B') ->",
      [c_["unique_id"] for c_ in cfgs(COLLIDE) if c_["unique_id"] not in site_ids])
check("two roster names that differ only in case and punctuation - which "
      "`validate.py` rule 9(c) accepts, because it refuses only EXACT "
      "duplicates - still get distinct unique_ids",
      len(col_ids) == len(set(col_ids)))

# ---------------------------------------------------------------------------
print("\n5. site-scoped sensors stay site-scoped")
# ---------------------------------------------------------------------------
# ⚠️ `p1_sample_age_s` IS A WATCHDOG INPUT (mqtt.py's `omit` note and the ESP32
# firmware's ingest guard). A second copy of it is a second answer to "is the
# meter fresh?", one of which is always the one nobody is looking at. `grid` is
# the same argument without the safety edge: there is ONE grid connection.
# ⚠️ TEL-10's THREE ARE IN HERE, AND THE ARGUMENT IS `grid`'s. There is ONE
# integrator - §6.2's architecture is one integrator on the P1 meter whose
# output is then SPLIT - so a per-inverter copy of it would invent N of a thing
# there is one of, and two of them would disagree the moment the allocator gave
# the units different shares.
SITE_ONLY = ("goodwe_ctl_grid", "goodwe_ctl_p1_age", "goodwe_ctl_circulation",
             "goodwe_ctl_phase", "goodwe_ctl_status",
             "goodwe_ctl_loop_i", "goodwe_ctl_loop_lead", "goodwe_ctl_loop_reason")
rows = []
for n in range(1, 7):
    names = [f"inverter-{chr(ord('a') + i)}" for i in range(n)]
    ids = field(cfgs(announce(units=names)), "unique_id")
    rows.append((n, [ids.count(s) for s in SITE_ONLY], len(ids)))
    print(f"    roster of {n}: {len(ids)} entities, "
          + ", ".join(f"{s}x{ids.count(s)}" for s in SITE_ONLY))
check("⭐ AC 3: `grid` and `p1_sample_age_s` are announced EXACTLY ONCE at every "
      "roster length from 1 to 6 - announcing them per inverter would invent "
      "two of a thing there is one of, and would duplicate a safety signal",
      all(counts == [1] * len(SITE_ONLY) for _n, counts, _t in rows))
check("...and the site set does not grow with the roster while the per-unit "
      "set does: 11 site entities always, 4 more per unit past the first",
      [t for _n, _c, t in rows] == [11, 19, 23, 27, 31, 35])

site_devices = {json.dumps(c_["device"], sort_keys=True)
                for c_ in cfgs(announce(units=NAMES2))
                if c_["unique_id"] in site_ids}
check("...and every site entity is still on the SITE device, at a roster of two",
      site_devices == {json.dumps(mqtt.DEVICE, sort_keys=True)})

# ---------------------------------------------------------------------------
print("\n6. the composed entity_ids")
# ---------------------------------------------------------------------------
# ⚠️ THIS SLUGIFIER IS DELIBERATELY NOT `mqtt._slug`. Importing the module's own
# spelling would make the model and the subject move together, so a mutation in
# the slugifier would change both sides equally and be invisible - "a check
# whose data cannot reach the failure it names", written in one import.
def compose(device_name, entity_name):
    """HA's entity_id for a discovery payload, per mqtt.py's own note."""
    s = re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower())
    return "sensor." + s.strip("_")


# The ONE anchor there is: mqtt.py records that the live entity is
# `sensor.goodwe_p1_sample_age`, verified against the live system 2026-08-26,
# and that the firmware subscribes to that composed name. If this model does not
# reproduce that id, the model is wrong and every claim below it is worthless.
p1 = [c_ for c_ in cfgs(BASELINE) if c_["unique_id"] == "goodwe_ctl_p1_age"][0]
check("the model reproduces the ONE composed entity_id that was verified "
      "against the live system - `sensor.goodwe_p1_sample_age`, which the ESP32 "
      "ingest guard subscribes to - from the device name and the entity name",
      compose(mqtt.DEVICE["name"], p1["name"]) == "sensor.goodwe_p1_sample_age")
check("...and that id is NOT the `object_id` in the payload "
      "(`p1_sample_age_s`), which is the whole reason the device name is "
      "load-bearing and mqtt.py says so in capitals",
      p1["object_id"] != "goodwe_p1_sample_age")

composed = [compose(c_["device"]["name"], c_["name"]) for c_ in cfgs(TWO)]
print("    every composed entity_id at a roster of two:")
for c_, eid in zip(cfgs(TWO), composed):
    print(f"      {eid:<46} <- device {c_['device']['name']!r} + "
          f"name {c_['name']!r}")
check("⭐ every composed entity_id at a roster of two is DISTINCT: no per-unit "
      "entity can land on top of a site entity that already has history",
      len(composed) == len(set(composed)) == 19)

base_composed = [compose(c_["device"]["name"], c_["name"]) for c_ in cfgs(BASELINE)]
check("...and the eight ids that exist today are unchanged by adding B, which "
      "is AC 1 restated in the units HA actually uses",
      composed[:8] == base_composed)

check("...and the per-unit ids come out SYMMETRIC even though the devices are "
      "not - sensor.goodwe_inverter_a_battery_power beside "
      "sensor.goodwe_inverter_b_battery_power - which is the entire benefit of "
      "the 'one device per inverter' design, without its cost",
      "sensor.goodwe_inverter_a_battery_power" in composed
      and "sensor.goodwe_inverter_b_battery_power" in composed)

check("...and for every per-unit entity `object_id` equals the composed id, so "
      "the two mechanisms agree and an HA that DID honour object_id would "
      "produce the same ids as one that does not",
      all("sensor." + c_["object_id"] == compose(c_["device"]["name"], c_["name"])
          for c_ in unit_cfgs))

# ---------------------------------------------------------------------------
print("\n7. the announcement and the publication agree")
# ---------------------------------------------------------------------------
# ⚠️ THE HAZARD mqtt.py's `omit` NOTE NAMES: an entity that exists and is never
# fed is a worse signal than one that does not exist. So the set of per-unit
# topics ANNOUNCED and the set `Controller.publish()` actually writes have to be
# the same set - and, because a matching set of KEYS is not a matching set of
# VALUES, the two units carry DIFFERENT numbers here so that a swap is red.
class _Store:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None


class _Pub:
    def __init__(self):
        self.last = {}
        self.units = []

    def publish(self, values):
        self.last = dict(values)

    def close(self):
        pass


def controlled(opts, soc, batt, tripped):
    """A real Controller with per-unit readings planted. SIMULATED values."""
    # ⚠️ `stop` is a REQUIRED keyword (CFG-01): a Controller that cannot stop
    # itself used to be the default one. These cases only call publish(), so any
    # Event will do - but they may not skip it.
    pub = _Pub()
    ctl = Controller(dict(opts), None, _Store(), pub, stop=asyncio.Event())
    for inv, s, b, t in zip(ctl.inverters, soc, batt, tripped):
        inv.soc, inv.batt, inv.target = s, b, -1.0 * b
        inv.sign_wd.tripped = t
    ctl.publish()
    return ctl, pub.last


# A 41 / 88 %, A discharging 1200 W, B charging 300 W, and B latched, not A.
# Every one of those six numbers differs from its sibling, which is what makes
# the assertions below able to see a swap.
ctl2, vals2 = controlled(OPTS2, [41.0, 88.0], [1200.0, -300.0], [False, True])
announced_unit_topics = {c_["state_topic"] for c_ in unit_cfgs}
published_unit_topics = {f"{mqtt.BASE}/{k}" for k in vals2 if k.startswith("unit/")}
print(f"    announced per-unit topics {len(announced_unit_topics)}, "
      f"published {len(published_unit_topics)}, "
      f"announced-but-never-fed {sorted(announced_unit_topics - published_unit_topics)}, "
      f"fed-but-never-announced {sorted(published_unit_topics - announced_unit_topics)}")
check("⭐ every per-unit entity announced is one the controller actually "
      "publishes to, and every per-unit topic published is one that was "
      "announced - no entity that nothing feeds, no topic nothing declared",
      announced_unit_topics == published_unit_topics)

check("⭐ AND THE VALUES LAND ON THE RIGHT UNIT: A's 41 % on A's topic and B's "
      "88 % on B's, not the other way round",
      vals2["unit/inverter_a/soc"] == 41.0
      and vals2["unit/inverter_b/soc"] == 88.0)
check("...and the same for battery power, whose SIGNS differ between the two "
      "units here (A discharging, B charging) so a swap cannot hide in a "
      "magnitude",
      vals2["unit/inverter_a/battery"] == 1200.0
      and vals2["unit/inverter_b/battery"] == -300.0)
check("...and the SAFETY-02 sign latch is per unit: B is latched and A is not, "
      "and the two rows say so separately",
      vals2["unit/inverter_a/sign_latch"] == "clear"
      and vals2["unit/inverter_b/sign_latch"] == "tripped")
check("...and each unit's setpoint row is its OWN share, not the site target",
      vals2["unit/inverter_a/setpoint"] == -1200.0
      and vals2["unit/inverter_b/setpoint"] == 300.0)

check("...and the SITE rows still mean what they meant: `soc` is the EMPTIEST "
      "pack (41 %, not 88 and not the mean) and `battery` is the SUM (900 W), "
      "so this card added rows beside them and did not re-point them",
      vals2["soc"] == 41.0 and round(vals2["battery"]) == 900)

_, vals1 = controlled({"meter_source": "off", "setpoint_entity": "number.sp"},
                      [55.0], [700.0], [False])
print(f"    roster of one publishes keys: {sorted(vals1)}")
check("⭐ AC 4 on the PUBLISH side: at a roster of one not one `unit/...` topic "
      "is written either, so the live install's broker sees no new topic at all",
      not [k for k in vals1 if k.startswith("unit/")])

# --- 7b. the wiring, which nothing above reaches -----------------------------
# ⚠️ FOUND BY MUTATION, NOT BY DESIGN. `units=()` in main()'s `MqttPublisher(...)`
# call survived every check in this file: parts 1-7 build the publisher
# themselves, so a controller publishing eight per-unit topics beside a
# publisher announcing none of them was GREEN. That is this repository's
# recurring defect - a check whose data cannot reach the failure it names - in
# the one line that joins the two halves together.
check("`roster_names()` and `build_roster()` give the same names in the same "
      "order, on every shape - the flat keys, a declared roster, a nameless "
      "entry and a non-dict entry - so the names the publisher is handed are "
      "the names the units answer to",
      all([inv.name for inv in build_roster(o, None)] == roster_names(o)
          for o in ({}, {"setpoint_entity": "number.sp"}, OPTS2,
                    {"inverters": [{}, {}]},
                    {"inverters": [{"name": "roof west"}, "junk"]})))
check("...and a nameless entry gets a POSITIONAL name, not `inverter-a` twice: "
      "two units under one name share one Store namespace, so one unit's "
      "SAFETY-02 latch would be the other's",
      roster_names({"inverters": [{}, {}]}) == ["inverter-1", "inverter-2"])

_call = [n for n in ast.walk(ast.parse((HERE / "app" / "main.py").read_text()))
         if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
         and n.func.id == "MqttPublisher"]
_kw = [k for c_ in _call for k in c_.keywords if k.arg == "units"]
print(f"    main.py builds MqttPublisher {len(_call)} time(s); "
      f"units= is " + (ast.unparse(_kw[0].value) if _kw else "ABSENT"))
check("⚠️ A SOURCE READ, AND THE WEAKEST CHECK IN THIS FILE - said so rather "
      "than dressed up. main()'s one `MqttPublisher(...)` hands it "
      "`roster_names(opts)`; nothing else here can reach that line, because "
      "main() opens a broker, a web server and a control loop",
      len(_call) == 1 and len(_kw) == 1
      and ast.unparse(_kw[0].value) == "roster_names(opts)")

# --- 7c. a name that is not ASCII -------------------------------------------
# HA transliterates before it slugs; this module cannot, so it treats anything
# outside ASCII as a separator. Either way the id must be an id.
_odd = announce(units=["caf\u00e9-a", "caf\u00e9-b"])
_odd_unit = [c_ for c_ in cfgs(_odd) if c_["unique_id"] not in site_ids]
print("    a roster named with non-ASCII ->",
      sorted({c_["unique_id"] for c_ in _odd_unit}))
check("every unique_id, object_id and state topic this module produces is "
      "lowercase ASCII, even from a roster name that is not - an id HA would "
      "not have composed is an entity nothing can subscribe to",
      _odd_unit
      and all(re.fullmatch(r"[a-z0-9_]+", c_[k])
              for c_ in _odd_unit for k in ("unique_id", "object_id"))
      and all(re.fullmatch(r"[a-z0-9_/]+", c_["state_topic"])
              for c_ in _odd_unit))

# The topics the SHIPPED controller actually feeds at a roster of two. Part 8
# grades every mutant announcement against this, which is what turns "the
# announcement changed" into "the announcement now claims an entity nothing
# will ever publish to".
FED = set(published_unit_topics)

# ---------------------------------------------------------------------------
print("\n8. verified by breaking it")
# ---------------------------------------------------------------------------
# ⚠️ MUTATED SOURCE, EXEC'D INTO A FRESH NAMESPACE. No file is edited and no
# `.pyc` is consulted, so TEST-15's stale-bytecode failure cannot produce a
# false survivor here. `app/mqtt.py` imports nothing from this package, which is
# what makes exec'ing it standalone honest rather than a stub.
def load(pairs):
    """`app/mqtt.py` with `pairs` applied, executed into a fresh module.

    ⚠️ EVERY ANCHOR MUST OCCUR EXACTLY ONCE. A mutation whose anchor has drifted
    out of the file would otherwise apply to nothing and be reported as a
    survivor - a false red on the instrument rather than on the module.
    """
    src = MQTT_SRC
    for old, new in pairs:
        if src.count(old) != 1:
            raise AssertionError(f"mutation anchor not unique: {old!r}")
        src = src.replace(old, new)
    mod = type(sys)("mqtt_mutant")
    exec(compile(src, "<mutant>", "exec"), mod.__dict__)
    return mod


def properties(mod):
    """Every claim this file makes, as booleans, over ONE module."""
    out = {}
    try:
        base = announce(units=["inverter-a"], module=mod)
        two = announce(units=["inverter-a", "inverter-b"], module=mod)
        none = announce(units=(), module=mod)
        col = announce(units=["inverter-b", "Inverter B"], module=mod)
        tc = cfgs(two)
        uc = [c_ for c_ in tc if c_["unique_id"] not in site_ids]
        bc = [c_ for c_ in uc if "inverter_b" in c_["unique_id"]]
        ac = [c_ for c_ in uc if "inverter_a" in c_["unique_id"]]
        comp = [compose(c_["device"]["name"], c_["name"]) for c_ in tc]
        out["A byte-identical at N=1"] = base == SITE_NOW
        out["A byte-identical at N=2"] = diff(two)[:2] == ([], [])
        out["N=1 adds no topic"] = len(base) == len(none) == len(SITE_NOW)
        out["unique_ids all distinct"] = (
            len(set(c_["unique_id"] for c_ in tc)) == len(tc))
        out["slug-alike names still distinct"] = (
            len(set(c_["unique_id"] for c_ in cfgs(col))) == len(cfgs(col)))
        out["no composed entity_id collision"] = len(set(comp)) == len(comp)
        out["object_id == composed id"] = all(
            "sensor." + c_["object_id"] == compose(c_["device"]["name"], c_["name"])
            for c_ in uc)
        out["A keeps the device it has"] = bool(ac) and all(
            c_["device"] == mqtt.DEVICE for c_ in ac)
        out["B has its own device"] = bool(bc) and all(
            c_["device"]["identifiers"] != mqtt.DEVICE["identifiers"] for c_ in bc)
        out["B is via_device the site"] = bool(bc) and all(
            c_["device"].get("via_device") == mqtt.DEVICE["identifiers"][0]
            for c_ in bc)
        out["site sensors announced once"] = all(
            [c_["unique_id"] for c_ in cfgs(announce(units=[f"i{j}" for j in range(n)],
                                                     module=mod))].count(s) == 1
            for n in (1, 2, 3) for s in SITE_ONLY)
        # ⚠️ TWO PROPERTIES THAT WERE MISSING, AND THE INSTRUMENT FOUND THEIR
        # ABSENCE ITSELF: on the first run of this table, "the unit is dropped
        # from the state topic" and "`grid` is moved to the per-unit set" were
        # both GREEN. The first collapses A's and B's four topics onto one set
        # of four, so both inverters read one number; the second announces a
        # per-inverter grid sensor, which is the exact thing AC 3 forbids. The
        # ids stayed distinct through both, so every property above missed them.
        out["per-unit state topics all distinct"] = (
            len({c_["state_topic"] for c_ in uc}) == len(uc))
        # ⚠️ GRADED AGAINST THE SHIPPED `Controller.publish()`, on purpose. An
        # announcement is only honest if something feeds it - mqtt.py's `omit`
        # note - so a per-unit entity the controller will never publish to is a
        # defect however tidy its id looks. (A rename applied to BOTH halves at
        # once is invisible here and is disclosed in part 9; a rename applied to
        # one half is what actually happens, and is red.)
        out["every announced per-unit topic is fed"] = (
            {c_["state_topic"] for c_ in uc} == FED)
    except Exception as err:  # a mutation that CRASHES proves nothing; say so
        out["did not raise"] = False
        out["__error__"] = repr(err)[:70]
    return out


SHIPPED = properties(mqtt)
check("the shipped module satisfies every property this instrument grades - "
      "so a red below is the mutation and not the instrument",
      all(v is True for k, v in SHIPPED.items()))

MUTATIONS = [
    ("⭐ RE-SCOPE A's DEVICE NAME (the destructive change, AC 5)",
     [('"name": "GoodWe",', '"name": "GoodWe Inverter A",')]),
    ("⭐ RE-SCOPE A's DEVICE IDENTIFIERS",
     [('"identifiers": ["goodwe_rs485_controller"],',
       '"identifiers": ["goodwe_inverter_a"],')]),
    ("⭐ THE TEMPTING TIDY DESIGN: one device per inverter, A included",
     [("device = DEVICE if i == 0 else unit_device(name, slug)",
       "device = unit_device(name, slug)"),
      ('prefix = f"{_title(name)} " if i == 0 else ""', 'prefix = ""')]),
    ("B's rows land on A's device instead of B's own",
     [("device = DEVICE if i == 0 else unit_device(name, slug)",
       "device = DEVICE")]),
    ("A's per-unit entity names lose the unit prefix",
     [('prefix = f"{_title(name)} " if i == 0 else ""', 'prefix = ""')]),
    ("per-unit entities announced at a roster of ONE",
     [("PER_UNIT_MIN_ROSTER = 2", "PER_UNIT_MIN_ROSTER = 1")]),
    ("the unit is dropped from the unique_id",
     [('unique_id = f"{BASE}_unit_{slug}_{key}"',
       'unique_id = f"{BASE}_unit_{key}"')]),
    ("the unit is dropped from the state topic",
     [('return f"unit/{slug}/{key}"', 'return f"unit/{key}"')]),
    ("the slug de-collision is removed (rule 9(c) does not cover this)",
     [("while slug in seen:", "while False and slug in seen:")]),
    ("every unit takes the FIRST unit's slug",
     [("for i, (name, slug) in enumerate(zip(names, slugs)):",
       "for i, (name, slug) in enumerate(zip(names, [slugs[0]] * len(slugs))):")]),
    ("`grid` is moved to the per-unit set (site sensor announced per inverter)",
     [('    ("setpoint", "setpoint", "Setpoint", "W", "power", "measurement", None),\n',
       '    ("setpoint", "setpoint", "Setpoint", "W", "power", "measurement", None),\n'
       '    ("grid", "grid_power", "Grid power", "W", "power", "measurement", None),\n')]),
    ("B's device drops `via_device`",
     [('        "via_device": DEVICE["identifiers"][0],\n', "")]),
    # --- TEL-10 ------------------------------------------------------------
    ("TEL-10's integrator row is DROPPED (published to a topic nothing "
     "announced, which AC 4 forbids in the other direction)",
     [('    ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n', "")]),
    ("TEL-10's integrator is announced PER INVERTER - two answers to a "
     "question §6.2 says there is one integrator for",
     [('    ("sign_latch", "sign_latch", "Sign latch", None, None, None, "mdi:flash-alert"),\n',
       '    ("sign_latch", "sign_latch", "Sign latch", None, None, None, "mdi:flash-alert"),\n'
       '    ("loop_i", "loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n')]),
    # ⚠️ THE ONE THAT CHANGES NO BYTE OF ANY PAYLOAD. Every id, name, unit and
    # topic is identical; only the POSITION of the integrator row moves. A
    # broker replaying the retained set then hands HA a different sequence, and
    # part 1's claim is about order as much as about bytes - which is why it is
    # asserted as a LIST and not as a set.
    ("TEL-10's integrator row is announced BEFORE release/1.0's nine",
     [('    ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n', ""),
      ('    ("setpoint", "goodwe_setpoint", "Setpoint", "W", "power", "measurement", None),\n',
       '    ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n'
       '    ("setpoint", "goodwe_setpoint", "Setpoint", "W", "power", "measurement", None),\n')]),
]

survivors = []
for label, pairs in MUTATIONS:
    got = properties(load(pairs))
    broke = sorted(k for k, v in SHIPPED.items()
                   if v is True and got.get(k) is not True)
    if not broke:
        survivors.append(label)
    print(f"    {'RED ' if broke else 'GREEN'} {label}")
    if broke:
        print(f"           breaks: {', '.join(broke)}")
    elif got.get("__error__"):
        print(f"           raised: {got['__error__']}")
check(f"⭐ ALL {len(MUTATIONS)} MUTATIONS ARE RED, each on a NAMED property: "
      "re-scoping A's device, the tidy one-device-per-inverter design, B "
      "landing on A's device, per-unit entities at a roster of one, a unit "
      "dropped from an id or a topic, two units sharing a slug, and a site "
      "sensor announced per inverter. A mutation nothing catches is a claim "
      "nothing is watching",
      not survivors)
if survivors:
    print("       SURVIVED:", "; ".join(survivors))

# ⚠️ AC 6, THE FIXTURE'S OWN NON-VACUITY. If A and B were given the same name
# the whole of part 4 would still be green and would be observing nothing.
same = announce(units=["inverter-a", "inverter-a"])
swapped = announce(units=["inverter-b", "inverter-a"])
check("...and the fixture's two units are genuinely distinguishable: swapping "
      "their order changes the announcement, and a roster of two IDENTICAL "
      "names - which rule 9(c) refuses, and which this file therefore never "
      "relies on - would still not collide",
      swapped != TWO
      and len({c_["unique_id"] for c_ in cfgs(same)}) == len(cfgs(same)))

# ---------------------------------------------------------------------------
print("\n9. what this file CANNOT see, enumerated and sized")
# ---------------------------------------------------------------------------
for line in [
    "1. HA ITSELF IS NOT IN THE LOOP. `compose()` in part 6 is a MODEL of how "
    "HA 2026.8 builds an entity_id, anchored on exactly ONE measured id "
    "(sensor.goodwe_p1_sample_age, mqtt.py's note, verified 2026-08-26). If HA "
    "changes that rule, or if it differs for a device with `via_device`, every "
    "id claim here moves with it and this file stays green. SIZE: the 8 "
    "per-unit ids at N=2. It cannot be closed from this repository - it needs a "
    "broker, an HA, and the owner present.",
    "2. THE TICKET AND THE MODULE DISAGREE ABOUT THE LIVE IDs. The card quotes "
    "`sensor.goodwe_rs485_controller_goodwe_battery_soc`; mqtt.py records "
    "`sensor.goodwe_p1_sample_age`. Those cannot both come from one composition "
    "rule - the first is the id an EARLIER device name composed, and HA keeps "
    "whatever it composed at first discovery. This card does not need to "
    "resolve it, because it renames no device and no entity, and the byte-diff "
    "protects both spellings equally. SIZE: unresolved, 8 live entity ids; it "
    "wants one look at the live registry, which this branch must not take.",
    "3. THE PER-UNIT PAYLOADS HAVE NO RELEASED BASELINE, AND NEITHER DO "
    "TEL-10's THREE. A's are diffed against release/1.0; B's can only be "
    "diffed against this branch (the sha256 pin in part 3), and `TEL10` above "
    "is the same kind of pin for the three site rows this branch adds. A wrong "
    "`device_class` or unit on any of them is therefore caught as a CHANGE "
    "from review, not as an error. SIZE: 4 per-unit sensors x 4 optional "
    "fields, plus 3 site sensors x 4.",
    "4. NO CHECK HERE RUNS A BROKER. `_announce()` is graded through a fake "
    "paho client, so QoS, retain semantics on a real broker, and what HA does "
    "with a retained config for a device it has never seen are all out of "
    "scope. SIZE: the whole transport.",
    "5. THE SLUG DE-COLLISION IS ORDER-DEPENDENT. `inverter-b` before "
    "`Inverter B` gives the SECOND one the `_2` suffix; reordering the roster "
    "swaps which unit's ids move. Rule 9(c) does not refuse slug-alike names, "
    "so this is reachable from a config an operator can write. SIZE: 4 ids on "
    "one unit, and only when two names slug alike. Filed as a follow-up rather "
    "than fixed in validate.py here, because a new refusal at startup on a "
    "LIVE install is a bigger change than this card is allowed to be.",
    "7. PART 8 MUTATES `app/mqtt.py` ONLY. `Controller.publish()`'s four "
    "per-unit lines are graded by part 7 - with different SoC, different "
    "battery sign and different latch states on the two units, so a swap or a "
    "mis-paired slug is red - but no automated table mutates main.py here. "
    "Those mutations were run by hand through `tools/mutation_run.sh` and are "
    "recorded in the PR. SIZE: 4 lines. Also invisible: a rename of "
    "`unit_state_key()` applied to BOTH halves at once, which is harmless "
    "(the topics are internal) but would move every per-unit state topic "
    "silently.",
    "6. `inverter AC power` AND PER-UNIT HEARTBEAT AGE ARE NOT ANNOUNCED. The "
    "card lists both as per-unit; neither has a source in the add-on today "
    "(nothing reads an AC-power register per unit, and `inv.last_beat` is a "
    "monotonic loop clock). Announcing them would create entities nothing "
    "feeds. SIZE: 2 sensors x N units, deliberately absent, named in the PR.",
]:
    print("    " + line)

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