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

Merged
admin merged 1 commits from safety-61-control-task-death into release/1.0 2026-09-25 00:07:42 +02:00
4 changed files with 551 additions and 6 deletions
@@ -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.
+11 -3
View File
@@ -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
+61 -3
View File
@@ -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__":
+373
View File
@@ -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")