SAFETY-60/62: operator tools refuse input they cannot read

predeploy_check.py and rehearse_roster.py now assert a positive property of
their input (a JSON object, most of whose top-level keys are known option or
store keys) and refuse anything else with the shape named. The `ha apps info
--raw-json` wrapper, a truncated or list-wrapped state file and a wrapped
config no longer produce "CLEARED" / "NO FINDINGS" with exit 0.

Also: a clean --no-live-entities rehearsal exits 3, not 0; alloc_sim.py
exits 1 when split-invariance FAILS; decode_bus_capture.py takes several
paths and exits 1 on a file that is mostly not frames.

Graded by the new test_tool_inputs.py; DOCS.md says to pass .data.options.

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

goodwe_controller/DOCS.md

@@ -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/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 "
+check("the tool refuses rather than passing when it cannot run: its four exit "
-      "codes are distinct and 2 is not 0",
+      "codes are distinct and only 0 is a pass",
-      (RR.EXIT_OK, RR.EXIT_FINDING, RR.EXIT_COULD_NOT_RUN) == (0, 1, 2))
+      (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",
+check("...and the explicit opt-out RUNS, so the escape hatch exists - but a "
-      _deg_code in (RR.EXIT_OK, RR.EXIT_FINDING))
+      "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,255 @@
+"""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)
+
+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,
+        held = splits(sitesim, control, watchdog, a.cycles, a.load, a.gain,
-               a.feedback)
+                      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])
+    rc = 0
-    print(f"== {argv[1]}")
+    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
+the operator exports it and passes `--options` - the bare options object
-half is SKIPPED and says so - a skipped check must never look like a passed one.
+(`.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
+    2   the rehearsal could not be performed (bad file, unparseable config, a
-        entity-id model failed its own anchor). ⚠️ NOT A PASS.
+        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
+from app.store import (DEFAULTS, MAINT_SCHEMA_KEY, MAINT_UNIT_KEYS,  # noqa: E402
-                       unit_key)
+                       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,29 @@ def load_options(path: str) -> dict:
             obj = _mini_yaml(text)
         else:
             obj = yaml.safe_load(text)
-    if not isinstance(obj, dict):
+    # SAFETY-62: a wrapped export is a dict too, so assert the keys are options.
-        raise Refused(f"{path}: options must be a mapping, got {type(obj).__name__}")
+    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]
+_STORE_KEYS = {*DEFAULTS, *MAINT_UNIT_KEYS, SCHEMA_KEY, MAINT_SCHEMA_KEY}
+
+
+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 +1087,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 +1206,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__":