SAFETY-62 review: DOCS exit 3 at the command block (R1); recognise the auto store key (R2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU

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

@@ -1,106 +0,0 @@
-# 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

@@ -2820,8 +2820,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 +2843,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 +2858,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 +3007,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,17 +61,10 @@ class HomeAssistant:
                 if resp.status == 404:
                     return None
                 resp.raise_for_status()
-                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:
+                return await resp.json()
+        except (aiohttp.ClientError, TimeoutError) 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."""
@@ -124,8 +117,7 @@ class HomeAssistant:
                                domain, service, resp.status, body[:200])
                     return False
                 return True
-        # ValueError: an undecodable error body (SAFETY-61) is a failed write.
-        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
+        except (aiohttp.ClientError, TimeoutError) as err:
             self._moan(f"svcerr:{domain}.{service}", "service %s.%s failed: %s",
                        domain, service, err)
             return False

goodwe_controller/app/main.py

@@ -25,11 +25,9 @@ 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
@@ -76,8 +74,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 (FIRMWARE_INGEST_VETO_S, ConfigRefused, refuse,
-                       validate_startup_config, validate_unit_ranges)
+from .validate import (ConfigRefused, refuse, validate_startup_config,
+                       validate_unit_ranges)
 from .watchdog import (
     AGAINST_METER, AGAINST_OWN_POWER, SIGN_OPTION_DEFAULTS,
     SignWatchdog, SignWatchdogConfig, loop_is_following,
@@ -987,8 +985,6 @@ 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:
@@ -4401,7 +4397,6 @@ 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:
@@ -4601,22 +4596,6 @@ 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:
@@ -5470,30 +5449,6 @@ 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(
@@ -5591,13 +5546,7 @@ async def amain() -> None:
             with contextlib.suppress(NotImplementedError):
                 loop.add_signal_handler(sig, stop.set)
 
-        # 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]
+        tasks = [asyncio.create_task(controller.run_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
@@ -5632,14 +5581,10 @@ 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:
@@ -5658,9 +5603,6 @@ 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/test_control_task.py

@@ -1,373 +0,0 @@
-"""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_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__":