SAFETY-61: the control task's death ends the process, after the wind-down

A done-callback on run_control()'s task logs the traceback at the death and
sets `stop`; amain() runs the ordinary wind-down (shutdown() ->
command_site_zero()) and main() exits 70 (EX_SOFTWARE). Normal stops and
run_control()'s refusal path are ignored by the callback (still 0 / 78).

hass.py: a malformed JSON body, a JSON non-object, and an undecodable error
body are an absence / a failed write, not an exception into the loop.

checks() gains a first "control loop" row, red past FIRMWARE_INGEST_VETO_S.

Why exit rather than restart in-process, and what the Supervisor actually
does on exit (restarts only with the per-install Watchdog toggle on, default
off): DECISIONS/SAFETY-61-control-task-death.md.

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

@@ -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/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/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")