Improve control daemon event loop and add health LED smoke test. Use Redis wait-with-timeout reconcile, non-blocking reconnect, and systemd watchdog notify; document health contracts and a GPIO LED cycle script for bench checks. Co-authored-by: Cursor <cursoragent@cursor.com>

+Type=notify

+NotifyAccess=main

+WatchdogSec=10

+Single-threaded loop in `daemon.py` — not separate event and timer processes.

+

+each iteration:

+ wait on Redis pub/sub (control.command + temps.reading)

+ until a message arrives OR loop_interval_s elapses (default 1 s)

+ on message: update in-memory intents / latest temps only (no GPIO)

+ always: reconcile (heat FSM + filter policy + safety) → GPIO → publish if changed or heartbeat due

+ sd_notify WATCHDOG=1

+Message handlers never touch relays. Every wake — whether from a command, a temperature reading, or the 1 s timeout — runs the same reconcile step. Filter auto schedules, stale-temperature cutoff, and startup timers depend on that timeout path; the Health Daemon does not publish events that Control waits on for safety.

+

+**Control loop timing:**

+

+| Setting | Default | Role |

+|---|---|---|

+| `loop_interval_s` | 1.0 | Maximum time between reconcile passes when idle |

+| `heartbeat_s` | 5.0 | Minimum interval for `health:control` refresh when state unchanged |

+| systemd `WatchdogSec` | 10 | Restart if the loop stops notifying |

+

+Commands are applied as soon as pub/sub delivers them (typically milliseconds). Redis reconnect uses non-blocking backoff so reconcile still runs about once per second when Redis is down; heat fails safe on stale temps inside Control.

+

+For **Control** specifically (see [src/health/overview.md](src/health/overview.md)): Control publishes `health:control` every `heartbeat_s` (default **5 s**) with `status` of `ok`, `degraded`, or `error`. The Health Daemon drives the Control LED green / orange / red from that payload, and treats the heartbeat timestamp as stale after **10 s** (orange) or **60 s** (red). Health only drives LEDs — it does not send safety events to Control.

+

+7. **Hardening** — Control already uses systemd watchdog + notify; remaining: stale-data tuning, backup of config/history, structured logging across all services.

+- Control Daemon: documented and implemented single-threaded wait-with-timeout loop (`loop_interval_s` 1 s), immediate command handling, non-blocking Redis reconnect, and systemd watchdog for Control.

+The installed unit uses **`Type=notify`** and **`WatchdogSec=10`**: Control runs a 1 s wait-with-timeout loop and pings systemd each pass. Full loop behavior is documented in [README.md](README.md#main-loop).

+

+ control.tests.test_filter_policy control.tests.test_heat_fsm control.tests.test_redis_io -v

+ daemon.py # wait-with-timeout loop → reconcile → GPIO → publish

+ systemd_notify.py # sd_notify READY/WATCHDOG (no-op without NOTIFY_SOCKET)

+## Main loop

+

+One thread in `run_loop()` (`daemon.py`):

+

+1. Block on Redis pub/sub (`control.command`, `temps.reading`) until a message arrives **or** `loop_interval_s` elapses (default **1 s**).

+2. Message handlers update persisted intents and the latest temperature reading only.

+3. Every wake runs `tick()`: heat FSM, filter policy, safety, then GPIO (reconcile).

+4. Publish `control.state` / `control:latest` and `health:control` when the snapshot changes or every `heartbeat_s` (default **5 s**).

+5. Call `sd_notify("WATCHDOG=1")` each pass; systemd unit uses `Type=notify` and `WatchdogSec=10`.

+

+Commands apply as soon as Redis delivers them (typically milliseconds). Timer-based behavior (stale temps, filter auto window, startup delays) relies on the 1 s timeout when no messages arrive — not on a separate timer thread or on Health Daemon events.

+

+If Redis is unavailable, reconnect uses exponential backoff **without blocking the loop**; reconcile still runs about once per second and heat fails safe when temperature data is stale.

+

+- JSON fields: `service`, `ts` (UTC ISO-8601), `status`, `detail`, `pid`

+- `status`: `ok` | `degraded` | `error` (Health Daemon maps these to green / orange / red on the Control LED)

+- `error` on GPIO failure or Redis publish failure after reconnect

+

+Health Daemon staleness for Control uses the `ts` field: **>10 s** → orange LED, **>60 s** → red (see [health/overview.md](../health/overview.md)).

+- Unit: `hot-tub-control.service` (see [deploy/control/hot-tub-control.service](../../deploy/control/hot-tub-control.service))

+- `Type=notify`, `WatchdogSec=10`, `NotifyAccess=main`

+Loop design details: [Main loop](#main-loop) above.

+

+ "target_f_max": 110,

+ "temps_stale_s": 15,

+ "loop_interval_s": 1.0,

+ "heartbeat_s": 5.0

+import json

+from .systemd_notify import notify as sd_notify

+def _health_status(

+ state: ControlState,

+ cfg: ControlConfig,

+ reading: dict[str, Any] | None,

+ *,

+ gpio_error: bool = False,

+ publish_failed: bool = False,

+) -> tuple[str, str]:

+ if gpio_error:

+ return "error", "gpio failure"

+ if publish_failed:

+ return "error", "redis publish failed"

+ self.gpio_error = False

+ if self.gpio_error:

+ self.state.heat_pump = "off"

+ self.state.filter_pump = "off"

+ return state_snapshot(self.state, ts=_utc_now())

+ try:

+ tick_heat(self.state, self.ctx, self.cfg, self.reading)

+ filt = desired_filter_pump(self.state, self.cfg, self.ctx)

+ self.state.filter_pump = filt

+ _apply_gpio(self.relays, self.state)

+ except Exception:

+ logger.exception("tick failed (GPIO or FSM)")

+ self.gpio_error = True

+ self.relays.all_off()

+ self.state.heat_pump = "off"

+ self.state.filter_pump = "off"

+def _ingest_redis_messages(engine: ControlEngine, msgs: list[tuple[str, str]]) -> None:

+ for channel, data in msgs:

+ if channel == CHAN_COMMAND:

+ engine.handle_command_raw(data)

+ elif channel == CHAN_TEMPS_READING:

+ try:

+ payload = json.loads(data)

+ if isinstance(payload, dict):

+ engine.ingest_temps(payload)

+ except json.JSONDecodeError:

+ logger.warning("invalid temps.reading JSON")

+

+

+ sd_notify("READY=1")

+ next_due = time.monotonic()

+

+ timeout = max(0.0, next_due - time.monotonic())

+ msgs = redis_io.wait_messages(timeout_s=timeout)

+ _ingest_redis_messages(engine, msgs)

+ else:

+ delay = max(0.0, next_due - time.monotonic())

+ if delay > 0 and not _stop:

+ time.sleep(delay)

+ publish_failed = False

+ status, detail = _health_status(

+ engine.state,

+ cfg,

+ engine.reading,

+ gpio_error=engine.gpio_error,

+ publish_failed=False,

+ )

+ ok_state = redis_io.publish_state(snapshot)

+ ok_health = redis_io.publish_health(health)

+ if ok_state and ok_health:

+ engine.mark_published(snapshot, now_mono)

+ else:

+ status, detail = _health_status(

+ engine.state,

+ cfg,

+ engine.reading,

+ gpio_error=engine.gpio_error,

+ publish_failed=True,

+ )

+ health["status"] = status

+ health["detail"] = detail

+ health["ts"] = _utc_now()

+ redis_io.publish_health(health)

+ else:

+ engine.mark_published(snapshot, now_mono)

+

+ if now_mono >= next_due:

+ next_due = now_mono + cfg.loop_interval_s

+

+ sd_notify("WATCHDOG=1")

+_MAX_DRAIN = 32

+

+ self._retry_after_mono = 0.0

+ self._client = redis.from_url(

+ self._url,

+ decode_responses=True,

+ socket_connect_timeout=2,

+ socket_timeout=2,

+ )

+ now = time.monotonic()

+ if now < self._retry_after_mono:

+ return None

+ self._retry_after_mono = 0.0

+ self._retry_after_mono = now + min(self._backoff_s, 30.0)

+ @staticmethod

+ def _append_message(out: list[tuple[str, str]], msg: dict[str, Any] | None) -> bool:

+ if msg is None:

+ return False

+ if msg.get("type") != "message":

+ return False

+ channel = msg.get("channel")

+ data = msg.get("data")

+ if isinstance(channel, str) and isinstance(data, str):

+ out.append((channel, data))

+ return True

+ return False

+

+ def wait_messages(self, timeout_s: float) -> list[tuple[str, str]]:

+ """Block until the first message or timeout, then drain queued messages."""

+ first = pubsub.get_message(timeout=max(0.0, timeout_s))

+ self._append_message(out, first)

+ while len(out) < _MAX_DRAIN:

+ msg = pubsub.get_message(timeout=0)

+ if not self._append_message(out, msg):

+ continue

+ try:

+ body = json.dumps(payload)

+ client.set(KEY_CONTROL_LATEST, body)

+ client.publish(CHAN_CONTROL_STATE, body)

+ return True

+ except Exception as exc:

+ logger.warning("publish state failed: %s", exc)

+ self._client = None

+ self._pubsub = None

+ return False

+ try:

+ body = json.dumps(payload)

+ client.set(KEY_HEALTH, body)

+ client.publish(CHAN_HEALTH, body)

+ return True

+ except Exception as exc:

+ logger.warning("publish health failed: %s", exc)

+ self._client = None

+ self._pubsub = None

+ return False

+"""Minimal systemd sd_notify via NOTIFY_SOCKET (no systemd-python dependency)."""

+

+from __future__ import annotations

+

+import os

+import socket

+

+

+def notify(state: str) -> bool:

+ """Send a state string to systemd (e.g. READY=1, WATCHDOG=1). Returns True if sent."""

+ addr = os.environ.get("NOTIFY_SOCKET")

+ if not addr:

+ return False

+ if addr[0] == "@":

+ addr = "\0" + addr[1:]

+ payload = state.encode("utf-8")

+ if addr.startswith("\0"):

+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)

+ try:

+ sock.connect(addr)

+ sock.sendall(payload)

+ return True

+ except OSError:

+ return False

+ finally:

+ sock.close()

+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)

+ try:

+ sock.connect(addr)

+ sock.sendall(payload)

+ return True

+ except OSError:

+ return False

+ finally:

+ sock.close()

+"""Tests for ControlRedis wait_messages and non-blocking reconnect."""

+

+from __future__ import annotations

+

+import sys

+import time

+import unittest

+from unittest.mock import MagicMock, patch

+

+sys.modules.setdefault("redis", MagicMock())

+

+from control.redis_io import CHAN_COMMAND, ControlRedis

+

+

+class FakePubSub:

+ def __init__(self, script: list[object]) -> None:

+ self._script = list(script)

+ self._i = 0

+

+ def get_message(self, timeout: float = 0.0) -> dict[str, object] | None:

+ if self._i >= len(self._script):

+ return None

+ item = self._script[self._i]

+ self._i += 1

+ if isinstance(item, float):

+ time.sleep(item)

+ return None

+ return item

+

+

+class TestWaitMessages(unittest.TestCase):

+ @patch.object(ControlRedis, "subscribe_commands_and_temps")

+ def test_returns_promptly_on_first_message(self, sub_mock: MagicMock) -> None:

+ msg = {"type": "message", "channel": CHAN_COMMAND, "data": "{}"}

+ sub_mock.return_value = FakePubSub([msg])

+ cr = ControlRedis("redis://127.0.0.1:6379/0")

+ t0 = time.monotonic()

+ out = cr.wait_messages(timeout_s=5.0)

+ elapsed = time.monotonic() - t0

+ self.assertEqual(out, [(CHAN_COMMAND, "{}")])

+ self.assertLess(elapsed, 0.5)

+

+ @patch.object(ControlRedis, "subscribe_commands_and_temps")

+ def test_timeout_returns_empty(self, sub_mock: MagicMock) -> None:

+ sub_mock.return_value = FakePubSub([0.15])

+ cr = ControlRedis("redis://127.0.0.1:6379/0")

+ t0 = time.monotonic()

+ out = cr.wait_messages(timeout_s=0.1)

+ elapsed = time.monotonic() - t0

+ self.assertEqual(out, [])

+ self.assertGreaterEqual(elapsed, 0.08)

+ self.assertLess(elapsed, 0.35)

+

+ @patch.object(ControlRedis, "subscribe_commands_and_temps")

+ def test_drains_queued_messages(self, sub_mock: MagicMock) -> None:

+ m1 = {"type": "message", "channel": CHAN_COMMAND, "data": "a"}

+ m2 = {"type": "message", "channel": CHAN_COMMAND, "data": "b"}

+ sub_mock.return_value = FakePubSub([m1, m2, None])

+ cr = ControlRedis("redis://127.0.0.1:6379/0")

+ out = cr.wait_messages(timeout_s=1.0)

+ self.assertEqual(out, [(CHAN_COMMAND, "a"), (CHAN_COMMAND, "b")])

+

+

+class TestNonBlockingReconnect(unittest.TestCase):

+ def test_ensure_returns_immediately_before_retry_after(self) -> None:

+ cr = ControlRedis("redis://127.0.0.1:6379/0")

+ cr._retry_after_mono = time.monotonic() + 60.0

+ t0 = time.monotonic()

+ client = cr._ensure()

+ elapsed = time.monotonic() - t0

+ self.assertIsNone(client)

+ self.assertLess(elapsed, 0.05)

+

+ @patch.object(ControlRedis, "_connect")

+ def test_ensure_backoff_without_sleep(self, connect_mock: MagicMock) -> None:

+ client = MagicMock()

+ client.ping.side_effect = OSError("down")

+ connect_mock.return_value = client

+ cr = ControlRedis("redis://127.0.0.1:6379/0")

+ t0 = time.monotonic()

+ self.assertIsNone(cr._ensure())

+ elapsed = time.monotonic() - t0

+ self.assertLess(elapsed, 0.05)

+ self.assertGreater(cr._retry_after_mono, time.monotonic())

+

+

+if __name__ == "__main__":

+ unittest.main()

+#!/usr/bin/env python3

+"""Cycle each health status LED through red, orange, green, then off.

+

+Hardware smoke test for the three RGB LEDs (red + green channels only).

+Pin map matches overview.md. Do not run while a health daemon owns these GPIOs.

+

+On the Pi:

+ python3 health/led_test.py

+ python3 health/led_test.py --dry-run # off-Pi / no GPIO

+

+Requires: python3-gpiozero (sudo apt install python3-gpiozero)

+"""

+

+from __future__ import annotations

+

+import argparse

+import signal

+import sys

+import time

+from dataclasses import dataclass

+from typing import Literal

+

+Color = Literal["red", "orange", "green", "off"]

+

+STEP_S = 1.0

+

+# (name, red GPIO, green GPIO) — see src/health/overview.md

+LEDS: tuple[tuple[str, int, int], ...] = (

+ ("Temp", 27, 18),

+ ("Control", 25, 24),

+ ("Web", 6, 12),

+)

+

+COLORS: tuple[Color, ...] = ("red", "orange", "green", "off")

+

+_stop = False

+

+

+def _handle_signal(signum: int, _frame: object) -> None:

+ global _stop

+ print(f"\nReceived signal {signum}, stopping…", flush=True)

+ _stop = True

+

+

+@dataclass

+class RgLed:

+ name: str

+ red_pin: int

+ green_pin: int

+ dry_run: bool

+ _red: object = None

+ _green: object = None

+

+ def open(self) -> None:

+ if self.dry_run:

+ return

+ from gpiozero import OutputDevice

+

+ self._red = OutputDevice(self.red_pin, active_high=True, initial_value=False)

+ self._green = OutputDevice(self.green_pin, active_high=True, initial_value=False)

+

+ def set_color(self, color: Color) -> None:

+ red_on = color in ("red", "orange")

+ green_on = color in ("orange", "green")

+ if self.dry_run:

+ print(f" {self.name}: {color}", flush=True)

+ return

+ assert self._red is not None and self._green is not None

+ self._red.on() if red_on else self._red.off()

+ self._green.on() if green_on else self._green.off()

+ print(f" {self.name}: {color}", flush=True)

+

+ def off(self) -> None:

+ self.set_color("off")

+

+ def close(self) -> None:

+ if self.dry_run:

+ return

+ if self._red is not None:

+ self._red.off()

+ self._red.close()

+ self._red = None

+ if self._green is not None:

+ self._green.off()

+ self._green.close()

+ self._green = None

+

+

+def run_cycle(leds: list[RgLed]) -> None:

+ for led in leds:

+ if _stop:

+ return

+ print(f"{led.name} LED", flush=True)

+ for color in COLORS:

+ if _stop:

+ return

+ led.set_color(color)

+ if color != "off":

+ time.sleep(STEP_S)

+

+

+def main() -> int:

+ global _stop

+ parser = argparse.ArgumentParser(description="Cycle health status LEDs through red/orange/green/off.")

+ parser.add_argument(

+ "--dry-run",

+ action="store_true",

+ help="Print sequence only; do not use GPIO.",

+ )

+ args = parser.parse_args()

+

+ signal.signal(signal.SIGINT, _handle_signal)

+ signal.signal(signal.SIGTERM, _handle_signal)

+

+ leds = [RgLed(name, r, g, args.dry_run) for name, r, g in LEDS]

+ try:

+ for led in leds:

+ led.open()

+ print("Cycling Temp → Control → Web (Ctrl+C to stop)", flush=True)

+ while not _stop:

+ run_cycle(leds)

+ finally:

+ for led in leds:

+ led.off()

+ led.close()

+ print("All LED channels off.", flush=True)

+ return 0

+

+

+if __name__ == "__main__":

+ sys.exit(main())

+- Control - Read `health:control` from Redis (optionally subscribe to `health.control` for faster updates). Evaluate on a short poll loop (same wait-with-timeout pattern as Control is fine when this daemon is built).

+ - Payload from Control Daemon: `status` is `ok`, `degraded`, or `error`; `ts` is UTC ISO-8601 when the heartbeat was written.

+ - **`status: ok`** → green LED

+ - **`status: degraded`** → orange LED (e.g. stale temps while maintain, latched heat error phase)

+ - **`status: error`** → red LED (e.g. GPIO failure, Redis publish failure)

+ - **Stale heartbeat:** if `ts` is older than **10 seconds**, orange LED (overrides a stale `ok` in the key — roughly two missed Control heartbeats at default `heartbeat_s` 5 s)

+ - **Very stale:** if `ts` is older than **60 seconds**, red LED

+ - Health Daemon **only drives LEDs**; it does not publish events that Control waits on for safety. Stale temperature cutoff is enforced inside Control’s own 1 s reconcile loop.

+ - **Redis down:** Control keeps reconciling locally but cannot refresh `health:control` → Control LED goes orange then red by age (expected).

+ - **Control restart:** brief stale window (orange possible) until Control sends `READY=1` and publishes health again.

+ - on response lasting longer than 1 second - Orange LED

+

+## Hardware Setup

+

+The LEDs are 3 color LEDs, Red, Green and Blue. To create red, orange and green we will be using some combination of red and green. Blue is never used. To make the colors here are the combinations:

+

+- Red

+ - Red Channel ON

+ - Green Channel OFF

+- Orange

+ - Red Channel ON

+ - Green Channel ON

+- Green

+ - Red Channel OFF

+ - Green Channel ON

+

+Here is the configuration for each LED:

+

+- Temp:

+ - Red - GPIO 27

+ - Green - GPIO 18

+- Control:

+ - Red - GPIO 25

+ - Green - GPIO 24

+- Web:

+ - Red - GPIO 6

+ - Green - GPIO 12

+

+## Hardware smoke test

+

+From the repo `src` directory on the Pi (stop any health daemon first):

+

+```bash

+python3 health/led_test.py

+python3 health/led_test.py --dry-run # print sequence only

+```

+

+[`led_test.py`](led_test.py) loops Temp → Control → Web. For each LED it shows red, orange, and green for one second each, then turns that LED off before moving to the next. Ctrl+C turns all channels off.