Merge release/1.0 (43b2740) into DOC-5-watchdog to sync base

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

FIELD-GUIDE.md

@@ -4,6 +4,10 @@ The guide technicians carry is the PDF:
 
 **`docs/GoodWe-RS485-Field-Guide.pdf`** — 10 pages, A4, print it double-sided.
 
+⚠️ **Outdated: it is v0.1.0.** Its `slew_w` row is corrected in `build_guide.py`
+and `field-guide.html`, but the PDF has not been rebuilt (see below). Where the
+guide disagrees with `goodwe_controller/DOCS.md`, DOCS.md wins.
+
 It covers, in the order the work is done: pre-visit checks, hardware, firmware,
 add-on install and configuration, how to read the Web UI, the six commissioning
 gates, the e-stop (and the statement a site without one must sign), proving a
@@ -18,7 +22,8 @@ copy cannot drift from the PDF someone is holding in a cellar.
 python3 docs/build_guide.py     # regenerates field-guide.html and the PDF
 ```
 
-Requires Chrome or Chromium (used headless for print-to-PDF). Screenshots live
+Requires Chrome or Chromium (used headless for print-to-PDF); `CHROME` in the
+script is a Windows path, so edit it on any other machine. Screenshots live
 in `docs/img/` and are base64-embedded, so `field-guide.html` is standalone.
 
 ⚠️ **Every section must stay under the A4 printable height** (1039 px at 96 dpi

README.md

@@ -30,7 +30,7 @@ goodwe_controller/         the add-on
                            returns, run against its own release image
 firmware/goodwe-master.yaml  ESPHome config for the T-CAN485
 estop/rs485_log.py           optional RS485 e-stop / bus witness
-docs/GoodWe-RS485-Field-Guide.pdf  the guide techs carry (10pp, A4)
+docs/GoodWe-RS485-Field-Guide.pdf  the guide techs carry (10pp, A4; outdated)
 docs/build_guide.py          its single source - rebuilds the HTML and PDF
 FIELD-GUIDE.md             how to edit and rebuild the guide
 docs/RELEASE.md            how a change reaches the battery - `master` is
@@ -50,6 +50,9 @@ https://gittea.kammenstraatha.duckdns.org/admin/goodwe-addon
 
 then install **GoodWe RS485 Controller**.
 
+Then turn on **Watchdog** on the add-on's Info tab — see
+[Before you start](goodwe_controller/DOCS.md#before-you-start).
+
 ⚠️ **The repository must be PUBLIC for this to work.** Home Assistant clones it
 anonymously; against a private repo the store shows nothing and gives no useful
 error. If it must stay private, install locally instead: copy
@@ -79,8 +82,8 @@ machine itself dying:
 2. **Wind-down before firmware updates** — 0 W written before the update starts.
 3. **RS485 e-stop (optional)** — writes 0 W after 30 s of total bus silence.
 
-Sites sold without the e-stop must have the acknowledgement in `FIELD-GUIDE.md`
-§10 signed.
+Sites sold without the e-stop must have the client acknowledgement in the field
+guide (Step 7b) signed.
 
 ## Development
 

docs/build_guide.py

@@ -95,7 +95,7 @@ figcaption { font-size:9pt; color:#5b6169; margin-top:1.2mm; }
 <div class="page" id="p1">
   <div class="hdr">
     <div><h1>GoodWe RS485 Controller</h1>
-         <div class="sub">Field installation &amp; commissioning guide — v0.1.0</div></div>
+         <div class="sub">Field installation &amp; commissioning guide — v0.1.0, <b>outdated</b>: where it disagrees with DOCS.md, DOCS.md wins</div></div>
     <small>Installer copy</small>
   </div>
 
@@ -503,7 +503,7 @@ figcaption { font-size:9pt; color:#5b6169; margin-top:1.2mm; }
   <table>
     <tr><th style="width:30mm">Setting</th><th style="width:16mm">Value</th><th>Reason</th></tr>
     <tr><td><code>gain</code></td><td>0.6</td><td>Inverter needs 3–6 s to settle (1.4 s dead time, 94 % by 3.3 s) against a ~5 s cycle. <b>At the limit — do not raise</b></td></tr>
-    <tr><td><code>slew_w</code></td><td>1000</td><td>Most of a correction in the first cycle without running far ahead of the hardware</td></tr>
+    <tr><td><code>slew_w</code></td><td>0</td><td>0 follows the rail (<code>max_w</code>&nbsp;/&nbsp;9). A faster fixed value is <b>refused at startup</b> (SAFETY-39) — see DOCS.md</td></tr>
     <tr><td><code>deadband_w</code></td><td>15</td><td>Measured residual: mean 15.4 W, max 27 W. At 10 W the command never rests</td></tr>
     <tr><td><code>saturation_w</code> / <code>cycles</code></td><td>500 / 3</td><td>Divergence = inverter at a limit → magnitude may fall, never rise. <b>The 3-cycle term is essential</b>: tested instantly it fires on every large correction</td></tr>
     <tr><td><code>step_w</code></td><td>10</td><td>Register is 1 W but response lands on a ~17.6 W ladder. Finer is meaningless</td></tr>

docs/field-guide.html

@@ -60,7 +60,7 @@ figcaption { font-size:9pt; color:#5b6169; margin-top:1.2mm; }
 <div class="page" id="p1">
   <div class="hdr">
     <div><h1>GoodWe RS485 Controller</h1>
-         <div class="sub">Field installation &amp; commissioning guide — v0.1.0</div></div>
+         <div class="sub">Field installation &amp; commissioning guide — v0.1.0, <b>outdated</b>: where it disagrees with DOCS.md, DOCS.md wins</div></div>
     <small>Installer copy</small>
   </div>
 
@@ -468,7 +468,7 @@ figcaption { font-size:9pt; color:#5b6169; margin-top:1.2mm; }
   <table>
     <tr><th style="width:30mm">Setting</th><th style="width:16mm">Value</th><th>Reason</th></tr>
     <tr><td><code>gain</code></td><td>0.6</td><td>Inverter needs 3–6 s to settle (1.4 s dead time, 94 % by 3.3 s) against a ~5 s cycle. <b>At the limit — do not raise</b></td></tr>
-    <tr><td><code>slew_w</code></td><td>1000</td><td>Most of a correction in the first cycle without running far ahead of the hardware</td></tr>
+    <tr><td><code>slew_w</code></td><td>0</td><td>0 follows the rail (<code>max_w</code>&nbsp;/&nbsp;9). A faster fixed value is <b>refused at startup</b> (SAFETY-39) — see DOCS.md</td></tr>
     <tr><td><code>deadband_w</code></td><td>15</td><td>Measured residual: mean 15.4 W, max 27 W. At 10 W the command never rests</td></tr>
     <tr><td><code>saturation_w</code> / <code>cycles</code></td><td>500 / 3</td><td>Divergence = inverter at a limit → magnitude may fall, never rise. <b>The 3-cycle term is essential</b>: tested instantly it fires on every large correction</td></tr>
     <tr><td><code>step_w</code></td><td>10</td><td>Register is 1 W but response lands on a ~17.6 W ladder. Finer is meaningless</td></tr>

goodwe_controller/DECISIONS/SAFETY-61-control-task-death.md

@@ -0,0 +1,106 @@
+# SAFETY-61 — the control task's death ends the process
+
+## 0. The defect
+
+`amain()` started `run_control()` as a bare task and waited on `stop`. Nothing
+watched the task. An uncaught exception in `read_inputs()`, `cycle()` or
+`publish()` ended the task silently, and `stop` was never set. The web UI, P1
+and tariff tasks kept running, and the Supervisor saw a healthy container. The
+traceback appeared only at shutdown. The ESP32's layer-1 veto zeroed the
+inverter about 30 s later, so the hardware was safe. The pack then sat at 0 W
+indefinitely while the add-on showed green.
+
+`test_control_task.py` reproduces this on the pre-fix tree. The child process
+reaches its 40 s deadline, has written no 0 W, and has logged nothing.
+
+## 1. Decision: exit with status 70 after the ordinary wind-down
+
+A done-callback on the control task (`control_task_done()`) handles the death.
+It logs the traceback at CRITICAL at the moment of death, then sets `stop`.
+`amain()` then runs the same wind-down that SIGTERM runs (`shutdown()` →
+`command_site_zero()`, then cancel, then clean up). After that it raises
+`ControlTaskDied`, and `main()` turns that into `SystemExit(70)` (sysexits
+EX_SOFTWARE). A normal stop and `run_control()`'s own refusal path both set
+`stop` before the task ends, so the callback ignores them. The refusal still
+exits with 78.
+
+**Why not restart the loop in the same process:**
+
+* `cycle()` writes controller state before its awaits: `cycle_at`,
+  `peak_guard`, `allocation`, then `target`/`reason`, then the writes. An
+  exception between those steps leaves the controller half-updated. A loop
+  restarted in the same process would continue from a state that no boot ever
+  produces. A process restart rebuilds the controller from the Store, which is
+  the path every boot already runs.
+* An unbounded retry is the invisible spin the card forbids. A bounded retry
+  would reimplement the Supervisor watchdog's own throttle: 10 restarts per
+  30 min, and 5 attempts with backoff (`apps/const.py`).
+
+## 2. What happens after the exit, verified rather than assumed
+
+* **Exit status reaches Docker.** The add-on image (s6-overlay, `init: false`,
+  `CMD /run.sh`) was run with `exec python3 -c 'raise SystemExit(70)'`.
+  `docker inspect` gave ExitCode 70. The same test with exit 0 gave 0, and an
+  uncaught exception gave 1.
+* **Supervisor 2026.08.0** (`ghcr.io/home-assistant/amd64-hassio-supervisor`,
+  source read from the image):
+  * A non-zero container exit is recorded as `ContainerState.FAILED`
+    (`docker/monitor.py`). The Supervisor logs *"App … exited with non-zero
+    exit code 70"* at ERROR (`apps/app.py container_state_changed`).
+  * The Supervisor restarts the add-on **only if the per-install Watchdog
+    toggle is on**. `watchdog_container()` returns when `not self.watchdog`.
+    The toggle defaults to **off** (`apps/validate.py`:
+    `vol.Optional(ATTR_WATCHDOG, default=False)`).
+  * Add-on containers get **no Docker restart policy**. The only
+    `restart_policy=` in the Supervisor belongs to the observer.
+  * `boot: auto` in `config.yaml` only controls starting at host boot. The
+    `watchdog:` key in `config.yaml` is a health-check URL, not a restart
+    policy, and this add-on does not set it.
+* **So on a default install the add-on stays stopped after this exit.** The UI
+  shows it as stopped, its log holds the traceback, and the Supervisor log holds
+  an ERROR line. The pack is at 0 W, which is the same physical state as before
+  this card. The difference is that 0 W is written immediately instead of after
+  the ~30 s veto, and the add-on no longer shows green. With the Watchdog toggle
+  on, the Supervisor restarts the add-on.
+* **Not verified:** an actual restart by a running Supervisor. That would need a
+  live Supervisor, and the only one available is the live house.
+
+## 3. Left open for the owner
+
+* Whether `DOCS.md` should tell installers to enable Watchdog. DOCS.md is not
+  this card's file.
+* Setting the Watchdog toggle from inside the add-on (`/addons/self/options`)
+  was rejected here, because it would change a user setting without the user
+  knowing.
+
+## 4. Scope: the siblings, and why they are not supervised
+
+* `run_tariff_refresh()` and the P1 HTTP and HA transports already catch
+  `Exception` inside their loops by design, and the values they feed are
+  age-bounded downstream.
+* The P1 MQTT source *returns* on purpose when it is misconfigured. Making every
+  long-lived task's end exit the process would turn that deliberate fail-safe
+  into a restart loop.
+* Only the control task is supervised.
+
+## 5. `hass.py`: an undecodable body is an absence
+
+A non-JSON `200` served as `text/html` was **already** treated as an absence,
+because aiohttp's `ContentTypeError` is a `ClientError`. The reads that actually
+escaped were these three:
+
+* a malformed body labelled `application/json`, which raises `JSONDecodeError`
+  (a `ValueError`);
+* a JSON body that is not an object, which raises `AttributeError` in
+  `number()` and `limits()`;
+* an undecodable error body in `call()`, which raises `UnicodeDecodeError`.
+
+All three were measured against aiohttp 3.13.5. The first two now read as
+`None`, and the third is now a failed write (`False`).
+
+## 6. Status row
+
+`checks()` now starts with a *control loop* row showing how long ago the loop
+last completed a pass. The row turns red once that age exceeds
+`FIRMWARE_INGEST_VETO_S`. The heartbeat is sent from this loop, so past that age
+layer 1 has zeroed the inverter.

goodwe_controller/DOCS.md

@@ -4,8 +4,8 @@ Drives a GoodWe ES/BP battery inverter over its RS485 meter bus: holds net grid
 exchange at zero, and runs a monthly battery maintenance cycle so the BMS can
 balance cells and recalibrate its coulomb counter.
 
-Installers: read `FIELD-GUIDE.md` in the repository. It is not optional reading —
-it contains the commissioning gates and the failure modes.
+Installers: the printed field guide (`docs/GoodWe-RS485-Field-Guide.pdf`) is
+v0.1.0 and outdated; where it disagrees with this page, this page wins.
 
 ## Before you start
 
@@ -16,6 +16,11 @@ You need:
 - A T-CAN485 (ESP32) flashed with `firmware/goodwe-master.yaml`
 - A grid-power sensor already working in Home Assistant, updating every ~5–10 s
 
+After installing, turn on **Watchdog** on the add-on's Info tab. A crashed
+control loop stops the add-on at 0 W, and the Supervisor restarts it only when
+Watchdog is on (it is off by default). See
+`DECISIONS/SAFETY-61-control-task-death.md`.
+
 ## The safety model, in short
 
 The inverter **holds its last command forever** — it has no meter-timeout. So:
@@ -165,12 +170,14 @@ inverters:
     soc_entity: sensor.goodwe_a_soc
     batt_entity: sensor.goodwe_a_battery_power
     max_w: 5000
+    bms_soc_floor: 10
     capacity_kwh: 15.36
   - name: inverter-b
     setpoint_entity: number.goodwe_b_setpoint_w
     soc_entity: sensor.goodwe_b_soc
     batt_entity: sensor.goodwe_b_battery_power
     max_w: 4600
+    bms_soc_floor: 10
     capacity_kwh: 20.48
 ```
 
@@ -190,16 +197,18 @@ inverter's own **AC** power. If it is DC the AC-side ceiling is nearer 4324 W.
 | `name` | yes | Short identifier. It namespaces this unit's SAFETY-02 latch, heartbeat and write bookkeeping in the add-on's state file, so **two units may not share one** |
 | `setpoint_entity` | yes | This inverter's own ESPHome `number.*_goodwe_setpoint_w` |
 | `max_w` | yes | The supervised cap on what may be written to **this** inverter. The site's ceiling is the **sum** |
+| `bms_soc_floor` | yes | This pack's own BMS cut-out percent (SAFETY-32). See [`bms_soc_floor`](#tariff-all-optional) |
 | `soc_entity` | | This pack's state of charge |
 | `batt_entity` | | This unit's battery power, `+` = discharging |
 | `batt_invert` | | Flip if needed |
+| `capacity_kwh` | | This pack's usable capacity. State it on every entry or on none — see [OPT-09](#how-the-site-target-is-divided-between-the-packs-opt-09) |
 
 ⚠️ **Nothing is inherited.** An entry that leaves out `soc_entity` has **no SoC
 sensor** — it does *not* fall back to the flat one. That is deliberate: every
 per-device value became per-instance in 0.5.x precisely so that no inverter is
 ever graded against a sibling's numbers, and a silent fallback would rebuild
-that in YAML. `name`, `setpoint_entity` and `max_w` are required of every entry
-for the same reason: there is no rating to borrow.
+that in YAML. `name`, `setpoint_entity`, `max_w` and `bms_soc_floor` are required
+of every entry for the same reason: there is no rating or cut-out to borrow.
 
 ⚠️ **Clear the flat `setpoint_entity` when you declare a roster,** or the add-on
 refuses to start and says so. It is the single-inverter form and it is not read
@@ -600,7 +609,7 @@ than in YAML nobody reviews underneath a safety input.
 > `ha_signed`.** It reads the same meter and produces the same numbers, but it
 > polls the meter directly instead of watching a Home Assistant entity — and
 > that is the difference between an age sensor a watchdog can threshold and one
-> it cannot. See "`sensor.p1_sample_age_s`" below; `ha_signed` remains for
+> it cannot. See "`sensor.goodwe_p1_sample_age`" below; `ha_signed` remains for
 > installs where the meter is only reachable through Home Assistant.
 
 > ⚠️ **`ha_dsmr` and `mqtt_p1` have never processed a telegram from real
@@ -755,7 +764,7 @@ connection-level reading is still accepted on its own.
 
 **Why this mode exists:** every HTTP response is an *arrival*. The meter
 answered, now, with its current reading — whether or not the number moved. That
-is the signal `sensor.p1_sample_age_s` needs and the one Home Assistant cannot
+is the signal `sensor.goodwe_p1_sample_age` needs and the one Home Assistant cannot
 give it at all (see the note below). It is also simply fewer moving parts: the
 five-second cadence is the meter's own, rather than an integration's polling of
 it re-published as a state change.
@@ -792,12 +801,13 @@ as `… 4 rejected, measurement unchanged for 312 s`. On a house drawing real
 power that figure stays in the seconds; minutes of it while the load is clearly
 not near zero is the meter to go and look at.
 
-#### `sensor.p1_sample_age_s`
+#### `sensor.goodwe_p1_sample_age`
 
 Published over MQTT discovery whenever a broker is available: **seconds since the
 newest accepted telegram**, refreshed every second rather than only when a
 telegram lands. The ESP32's stale-input watchdog subscribes to this exact entity
-id, so do not rename it.
+id; do not rename the `GoodWe` device or this entity, since HA composes the id
+from both (see [What appears in Home Assistant at two inverters](#what-appears-in-home-assistant-at-two-inverters)).
 
 The reason it is recomputed against the clock is that Home Assistant only pushes
 a state when the state *changes*. A meter sitting at a genuinely constant reading
@@ -833,7 +843,7 @@ behind it would trip the firmware watchdog on a system that is working fine.
 > went **42.2 s and 97.0 s** between changes, and 23 Aug peaks at 29.1 s — all
 > past the default `meter_max_age_s` of 30.
 >
-> So `sensor.p1_sample_age_s` on `ha_signed` is safe to *read*, and it is
+> So `sensor.goodwe_p1_sample_age` on `ha_signed` is safe to *read*, and it is
 > correct whenever the value is moving, but it must not yet be thresholded by
 > the ESP32 stale-input watchdog: a quiet house would trip the battery to 0 W.
 > Raising `meter_max_age_s` is **not** the fix — the two conditions produce an
@@ -2820,8 +2830,8 @@ a **copy** of the state file and reports the difference.
         --live-entities /w/live-entities.txt \
         --reverse --round-trip'
 
-Exit `0` clean · `1` at least one finding · `2` could not run. ⚠️ **`2` is not a
-pass.**
+Exit `0` clean · `1` at least one finding · `2` could not run · `3` clean but
+degraded (`--no-live-entities`). ⚠️ **`2` and `3` are not a pass.**
 
 ### The owner steps, because this session could not take them
 
@@ -2843,7 +2853,8 @@ check, **before** the roster is declared:
 
 2. **The options blob**, as `--before`. It is not reachable through the Home
    Assistant REST API and no entity carries it; export it from the Supervisor,
-   the same way `tools/predeploy_check.py` needs it. Write the roster you intend
+   the same way `tools/predeploy_check.py` needs it: the bare options object,
+   not the `ha apps info` wrapper, which is refused. Write the roster you intend
    to deploy into a second file and pass it as `--after`.
 
 3. **The entity list**, as `--live-entities` — one entity id per line:
@@ -2857,7 +2868,8 @@ check, **before** the roster is declared:
    hand back a clean bill of health for a half of the transition it did not
    check. If you genuinely have to run without it, pass `--no-live-entities`:
    the run then carries a standing banner in the report header **and** on the
-   exit line saying the orphan check was not graded. The asymmetry used to be
+   exit line saying the orphan check was not graded, and a clean result exits
+   3, not 0. The asymmetry used to be
    the other way round — an unreadable file refused, a missing flag passed with
    exit 0 and no disclosure at all — and the owner runs this once, at
    commissioning, under time pressure.
@@ -3005,7 +3017,9 @@ Run it inside the container, so the validator half actually runs:
       --deployed-version 0.7.0 --options /w/live-options.json
 
 Export `live-options.json` from the Supervisor first — the options blob is not
-reachable through the Home Assistant REST API, and no entity carries it.
+reachable through the Home Assistant REST API, and no entity carries it. Pass the
+bare object (`ha apps info <slug> --raw-json | jq .data.options`); the wrapped
+output is refused.
 
 **Exit codes.** `0` cleared · `1` REFUSED, do not deploy · `2` NOT CLEARED,
 meaning a check did not run. ⚠️ **2 is not a pass.** On a bare host without the

goodwe_controller/app/hass.py

@@ -61,10 +61,17 @@ class HomeAssistant:
                 if resp.status == 404:
                     return None
                 resp.raise_for_status()
-                return await resp.json()
-        except (aiohttp.ClientError, TimeoutError) as err:
+                obj = await resp.json()
+        # SAFETY-61: an undecodable body is an absence like any other failed
+        # read, never an exception into the control loop.
+        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
             self._moan(f"read:{entity_id}", "read %s failed: %s", entity_id, err)
             return None
+        if not isinstance(obj, dict):
+            self._moan(f"shape:{entity_id}", "read %s: not a state object: %.80r",
+                       entity_id, obj)
+            return None
+        return obj
 
     async def number(self, entity_id: str, invert: bool = False) -> float | None:
         """Numeric state, or None. Never substitutes a default."""
@@ -117,7 +124,8 @@ class HomeAssistant:
                                domain, service, resp.status, body[:200])
                     return False
                 return True
-        except (aiohttp.ClientError, TimeoutError) as err:
+        # ValueError: an undecodable error body (SAFETY-61) is a failed write.
+        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
             self._moan(f"svcerr:{domain}.{service}", "service %s.%s failed: %s",
                        domain, service, err)
             return False

goodwe_controller/app/main.py

@@ -25,9 +25,11 @@ battery to zero on its own. Never "hold the last value to be safe".
 
 import asyncio
 import contextlib
+import functools
 import json
 import logging
 import signal
+import time
 from datetime import datetime, timedelta, timezone
 
 import aiohttp
@@ -74,8 +76,8 @@ from .price_history import PriceHistory
 from .static_tariff import resolve_tariff, static_status
 from .tariff import PeakStore, PeakTracker
 from .store import migrate_maint_keys, migrate_unit_keys
-from .validate import (ConfigRefused, refuse, validate_startup_config,
-                       validate_unit_ranges)
+from .validate import (FIRMWARE_INGEST_VETO_S, ConfigRefused, refuse,
+                       validate_startup_config, validate_unit_ranges)
 from .watchdog import (
     AGAINST_METER, AGAINST_OWN_POWER, SIGN_OPTION_DEFAULTS,
     SignWatchdog, SignWatchdogConfig, loop_is_following,
@@ -985,6 +987,8 @@ class Controller:
         # the value a clamp bit is a value written to ONE device.
         self.loop_clamped: tuple[bool, str] = (False, "")
         self.stopping = False
+        # SAFETY-61. time.monotonic() at the end of the last run_control() pass.
+        self.loop_pass_mono = None
 
     # -- helpers -------------------------------------------------------------
     def log_event(self, msg: str) -> None:
@@ -4397,6 +4401,7 @@ class Controller:
             await self.send_heartbeats(asyncio.get_running_loop().time(),
                                        heartbeat)
             self.publish()
+            self.loop_pass_mono = time.monotonic()
             await asyncio.sleep(1)
 
     async def send_heartbeats(self, loop_now: float, heartbeat_s: float) -> None:
@@ -4596,6 +4601,22 @@ class Controller:
         """
         o = self.o
         out = []
+        # SAFETY-61. First, so a stalled loop is the headline: the heartbeat is
+        # sent from this loop, so past layer 1's veto the inverter is at 0 W.
+        if self.loop_pass_mono is None:
+            out.append({"ok": True, "warn": False,
+                        "text": "control loop: starting"})
+        else:
+            _pass_age = time.monotonic() - self.loop_pass_mono
+            if _pass_age > FIRMWARE_INGEST_VETO_S:
+                out.append({"ok": False, "warn": False,
+                            "text": f"control loop: NO PASS FOR {_pass_age:.0f} s"
+                                    f" (layer 1 zeroes the inverter after "
+                                    f"{FIRMWARE_INGEST_VETO_S:.0f} s without a "
+                                    f"heartbeat)"})
+            else:
+                out.append({"ok": True, "warn": False,
+                            "text": f"control loop: last pass {_pass_age:.0f} s ago"})
         if self.p1_enabled:
             age = self.p1.published_age_s
             if self.p1.stale:
@@ -5449,6 +5470,30 @@ class Controller:
         return None
 
 
+# SAFETY-61. Exit status when the control task dies; 70 is sysexits(3)
+# EX_SOFTWARE. See DECISIONS/SAFETY-61-control-task-death.md.
+EXIT_CONTROL_DIED = 70
+
+
+class ControlTaskDied(RuntimeError):
+    """The control task ended while nothing had asked it to stop."""
+
+
+def control_task_done(task: asyncio.Task, *, controller, stop: asyncio.Event,
+                      died: list) -> None:
+    """Done-callback on the control task: log its death now and wake amain()
+    to wind down and exit (SAFETY-61, see its DECISIONS record)."""
+    if stop.is_set() or controller.stopping:
+        return          # the normal wind-down, or run_control()'s refusal path
+    exc = None if task.cancelled() else task.exception()
+    _LOG.critical("CONTROL LOOP DIED (%s) - commanding 0 W and exiting with "
+                  "status %d", "cancelled" if task.cancelled() else
+                  (repr(exc) if exc else "returned"), EXIT_CONTROL_DIED,
+                  exc_info=exc)
+    died.append(exc)
+    stop.set()
+
+
 async def amain() -> None:
     opts = load_options()
     logging.basicConfig(
@@ -5546,7 +5591,13 @@ async def amain() -> None:
             with contextlib.suppress(NotImplementedError):
                 loop.add_signal_handler(sig, stop.set)
 
-        tasks = [asyncio.create_task(controller.run_control())]
+        # SAFETY-61: the control task's death ends the process, after the
+        # wind-down below. See DECISIONS/SAFETY-61-control-task-death.md.
+        died: list = []
+        control = asyncio.create_task(controller.run_control())
+        control.add_done_callback(functools.partial(
+            control_task_done, controller=controller, stop=stop, died=died))
+        tasks = [control]
 
         # OPT-04. The tariff refresh runs as its OWN long-lived task, for the
         # reason P1 ingestion does and then some: `resolve_tariff()` may open an
@@ -5581,10 +5632,14 @@ async def amain() -> None:
         for task in tasks:
             task.cancel()
         for task in tasks:
+            if task is control and died:
+                continue            # already logged by control_task_done()
             with contextlib.suppress(asyncio.CancelledError):
                 await task
         await runner.cleanup()
         _LOG.info("stopped")
+        if died:
+            raise ControlTaskDied(died[0])
         # The late-arriving-device refusal from run_control(). Re-raised only
         # now, so the wind-down above still ran and the inverter is at 0 W.
         if controller.refused:
@@ -5603,6 +5658,9 @@ def main() -> None:
         # EX_CONFIG, so even a script reading the exit status can tell a bad
         # config from a bug.
         raise SystemExit(78)
+    except ControlTaskDied:
+        # No second traceback: control_task_done() logged it at the death.
+        raise SystemExit(EXIT_CONTROL_DIED)
 
 
 if __name__ == "__main__":

goodwe_controller/config.yaml

@@ -93,7 +93,7 @@ options:
   # against a sibling's numbers; an entry silently inheriting `soc_entity` would
   # rebuild that defect in YAML. `name`, `setpoint_entity`, `max_w` and
   # `bms_soc_floor` are therefore REQUIRED of every entry (the schema below enforces it, and
-  # validate.py rule 9 enforces it again for the callers the Supervisor never
+  # validate.py rules 8 and 9 enforce it again for the callers the Supervisor never
   # sees). An omitted `soc_entity`/`batt_entity` means THIS UNIT HAS NO SUCH
   # SENSOR - `inputs_ok` already stops the site for that - and never means
   # "use the other one's".
@@ -224,7 +224,7 @@ options:
   # ha_signed reads ONE signed HA entity (+ import / - export), which is what a
   # HomeWizard P1 publishes and what ha_dsmr cannot consume.
   # homewizard_local skips Home Assistant entirely and polls the meter's own
-  # local API. ⚠️ It is the ONLY mode whose sensor.p1_sample_age_s measures when
+  # local API. ⚠️ It is the ONLY mode whose sensor.goodwe_p1_sample_age measures when
   # the meter REPORTED rather than when the value last CHANGED - HA emits
   # nothing at all for a repeated reading - so it is the only mode a firmware
   # watchdog may threshold. See DOCS.md.
@@ -916,12 +916,13 @@ schema:
   # batteries, whereas an absent capacity has an honest answer - no opinion
   # about the split, every unit weighs 1.0, which is what shipped.
   #
-  # ⚠️ THREE KEYS ARE REQUIRED PER ENTRY AND THAT IS DELIBERATE. An entry with
+  # ⚠️ FOUR KEYS ARE REQUIRED PER ENTRY AND THAT IS DELIBERATE. An entry with
   # no `name` has no Store namespace and would share another unit's SAFETY-02
   # latch; one with no `setpoint_entity` is a unit nothing can command; one with
   # no `max_w` would have to borrow a rating, which is the sibling-grading
-  # defect DEV-02 removed. The optional pair is the sensors, whose absence
-  # honestly means "this unit reports none".
+  # defect DEV-02 removed; `bms_soc_floor` is SAFETY-32's (options block above).
+  # The optional keys are the sensors, whose absence honestly means "this unit
+  # reports none", and `capacity_kwh`.
   #
   # ⚠️ THIS BLOCK IS DECLARED BEFORE `max_w:` ON PURPOSE. `max_w` now appears
   # TWICE under `schema:` - once nested here, once flat below - and a reader

goodwe_controller/test_control_task.py

@@ -0,0 +1,373 @@
+"""Runnable check for SAFETY-61: the control task's death.  `python3 test_control_task.py`
+
+If `run_control()` raised, nothing noticed: the task ended, `stop` was never
+set, the web UI kept serving and the Supervisor saw a healthy container. The
+card's fix and its reasons are in DECISIONS/SAFETY-61-control-task-death.md.
+
+Parts 1-3 drive the REAL `app.main.main()` - the process entry point - in a
+child process (this file re-invoked with `--driver`), with only the outside
+world faked: options, Home Assistant, the store, the web server and the core
+API probe. The child's exit status is the real one. Each child runs under a
+deadline, because the defect this card closes is a HANG (MUTATION.md habit 5).
+
+Takes ~15 s. Purge `__pycache__` before a mutation run (MUTATION.md).
+"""
+
+import asyncio
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+FAULT = "injected SAFETY-61 fault"
+DEADLINE_S = 40
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+# --------------------------------------------------------------------------- #
+# The child: the real main() with the outside world faked.
+# --------------------------------------------------------------------------- #
+BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+        "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+        "meter_source": "off", "max_w": 2000, "gain": 0.6, "slew_w": 0,
+        "deadband_w": 15, "target_grid_w": -10, "step_w": 10,
+        "heartbeat_s": 10, "stale_input_s": 15, "cycle_floor_s": 1,
+        "auto_start": True}
+
+
+def driver(scenario):
+    import aiohttp
+    import app.main as M
+    import app.store
+    import app.web
+
+    def event(*e):
+        """Streamed as it happens, so a child killed at the deadline or by
+        os._exit() still reports what it did."""
+        print("EV " + json.dumps(list(e)), flush=True)
+    t0 = time.monotonic()
+
+    class FakeHass:
+        def __init__(self):
+            self.batt = 0.0
+            self.limit_calls = 0
+
+        async def limits(self, entity):
+            self.limit_calls += 1
+            # "refuse": ESPHome not up at the startup gate, then a rating
+            # below max_w once run_control() re-reads it -> its refusal path.
+            if scenario == "refuse":
+                return (None, None, None) if self.limit_calls == 1 else (-1000.0, 1000.0, 1.0)
+            return (-5000.0, 5000.0, 1.0)
+
+        async def number(self, entity, invert=False):
+            if entity == "sensor.grid":
+                # a steady 600 W import, alternating by 1 W so every pass cycles
+                return 600.0 + (int(time.monotonic() - t0) % 2)
+            if entity == "sensor.batt":
+                return self.batt
+            if entity == "sensor.soc":
+                return 50.0
+            return None
+
+        async def state(self, entity):
+            return None
+
+        async def set_number(self, entity, value):
+            event("w", float(value))
+            self.batt = float(value)
+            return True
+
+        async def call(self, *a, **k):
+            return True
+
+        async def mqtt_service(self):
+            raise RuntimeError("no broker in this test")
+
+    class _Store:
+        data = {}
+
+        def get(self, key, default=None):
+            return default
+
+        def set(self, *a, **k):
+            pass
+
+    class _Runner:
+        async def cleanup(self):
+            event("cleanup")
+
+    async def _web_start(controller, port=8099, selfcfg=None):
+        return _Runner()
+
+    class _NoProbeSession:
+        """The core API probe in amain() must not touch a network."""
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+        def get(self, *a, **k):
+            raise OSError("no network in this test")
+
+    M.load_options = lambda: dict(BASE)
+    M.HomeAssistant = lambda session: FakeHass()
+    app.store.Store = _Store
+    app.web.start = _web_start
+    aiohttp.ClientSession = _NoProbeSession
+
+    real_cycle = M.Controller.cycle
+    n = {"cycles": 0}
+
+    async def cycle(self):
+        n["cycles"] += 1
+        if scenario == "fault" and n["cycles"] == 4:
+            event("fault")
+            raise RuntimeError(FAULT)
+        await real_cycle(self)
+        if scenario == "clean" and n["cycles"] == 4:
+            os.kill(os.getpid(), signal.SIGTERM)     # the Supervisor's stop
+
+    M.Controller.cycle = cycle
+    M.main()
+
+
+def run_child(scenario):
+    env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
+    try:
+        p = subprocess.run([sys.executable, "-B", __file__, "--driver", scenario],
+                           cwd=HERE, env=env, capture_output=True, text=True,
+                           timeout=DEADLINE_S)
+    except subprocess.TimeoutExpired as e:
+        rc, out, err = None, e.stdout or "", e.stderr or ""
+    else:
+        rc, out, err = p.returncode, p.stdout, p.stderr
+    out = out.decode() if isinstance(out, bytes) else out
+    err = err.decode() if isinstance(err, bytes) else err
+    events = [json.loads(ln[3:]) for ln in out.splitlines() if ln.startswith("EV ")]
+    return rc, out, err, events
+
+
+def tail(text, n=15):
+    return "\n".join("      | " + ln for ln in text.splitlines()[-n:])
+
+
+if len(sys.argv) == 3 and sys.argv[1] == "--driver":
+    driver(sys.argv[2])
+    sys.exit(0)            # unreachable in every scenario: main() exits itself
+
+
+# --------------------------------------------------------------------------- #
+print("1. an exception inside cycle() - the real main(), in a child process")
+rc, out, err, ev = run_child("fault")
+print(f"    exit status {rc}; events {ev}")
+check(f"the process ENDS inside {DEADLINE_S} s rather than sitting there "
+      "healthy (the defect is a hang)", rc is not None)
+check("exit status is 70 (EX_SOFTWARE), not 0 and not an unhandled crash's 1",
+      rc == 70)
+fault_at = ev.index(["fault"]) if ["fault"] in ev else None
+before = [e[1] for e in ev[:fault_at] if e[0] == "w"] if fault_at is not None else []
+after = [e[1] for e in ev[fault_at + 1:] if e[0] == "w"] if fault_at is not None else []
+# Precondition (habit 23): the inverter was holding a NON-ZERO command when the
+# fault hit, so a 0 W write afterwards is evidence of the wind-down, not rest.
+check("precondition: the fault hit while the last command was non-zero "
+      f"(last before fault = {before[-1] if before else None})",
+      fault_at is not None and bool(before) and before[-1] != 0.0)
+check(f"the wind-down ran: 0 W written AFTER the fault (writes after = {after})",
+      after == [0.0])
+check("the web runner was cleaned up, i.e. the ordinary shutdown path ran",
+      ["cleanup"] in ev[(fault_at or 0):])
+died_line = next((i for i, ln in enumerate(err.splitlines())
+                  if "CONTROL LOOP DIED (RuntimeError('" + FAULT + "'))" in ln), None)
+wind_line = next((i for i, ln in enumerate(err.splitlines())
+                  if "shutting down - commanding 0 W" in ln), None)
+check("a CRITICAL line names the exception and the exit status",
+      died_line is not None
+      and "CRITICAL" in err.splitlines()[died_line]
+      and "commanding 0 W and exiting with status 70" in err.splitlines()[died_line])
+check("the traceback is logged, down to the frame that raised",
+      "Traceback (most recent call last)" in err
+      and f"RuntimeError: {FAULT}" in err and "in cycle" in err)
+check("it is logged AT THE DEATH, before the wind-down - not only at exit",
+      died_line is not None and wind_line is not None and died_line < wind_line)
+check("exactly one traceback: main() does not print a second one",
+      err.count("Traceback (most recent call last)") == 1)
+if fails:
+    print(tail(err, 30))
+
+# --------------------------------------------------------------------------- #
+print("2. the same entry point stopped by SIGTERM - the callback must stay quiet")
+rc2, out2, err2, ev2 = run_child("clean")
+print(f"    exit status {rc2}; events {ev2}")
+check("an ordinary stop still exits 0", rc2 == 0)
+check("an ordinary stop logs no CONTROL LOOP DIED", "CONTROL LOOP DIED" not in err2)
+check("an ordinary stop still winds down to 0 W, from a non-zero command",
+      any(e[0] == "w" and e[1] != 0.0 for e in ev2)
+      and [e for e in ev2 if e[0] == "w"][-1:] == [["w", 0.0]])
+
+# --------------------------------------------------------------------------- #
+print("3. run_control()'s own refusal path returns on purpose - still exit 78")
+rc3, out3, err3, ev3 = run_child("refuse")
+print(f"    exit status {rc3}")
+check("a late-arriving device rating refusal exits 78 (EX_CONFIG), not 70",
+      rc3 == 78)
+check("and is not reported as a death", "CONTROL LOOP DIED" not in err3)
+if rc3 != 78:
+    print(tail(err3))
+
+# --------------------------------------------------------------------------- #
+print("4. hass.py: a body that is not a state object is an ABSENCE")
+import aiohttp                                      # noqa: E402
+from aiohttp import web as aioweb                   # noqa: E402
+
+import app.hass as H                                # noqa: E402
+
+
+async def hass_cases():
+    async def handler(req):
+        k = req.match_info["k"]
+        if k == "sensor.good":
+            return aioweb.json_response({"entity_id": k, "state": "42"})
+        if k == "sensor.badjson":
+            return aioweb.Response(text="{not json", content_type="application/json")
+        if k == "sensor.list":
+            return aioweb.Response(text="[1, 2]", content_type="application/json")
+        if k == "sensor.html":
+            return aioweb.Response(text="<html>", content_type="text/html")
+        return aioweb.Response(status=404)
+
+    async def svc(req):
+        return aioweb.Response(body=b"\xff\xfe\xfa", status=400,
+                               headers={"Content-Type": "text/plain; charset=utf-8"})
+
+    app_ = aioweb.Application()
+    app_.router.add_get("/states/{k}", handler)
+    app_.router.add_post("/services/{d}/{s}", svc)
+    runner = aioweb.AppRunner(app_)
+    await runner.setup()
+    site = aioweb.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]
+    H.CORE_API = f"http://127.0.0.1:{port}"
+    got = {}
+    try:
+        async with aiohttp.ClientSession() as session:
+            h = H.HomeAssistant(session, token="t")
+            for e in ("sensor.good", "sensor.badjson", "sensor.list", "sensor.html"):
+                try:
+                    got[e] = ("ok", await h.number(e))
+                except Exception as err:                    # noqa: BLE001
+                    got[e] = ("raised", type(err).__name__)
+            try:
+                got["call"] = ("ok", await h.set_number("number.sp", 0))
+            except Exception as err:                        # noqa: BLE001
+                got["call"] = ("raised", type(err).__name__)
+    finally:
+        await runner.cleanup()
+    return got
+
+got = asyncio.run(hass_cases())
+print(f"    {got}")
+check("baseline: a well-formed state reads as its value (42.0)",
+      got["sensor.good"] == ("ok", 42.0))
+check("a malformed JSON body labelled application/json reads as None, not a raise",
+      got["sensor.badjson"] == ("ok", None))
+check("a JSON body that is not an object reads as None, not a raise",
+      got["sensor.list"] == ("ok", None))
+check("a text/html 200 reads as None (already true before SAFETY-61)",
+      got["sensor.html"] == ("ok", None))
+check("a rejected write with an undecodable body is a failed write (False)",
+      got["call"] == ("ok", False))
+
+# --------------------------------------------------------------------------- #
+print("5. the control-loop status row")
+from app.main import Controller                     # noqa: E402
+
+
+class _Pub:
+    enabled = False
+
+    def publish(self, *a, **k):
+        pass
+
+    def close(self):
+        pass
+
+
+class _Store:
+    data = {}
+
+    def set(self, *a, **k):
+        pass
+
+
+class _RowHass:
+    async def limits(self, entity):
+        return (-5000.0, 5000.0, 1.0)
+
+    async def number(self, entity, invert=False):
+        return {"sensor.grid": 600.0, "sensor.soc": 50.0}.get(entity, 0.0)
+
+    async def state(self, entity):
+        return None
+
+    async def set_number(self, entity, value):
+        return True
+
+    async def call(self, *a, **k):
+        return True
+
+
+def rows_for(pass_age):
+    c = Controller(dict(BASE), _RowHass(), _Store(), _Pub(), stop=asyncio.Event())
+    c.loop_pass_mono = None if pass_age is None else time.monotonic() - pass_age
+    return c.checks()
+
+
+r_new, r_29, r_31 = rows_for(None), rows_for(29), rows_for(31)
+print(f"    {r_new[0]} | {r_29[0]} | {r_31[0]}")
+check("before the first pass the row is ok and says 'starting'",
+      r_new[0] == {"ok": True, "warn": False, "text": "control loop: starting"})
+check("29 s after a pass it is ok",
+      r_29[0] == {"ok": True, "warn": False,
+                  "text": "control loop: last pass 29 s ago"})
+check("31 s after a pass it is RED and first, naming the age and the veto",
+      r_31[0] == {"ok": False, "warn": False,
+                  "text": "control loop: NO PASS FOR 31 s (layer 1 zeroes the "
+                          "inverter after 30 s without a heartbeat)"})
+
+
+async def one_pass():
+    c = Controller(dict(BASE), _RowHass(), _Store(), _Pub(), stop=asyncio.Event())
+    c.auto = True
+    before = getattr(c, "loop_pass_mono", "absent")
+    t = asyncio.create_task(c.run_control())
+    await asyncio.sleep(2.5)
+    c.stopping = True
+    await asyncio.wait_for(t, timeout=10)
+    return before, getattr(c, "loop_pass_mono", None)
+
+_before, _after = asyncio.run(one_pass())
+print(f"    stamp before run: {_before}; age after 2.5 s: "
+      f"{None if _after is None else round(time.monotonic() - _after, 2)} s")
+check("the real run_control() stamps each pass",
+      _before is None and _after is not None and time.monotonic() - _after < 3)
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all SAFETY-61 checks passed")

goodwe_controller/test_docs.py

@@ -0,0 +1,210 @@
+"""Runnable check that the operator docs say what the code does.  `python3 test_docs.py`
+
+DOC-1. Two things the docs got wrong and an installer would have copied:
+
+1. The `inverters:` example in DOCS.md must be ACCEPTED by the shipped
+   validator (`app/validate.py`) and carry every key config.yaml's schema
+   requires per entry. It omitted `bms_soc_floor` and would not start.
+2. The P1 age entity the ESP32 watchdog subscribes to must be documented under
+   the id Home Assistant composes from what `app/mqtt.py` announces, and the
+   old wrong id must appear in no shipped doc.
+
+Each part also runs its own mutation (habit 1 of MUTATION.md): the spoilt input
+must turn it red, or the green above it means nothing.
+"""
+
+import os
+import re
+import sys
+
+from app import mqtt
+from app.validate import validate_startup_config
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.dirname(HERE)
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def read(*parts):
+    with open(os.path.join(*parts), encoding="utf-8") as fh:
+        return fh.read()
+
+
+DOCS = read(HERE, "DOCS.md")
+CFG = read(HERE, "config.yaml")
+
+
+# --------------------------------------------------------------------------- #
+# A reader for the two YAML shapes used here. No PyYAML in the checks image
+# (test_defaults_pin.py says why), so this is by hand, and part 1a checks it
+# actually found something.
+def scalar(raw):
+    raw = raw.split("#", 1)[0].strip()
+    if raw[:1] in "\"'" and raw[-1:] == raw[:1] and len(raw) >= 2:
+        return raw[1:-1]
+    if raw in ("true", "false"):
+        return raw == "true"
+    if raw == "[]":
+        return []
+    for cast in (int, float):
+        try:
+            return cast(raw)
+        except ValueError:
+            pass
+    return raw
+
+
+def parse_example(text):
+    """Top-level `key: value` lines plus one list of flat mappings."""
+    out, cur_list, cur = {}, None, None
+    for line in text.splitlines():
+        if not line.split("#", 1)[0].strip():
+            continue
+        m = re.match(r"^([a-z0-9_]+):\s*(.*)$", line)
+        if m:
+            if m.group(2).split("#", 1)[0].strip():
+                out[m.group(1)] = scalar(m.group(2))
+                cur_list = None
+            else:
+                cur_list = out[m.group(1)] = []
+            continue
+        m = re.match(r"^\s+-\s+([a-z0-9_]+):\s*(.*)$", line)
+        if m and cur_list is not None:
+            cur = {m.group(1): scalar(m.group(2))}
+            cur_list.append(cur)
+            continue
+        m = re.match(r"^\s+([a-z0-9_]+):\s*(.*)$", line)
+        if m and cur is not None:
+            cur[m.group(1)] = scalar(m.group(2))
+            continue
+        raise ValueError(f"unparsed line in the DOCS.md example: {line!r}")
+    return out
+
+
+# --------------------------------------------------------------------------- #
+print("DOC-1 - the docs say what the code does\n")
+print("1. the `inverters:` example in DOCS.md starts")
+
+section = DOCS[DOCS.index("### More than one inverter (`inverters`)"):]
+section = section[:section.index("\n### ", 1)]
+block = re.search(r"```yaml\n(.*?)```", section, re.S).group(1)
+example = parse_example(block)
+entries = example.get("inverters") or []
+
+# The options a fresh install is given, from config.yaml itself: the example is
+# pasted on top of these, exactly as an installer does.
+opts_block = CFG[CFG.index("\noptions:"):CFG.index("\nschema:")]
+shipped = {m.group(1): scalar(m.group(2))
+           for m in re.finditer(r"^  ([a-z0-9_]+): *(.*)$", opts_block, re.M)}
+
+# config.yaml's per-entry schema: required keys are the ones without `?`.
+schema_block = CFG[CFG.index("\nschema:"):]
+inv_schema = re.search(r"^  inverters:\n((?:    .*\n)+)", schema_block, re.M).group(1)
+schema_keys = dict(re.findall(r"^    -? ?\s*([a-z0-9_]+): *(\S+)", inv_schema, re.M))
+required = sorted(k for k, t in schema_keys.items() if not t.endswith("?"))
+
+check(f"1a. the reader found the example: {len(entries)} entries, "
+      f"{len(shipped)} shipped options, schema keys {sorted(schema_keys)}",
+      len(entries) == 2 and all(len(e) >= 4 for e in entries)
+      and len(shipped) > 50 and "bms_soc_floor" in required)
+
+missing = [(e.get("name"), k) for e in entries for k in required if k not in e]
+check(f"1b. every entry carries every key the schema requires {required} "
+      f"(missing: {missing})", not missing)
+
+unknown = [(e.get("name"), k) for e in entries for k in e if k not in schema_keys]
+check(f"1c. no entry uses a key the schema does not declare (found: {unknown})",
+      not unknown)
+
+
+def with_example(ex):
+    opts = dict(shipped)
+    opts.update(ex)
+    return opts
+
+
+problems = validate_startup_config(with_example(example))
+for p in problems:
+    print(f"      refused: {p[:160]}")
+check("1d. the shipped validator accepts the example as documented", problems == [])
+
+# Precondition: the example clears the flat setpoint_entity, so without its
+# roster there is no actuator. If this passed, 1d graded the flat keys, not the
+# roster.
+flat_only = {k: v for k, v in example.items() if k != "inverters"}
+check("1e. ...and it is the ROSTER that 1d graded (the example without it is "
+      "refused)", validate_startup_config(with_example(flat_only)) != [])
+
+# The mutation this card is about: the key the docs used to leave out.
+spoilt = dict(example, inverters=[
+    {k: v for k, v in e.items() if k != "bms_soc_floor"} for e in entries])
+sp = validate_startup_config(with_example(spoilt))
+check("1f. MUTATION: the example with `bms_soc_floor` removed is refused, "
+      "and the refusal names it",
+      sp != [] and any("bms_soc_floor" in p for p in sp))
+
+# --------------------------------------------------------------------------- #
+print("\n2. the P1 age entity is documented under its real id")
+
+
+def compose(device_name, entity_name):
+    """HA's entity_id from device name + entity name (mqtt.py's note on
+    DEVICE). Written here, not borrowed from mqtt._slug, so a change to that
+    helper cannot move both sides of the comparison."""
+    s = re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower())
+    return "sensor." + s.strip("_")
+
+
+p1_rows = [r for r in mqtt.SENSORS if r[0] == "p1_age"]
+check("2a. mqtt.py announces exactly one `p1_age` sensor", len(p1_rows) == 1)
+real_id = compose(mqtt.DEVICE["name"], p1_rows[0][2]) if p1_rows else ""
+print(f"      composed from mqtt.py: {real_id}")
+
+# Anchor outside mqtt.py: the id recorded from the live install.
+recorded = read(HERE, "tools", "rehearsal", "announced-reference.txt")
+check("2b. ...and that composes to the id recorded from the live install "
+      "(tools/rehearsal/announced-reference.txt)",
+      re.search(rf"^{re.escape(real_id)}\s+goodwe_ctl_p1_age\b", recorded, re.M)
+      is not None)
+
+m = re.search(r"^#### `(sensor\.[a-z0-9_]+)`\n\n(?:.*\n){0,6}?.*stale-input watchdog",
+              DOCS, re.M)
+documented = m.group(1) if m else None
+check(f"2c. DOCS.md's watchdog-entity section is headed with that id "
+      f"(found {documented!r})", documented == real_id)
+
+# The wrong id, anywhere an installer reads. CHANGELOG.md and DECISIONS/ are
+# history and keep what was said at the time.
+SHIPPED_DOCS = [
+    os.path.join(HERE, "DOCS.md"), os.path.join(HERE, "config.yaml"),
+    os.path.join(REPO, "README.md"), os.path.join(REPO, "FIELD-GUIDE.md"),
+    os.path.join(REPO, "docs", "build_guide.py"),
+    os.path.join(REPO, "docs", "field-guide.html"),
+]
+WRONG = "p1_sample_age_s"
+hits = []
+for path in SHIPPED_DOCS:
+    for n, line in enumerate(read(path).splitlines(), 1):
+        if WRONG in line:
+            hits.append(f"{os.path.relpath(path, REPO)}:{n}")
+check(f"2d. `{WRONG}` appears in no shipped doc (hits: {hits})", not hits)
+
+# MUTATION: the old heading must fail 2c's comparison.
+bad = DOCS.replace(f"#### `{real_id}`", f"#### `sensor.{WRONG}`")
+mb = re.search(r"^#### `(sensor\.[a-z0-9_]+)`\n\n(?:.*\n){0,6}?.*stale-input watchdog",
+               bad, re.M)
+check("2e. MUTATION: the old heading is found and does not match",
+      mb is not None and mb.group(1) != real_id)
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    sys.exit(1)
+print("all ok")

goodwe_controller/test_rehearse.py

@@ -75,9 +75,10 @@ print("1. the harness's own preconditions")
 sys.path.insert(0, str(HERE / "tools"))
 import rehearse_roster as RR  # noqa: E402
 
-check("the tool refuses rather than passing when it cannot run: its three exit "
-      "codes are distinct and 2 is not 0",
-      (RR.EXIT_OK, RR.EXIT_FINDING, RR.EXIT_COULD_NOT_RUN) == (0, 1, 2))
+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
@@ -204,6 +205,10 @@ def mutant_tree(edits) -> str:
     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")
@@ -356,8 +361,9 @@ check("omitting --live-entities REFUSES with exit 2 rather than reporting a "
       _no_code == RR.EXIT_COULD_NOT_RUN and "REFUSED" in _no_out + _no_err)
 
 _deg_code, _deg_out, _ = raw("--no-live-entities")
-check("...and the explicit opt-out RUNS, so the escape hatch exists",
-      _deg_code in (RR.EXIT_OK, RR.EXIT_FINDING))
+check("...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

goodwe_controller/test_tool_inputs.py

@@ -0,0 +1,262 @@
+"""SAFETY-60/62: the operator tools refuse input they cannot read.  `python3 test_tool_inputs.py`
+
+Each tool is driven as the operator runs it, first on good input (the accepted
+baseline, MUTATION.md habit 7) and then on the shapes that used to pass.
+"""
+import contextlib
+import copy
+import importlib.util
+import io
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+TOOLS = HERE / "tools"
+FIX = TOOLS / "rehearsal"
+CAPTURE = HERE / "scenarios" / "safety57-2026-09-21" / \
+    "bus-a-2026-09-21-1030-1110.log.gz"
+TMP = tempfile.mkdtemp(prefix="test_tool_inputs.")
+ENV = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}
+
+fails = []
+
+
+def check(name, ok):
+    print(f"  {'ok  ' if ok else 'FAIL'}  {name}")
+    if not ok:
+        fails.append(name)
+
+
+def dump(name, obj=None, text=None):
+    p = Path(TMP) / name
+    p.write_text(text if text is not None else json.dumps(obj, indent=1),
+                 encoding="utf-8")
+    return str(p)
+
+
+def tool(script, *args):
+    r = subprocess.run([sys.executable, str(TOOLS / script), *args],
+                       capture_output=True, text=True, env=ENV)
+    return r.returncode, r.stdout + r.stderr
+
+
+def ha_apps_info(options):
+    """What `ha apps info <slug> --raw-json` prints: the options sit under data.options."""
+    text = {"en": ("Setpoint entity", "The ESPHome number carrying the setpoint."),
+            "nl": ("Setpoint-entiteit", "Het ESPHome-getal met het setpoint.")}
+    return {"result": "ok", "data": {
+        "name": "GoodWe RS485 Controller", "slug": "local_goodwe_controller",
+        "version": "0.12.0", "version_latest": "0.12.0", "state": "started",
+        "boot": "auto", "ingress": True, "hassio_api": True,
+        "options": copy.deepcopy(options),
+        "schema": [{"name": k, "required": True, "optional": False,
+                    "type": "string"} for k in options],
+        "translations": {lang: {"configuration": {
+            "setpoint_entity": {"name": name, "description": desc},
+            "inverters": {"name": "Inverters", "description": "Roster.",
+                          "fields": {"setpoint_entity": {
+                              "name": name, "description": f"{desc} (per unit)"}}}}}
+            for lang, (name, desc) in text.items()}}}
+
+
+# --------------------------------------------------------------------------- #
+print("1. predeploy_check.py --options (SAFETY-60)")
+# --------------------------------------------------------------------------- #
+# A deployed version that differs from the shipped one, so only the options half decides.
+BARE = {"setpoint_entity": "number.goodwe_setpoint_w", "inverters": [],
+        "max_w": 5000, "slew_w": 0, "gain": 0.6, "target_grid_w": -10,
+        "deadband_w": 15, "meter_source": "off", "log_level": "info"}
+NOREPO = tempfile.mkdtemp(prefix="norepo.", dir=TMP)
+
+
+def gate(opts_path):
+    return tool("predeploy_check.py", "--deployed-version", "0.0.0-test",
+                "--repo", NOREPO, "--options", opts_path)
+
+
+code, out = gate(dump("bare.json", BARE))
+print(f"      bare options -> exit {code}")
+check("baseline: the bare options object is CLEARED (exit 0)",
+      code == 0 and "CLEARED for deploy" in out)
+code, out = gate(dump("bare-bad.json", dict(BARE, slew_w=5000)))
+check("baseline: the bare object with slew_w 5000 is REFUSED by the validator",
+      code == 1 and "slew_w is 5000 W" in out)
+
+for label, doc in (("wrapped", ha_apps_info(BARE)),
+                   ("wrapped, slew_w 5000", ha_apps_info(dict(BARE, slew_w=5000)))):
+    code, out = gate(dump("wrapped.json", doc))
+    print(f"      {label} -> exit {code}")
+    check(f"the `ha apps info` capture ({label}) is REFUSED, exit 1",
+          code == 1 and "CLEARED" not in out)
+    check("...naming the shape and the fix, not a validator rule",
+          "0 of 2 top-level key(s) recognised" in out and ".data.options" in out
+          and "VALIDATOR" not in out)
+
+from app.validate import actuator_writers, validate_startup_config  # noqa: E402
+_w = ha_apps_info(dict(BARE, slew_w=5000))
+print(f"      (for reference: the validator alone, handed that capture, finds "
+      f"{len(validate_startup_config(_w))} problem(s) and "
+      f"{len(actuator_writers(_w))} 'actuators' - the gate has to refuse it)")
+
+code, out = gate(dump("list.json", [BARE]))
+check("a list-wrapped options object is REFUSED, naming the type",
+      code == 1 and "is a JSON list, not an object" in out)
+code, out = gate(dump("mostly-foreign.json",
+                      {"setpoint_entity": "number.x", "a": 1, "b": 2}))
+check("an object with fewer than half its keys recognised is REFUSED",
+      code == 1 and "1 of 3 top-level key(s) recognised" in out)
+
+# --------------------------------------------------------------------------- #
+print("\n2. rehearse_roster.py --state/--before/--after (SAFETY-62)")
+# --------------------------------------------------------------------------- #
+sys.path.insert(0, str(TOOLS))
+import rehearse_roster as RR  # noqa: E402
+
+STATE = FIX / "state-legacy.json"
+LIVE = ["--live-entities", str(FIX / "live-entities-recorded.txt"),
+        "--reference", str(FIX / "announced-reference.txt")]
+
+
+def rehearse(state=str(STATE), before=str(FIX / "flat.yaml"),
+             after=str(FIX / "roster.yaml"), extra=LIVE):
+    return tool("rehearse_roster.py", "--state", state, "--before", before,
+                "--after", after, *extra)
+
+
+code, out = rehearse()
+print(f"      control -> exit {code}")
+check("baseline: the control run reports its 2 findings, exit 1",
+      code == RR.EXIT_FINDING and "2 FINDING(S)" in out)
+code, out = rehearse(state=str(FIX / "state-fresh.json"))
+check("baseline: a fresh (comment-only) state file still runs clean, exit 0",
+      code == RR.EXIT_OK and "NO FINDINGS" in out)
+
+raw = STATE.read_text(encoding="utf-8")
+state = json.loads(raw)
+cases = {
+    "wrapped": (dump("s-wrapped.json", {"result": "ok", "data": state}),
+                "0 of 2 top-level key(s) recognised"),
+    "truncated by 20 bytes": (dump("s-trunc.json", text=raw[:-20]),
+                              "not a readable JSON state file"),
+    "list-wrapped": (dump("s-list.json", [state]),
+                     "is a JSON list, not an object"),
+}
+for label, (path, words) in cases.items():
+    code, out = rehearse(state=path)
+    print(f"      state {label} -> exit {code}")
+    check(f"state {label}: REFUSED with exit 2 and the reason",
+          code == RR.EXIT_COULD_NOT_RUN and words in out
+          and "NO FINDINGS" not in out)
+
+try:
+    _auto = RR.load_state(dump("s-auto.json", {"auto": True}))
+except RR.Refused as err:
+    _auto = str(err)
+check("a state holding only the Automatic toggle (`auto`, app/main.py) is "
+      "recognised as a store", _auto == {"auto": True})
+
+flat = RR.load_options(str(FIX / "flat.yaml"))
+roster = RR.load_options(str(FIX / "roster.yaml"))
+for side, kw in (("--before", {"before": dump("b.json", ha_apps_info(flat))}),
+                 ("--after", {"after": dump("a.json", ha_apps_info(roster))})):
+    code, out = rehearse(**kw)
+    print(f"      {side} wrapped -> exit {code}")
+    check(f"{side} as an `ha apps info` capture: REFUSED with exit 2",
+          code == RR.EXIT_COULD_NOT_RUN and ".data.options" in out
+          and "NO FINDINGS" not in out)
+
+code, out = rehearse(after=str(FIX / "roster-inverter-a.yaml"),
+                     extra=["--no-live-entities", "--reference",
+                            str(FIX / "announced-reference.txt")])
+check("a clean --no-live-entities run is exit 3, not 0",
+      code == RR.EXIT_DEGRADED and "NO FINDINGS" in out)
+
+# --------------------------------------------------------------------------- #
+print("\n3. alloc_sim.py - the split-invariance verdict reaches the exit code")
+# --------------------------------------------------------------------------- #
+spec = importlib.util.spec_from_file_location("_alloc_sim", TOOLS / "alloc_sim.py")
+AS = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(AS)
+
+
+class _Run:
+    """One rig run, reduced to what `splits()` reads."""
+
+    def __init__(self, grid):
+        self.samples = [{"shares": [1000.0, 1000.0]}]
+        self._grid = grid
+
+    def grids(self):
+        return self._grid
+
+    def targets(self):
+        return [2000.0] * len(self._grid)
+
+
+class _Sitesim:
+    def load_addon(self, *_):
+        return None
+
+
+def alloc_main(same):
+    """`main(--splits)` with the rig stubbed; the real `splits()` grades the traces."""
+    seen = []
+
+    def fake(_sitesim, _control, mode, *a, **kw):
+        seen.append(mode)
+        return _Run([0.0, 1.0] if same or len(seen) == 1 else [0.0, 2.0])
+
+    AS.find_sim = lambda: Path("/nonexistent-rig")
+    AS.load = lambda _sim: (None, _Sitesim(), None, None)
+    AS.run_site_feedback = fake
+    os.environ["GOODWE_ADDON_REPO"] = str(HERE.parent)
+    sys.argv = ["alloc_sim.py", "--splits"]
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        rc = AS.main()
+    return rc, buf.getvalue()
+
+
+rc, out = alloc_main(same=True)
+check("baseline: identical traces HOLD, exit 0",
+      rc == 0 and "split-invariance HOLDS" in out)
+rc, out = alloc_main(same=False)
+print(f"      differing traces -> exit {rc}")
+check("differing traces FAIL, exit 1",
+      rc == 1 and "split-invariance FAILS" in out)
+
+# --------------------------------------------------------------------------- #
+print("\n4. decode_bus_capture.py - a file that is not a capture")
+# --------------------------------------------------------------------------- #
+code, out = tool("decode_bus_capture.py", str(CAPTURE))
+check("baseline: the committed capture decodes, exit 0",
+      code == 0 and "writes         1200" in out and "REFUSED" not in out)
+junk = dump("junk.log", text="hello\nthis is not a capture\n")
+code, out = tool("decode_bus_capture.py", junk)
+print(f"      junk -> exit {code}")
+check("a junk file is REFUSED, exit 1",
+      code == 1 and "REFUSED" in out and "0 of 2 line(s) decoded" in out)
+code, out = tool("decode_bus_capture.py", dump("empty.log", text=""))
+check("an empty file is REFUSED, exit 1",
+      code == 1 and "0 of 0 line(s) decoded" in out)
+code, out = tool("decode_bus_capture.py", str(CAPTURE), junk)
+check("several paths, as the docstring's glob passes them: each is decoded and "
+      "one bad file makes the run exit 1",
+      code == 1 and f"== {CAPTURE}" in out and f"== {junk}" in out)
+code, out = tool("decode_bus_capture.py", str(CAPTURE), str(CAPTURE))
+check("...and two good files exit 0", code == 0 and out.count("REFUSED") == 0)
+
+shutil.rmtree(TMP, ignore_errors=True)
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all tool-input checks passed")

goodwe_controller/tools/alloc_sim.py

@@ -6,6 +6,8 @@
     python3 tools/alloc_sim.py --sweep       # AC 3: gain to 1.0 at every split
     python3 tools/alloc_sim.py --all
 
+    Exit 0 only if every split-invariance run HOLDS; 1 if one FAILS; 2 if no rig.
+
     # DEV-02: the split-invariance finding ALLOC-01's review handed on, and the
     # sensing change that closed it. The two rows are the evidence.
     python3 tools/alloc_sim.py --splits --gain 0.4 --feedback unit   # 250 W apart
@@ -279,13 +281,14 @@ def main():
     if a.table or a.all:
         table(sitesim, control, watchdog, a.cycles, a.load)
         print()
+    held = True
     if a.splits or a.all:
-        splits(sitesim, control, watchdog, a.cycles, a.load, a.gain,
-               a.feedback)
+        held = splits(sitesim, control, watchdog, a.cycles, a.load, a.gain,
+                      a.feedback)
         print()
     if a.sweep or a.all:
         sweep(sitesim, control, watchdog, a.cycles, a.load)
-    return 0
+    return 0 if held else 1
 
 
 if __name__ == "__main__":

goodwe_controller/tools/decode_bus_capture.py

@@ -195,14 +195,37 @@ def _hms(t):
     return f"{int(t // 3600):02d}:{int(t % 3600 // 60):02d}:{t % 60:06.3f}"
 
 
+def unreadable(stats):
+    """Why a capture decoded to too little to trust, or None: most lines must be frames."""
+    framed = sum(stats[k] for k in ("requests", "responses", "writes",
+                                     "write_echoes"))
+    if framed > 0 and stats["lines"] - framed <= framed:
+        return None
+    return (f"{framed} of {stats['lines']} line(s) decoded as frames "
+            f"(unparsed {stats['unparsed']}, bad_crc {stats['bad_crc']}, "
+            f"unknown {stats['unknown']}) - not a capture this decoder can read")
+
+
 def main(argv):
-    if len(argv) != 2:
+    """Decode each capture named; exit 1 if any is unreadable, 2 on no argument."""
+    if len(argv) < 2:
         print(__doc__.splitlines()[2].strip())
         return 2
-    events, stats = decode(argv[1])
-    print(f"== {argv[1]}")
+    rc = 0
+    for path in argv[1:]:
+        rc |= _report(path)
+    return rc
+
+
+def _report(path):
+    events, stats = decode(path)
+    print(f"== {path}")
     for k, v in stats.items():
         print(f"   {k:14} {v}")
+    bad = unreadable(stats)
+    if bad:
+        print(f"REFUSED: {path}: {bad}")
+        return 1
     for kind in ("setpoint_w", "batt_w", "soc_pct"):
         rows = series(events, kind)
         if not rows:

goodwe_controller/tools/predeploy_check.py

@@ -39,8 +39,10 @@ and not at merge. The check belongs where the mistake is made.
 
 ⚠️ WHAT THIS DOES NOT DO. It does not read the live box. The options blob is
 reachable only through the Supervisor API, which needs a supervisor token, so
-the operator exports it and passes `--options`. Without that flag the validator
-half is SKIPPED and says so - a skipped check must never look like a passed one.
+the operator exports it and passes `--options` - the bare options object
+(`.data.options` of `ha apps info <slug> --raw-json`); any other shape is
+refused. Without that flag the validator half is SKIPPED and says so - a
+skipped check must never look like a passed one.
 """
 import argparse
 import json
@@ -79,6 +81,43 @@ def shipped_version(tree=ADDON):
     return m.group(1) if m else None
 
 
+def option_keys(addon=ADDON):
+    """Top-level option names in `addon`'s config.yaml `options:`/`schema:` blocks."""
+    keys, section = set(), None
+    with open(os.path.join(addon, "config.yaml"), encoding="utf-8") as fh:
+        for line in fh:
+            if not line.strip() or line.startswith("#"):
+                continue
+            if not line[0].isspace():
+                section = line.split(":", 1)[0].strip()
+                continue
+            m = re.match(r"  ([A-Za-z_][A-Za-z0-9_]*):", line)
+            if m and section in ("options", "schema"):
+                keys.add(m.group(1))
+    return keys
+
+
+def shape_problem(doc, known, what, allow_empty=False):
+    """Why `doc` is not a document of `known` keys, or None (SAFETY-60/62).
+
+    Positive property: a JSON object, at least half of whose top-level keys
+    (`_`-prefixed comments aside) `known(key)` recognises.
+    """
+    if not isinstance(doc, dict):
+        return f"{what} is a JSON {type(doc).__name__}, not an object"
+    keys = [k for k in doc if not str(k).startswith("_")]
+    hits = [k for k in keys if known(k)]
+    if (keys or allow_empty) and len(hits) * 2 >= len(keys):
+        return None
+    msg = (f"{what} is not the expected document: {len(hits)} of {len(keys)} "
+           f"top-level key(s) recognised {sorted(map(str, keys))[:8]}")
+    inner = doc.get("data")
+    if isinstance(inner, dict) and isinstance(inner.get("options"), dict):
+        msg += ("; this is `ha apps info --raw-json` output - pass its "
+                "`.data.options` instead")
+    return msg
+
+
 def is_runtime(path):
     return any(path == p or path.startswith(p) for p in RUNTIME)
 
@@ -256,6 +295,12 @@ def main():
         except (json.JSONDecodeError, OSError, UnicodeDecodeError) as exc:
             problems.append(f"--options file could not be read as JSON "
                             f"({type(exc).__name__}: {exc}); the config was NOT checked")
+        # SAFETY-60: the validator accepts any dict, so the shape is checked here.
+        bad = (shape_problem(opts, option_keys().__contains__, "--options")
+               if opts is not None else None)
+        if bad:
+            problems.append(f"{bad}; the config was NOT checked")
+            opts = None
     if a.options and opts is not None:
         try:
             from app.validate import validate_startup_config
@@ -281,7 +326,7 @@ def main():
                     print(f"      - {p}")
             else:
                 print("  live options pass this version's validator")
-    else:
+    elif not a.options:
         skipped.append("--options was not given, so the live config was NEVER CHECKED "
                        "against this version's validator. A config refused at startup is "
                        "exit 78. Export the options and re-run before deploying.")

goodwe_controller/tools/rehearse_roster.py

@@ -46,8 +46,11 @@ EXIT CODES
     1   ⭐ AT LEAST ONE OF THOSE HAPPENED. This is the code that must be
         reachable: `test_rehearse.py` mutates the shipped migration nine
         different ways and requires this exit code for each.
-    2   the rehearsal could not be performed (bad file, unparseable config, the
-        entity-id model failed its own anchor). ⚠️ NOT A PASS.
+    2   the rehearsal could not be performed (bad file, unparseable config, a
+        state or options file of the wrong shape, the entity-id model failed
+        its own anchor). ⚠️ NOT A PASS.
+    3   no findings, but --no-live-entities: the orphan check was not graded
+        against the install. ⚠️ NOT A PASS.
 """
 
 import argparse
@@ -64,10 +67,12 @@ from app.inverter import UNIT_A, UNIT_STORE_KEYS, Inverter, roster_names  # noqa
 from app.main import Controller  # noqa: E402
 from app import mqtt  # noqa: E402
 from app.p1 import is_enabled  # noqa: E402
-from app.store import (DEFAULTS, MAINT_UNIT_KEYS, Store,  # noqa: E402
-                       unit_key)
+from app.store import (DEFAULTS, MAINT_SCHEMA_KEY, MAINT_UNIT_KEYS,  # noqa: E402
+                       SCHEMA_KEY, Store, unit_key)
+from predeploy_check import option_keys, shape_problem  # noqa: E402
 
-EXIT_OK, EXIT_FINDING, EXIT_COULD_NOT_RUN = 0, 1, 2
+EXIT_OK, EXIT_FINDING, EXIT_COULD_NOT_RUN, EXIT_DEGRADED = 0, 1, 2, 3
+ADDON = Path(__file__).resolve().parent.parent
 
 
 class Refused(Exception):
@@ -184,8 +189,30 @@ def load_options(path: str) -> dict:
             obj = _mini_yaml(text)
         else:
             obj = yaml.safe_load(text)
-    if not isinstance(obj, dict):
-        raise Refused(f"{path}: options must be a mapping, got {type(obj).__name__}")
+    # SAFETY-62: a wrapped export is a dict too, so assert the keys are options.
+    bad = shape_problem(obj, option_keys(str(ADDON)).__contains__, path)
+    if bad:
+        raise Refused(bad)
+    return obj
+
+
+_UNIT_PREFIX = unit_key("\0", "").split("\0")[0]
+# "auto" is the Automatic toggle, written by app/main.py outside DEFAULTS.
+_STORE_KEYS = {*DEFAULTS, *MAINT_UNIT_KEYS, SCHEMA_KEY, MAINT_SCHEMA_KEY, "auto"}
+
+
+def load_state(path: str) -> dict:
+    """Refuse a state file `Store.load()` would silently start fresh on (SAFETY-62)."""
+    try:
+        obj = json.loads(Path(path).read_text(encoding="utf-8"))
+    except ValueError as err:
+        raise Refused(f"{path}: not a readable JSON state file ({err}); "
+                      "Store.load() would start fresh and grade an empty store")
+    bad = shape_problem(
+        obj, lambda k: k in _STORE_KEYS or str(k).startswith(_UNIT_PREFIX),
+        path, allow_empty=True)
+    if bad:
+        raise Refused(bad)
     return obj
 
 
@@ -1061,6 +1088,7 @@ def main(argv=None) -> int:
                      if args.live_entities else None)
         if not Path(args.state).is_file():
             raise Refused(f"{args.state}: not a file")
+        load_state(args.state)
     except (OSError, Refused) as err:
         print(f"REFUSED: {err}", file=sys.stderr)
         return EXIT_COULD_NOT_RUN
@@ -1179,7 +1207,9 @@ def main(argv=None) -> int:
         print("     the install's own entity list. This result does not cover "
               "orphaned entities.")
     print("=" * 78)
-    return EXIT_FINDING if findings else EXIT_OK
+    if findings:
+        return EXIT_FINDING
+    return EXIT_DEGRADED if degraded else EXIT_OK
 
 
 if __name__ == "__main__":