From 02d8815e53d3f98dd23eafeabec2eb088192c0ed Mon Sep 17 00:00:00 2001 From: Sam Rolfe Date: Wed, 14 Oct 2026 11:10:30 +1100 Subject: [PATCH] Directive 1: voice intake watchdog (layer attribution, daily repeat, Apprise) Build the Hub Fixes Directive 1 watchdog as a small standalone service on .13. - Watch voice/audio_stream (paho); 15 min silence = 3 missed heartbeats. - Attribute failures in order: broker -> voice_bridge -> ESP32/no-publisher. The ESP32 is never blamed when the broker or the bridge broke. - Alert via Apprise on .35:8210 (tag sam_ntfy; daily repeat escalates to sam). - Persist state; repeat once a day while broken; recovery notice; alerts stop. - No auto-heal of any kind. Notification only. - Require publisher evidence (heartbeat or retained presence) before treating silence as a failure, so the unflashed Directive 3 firmware cannot cause the cry-wolf false alarms the spec warns about. - Docker: host network, read-only Docker socket, state volume, unless-stopped. - 23 unit tests; live evidence captured in watchdog/README.md. Co-Authored-By: Paperclip --- watchdog/Dockerfile | 14 + watchdog/README.md | 130 ++++++ watchdog/docker-compose.yml | 43 ++ watchdog/requirements.txt | 1 + watchdog/start_watchdog.sh | 19 + watchdog/tests/test_watchdog.py | 209 ++++++++++ watchdog/watchdog.py | 692 ++++++++++++++++++++++++++++++++ 7 files changed, 1108 insertions(+) create mode 100644 watchdog/Dockerfile create mode 100644 watchdog/README.md create mode 100644 watchdog/docker-compose.yml create mode 100644 watchdog/requirements.txt create mode 100755 watchdog/start_watchdog.sh create mode 100644 watchdog/tests/test_watchdog.py create mode 100644 watchdog/watchdog.py diff --git a/watchdog/Dockerfile b/watchdog/Dockerfile new file mode 100644 index 0000000..cd615b8 --- /dev/null +++ b/watchdog/Dockerfile @@ -0,0 +1,14 @@ +FROM python:3.12-slim + +# paho-mqtt only. Apprise is reached over plain HTTP with urllib, and the Docker +# socket is read through the stdlib http.client, so there is no other dependency. +RUN pip install --no-cache-dir "paho-mqtt==2.1.0" + +WORKDIR /app +COPY watchdog.py /app/watchdog.py + +# The service only reads state; the compose file provides the state volume and +# the (read-only) Docker socket. +VOLUME ["/var/lib/voice_intake_watchdog"] + +CMD ["python", "-u", "/app/watchdog.py"] diff --git a/watchdog/README.md b/watchdog/README.md new file mode 100644 index 0000000..46cbc9b --- /dev/null +++ b/watchdog/README.md @@ -0,0 +1,130 @@ +# Voice intake watchdog (Hub Fixes — Directive 1) + +Watches the house voice path and alerts Sam through Apprise when intake dies. +It **never heals anything** — no smart plug, no power-cycle, no restart. +Notification only. Sam fixes the failed layer by hand. + +This is the fix for the 2026-10-13 outage: the `.13` broker restarted, +`voice_bridge` went deaf and the ESP32 stopped publishing, and nobody noticed +for two days. + +## What it watches + +"Voice intake alive" means all three layers are healthy, checked in this order: + +| Layer | Signal | Failure message says | +|---|---|---| +| **broker** | `.13:1883` reachable (live MQTT session or TCP probe) | check the Mosquitto container; **do not** power-cycle the ESP32 | +| **bridge** | `voice_bridge` container running **and** subscribed to `voice/audio_stream` (Docker socket) | start the bridge; **do not** power-cycle the ESP32 | +| **publisher** | a message on `voice/audio_stream` within `SILENCE_TIMEOUT_SECONDS`, or retained `voice/status` presence | "Likely the ESP32 — try a power-cycle." | + +Layer attribution is the point: if something else broke, the ESP32 is not blamed. + +### Why 15 minutes is meaningful + +The Directive 3 firmware publishes a **heartbeat on `voice/audio_stream` every +5 minutes**, so 15 minutes of silence is three missed beats — not "nobody +spoke". The firmware also sets a retained `voice/status` presence (`online` on +connect, `offline` as a Last Will) for immediate attribution. + +Until a heartbeat or presence has been seen, the watchdog does **not** treat +silence as a publisher failure (that state means the heartbeat firmware is not +flashed yet, and silence is normal). This is controlled by +`REQUIRE_PUBLISHER_EVIDENCE` (default `1`). + +## Repeat policy + +- First failure → alert immediately. +- Same layer still down → repeat once per `REPEAT_INTERVAL_SECONDS` (default one + day), for as long as it stays broken. +- A different layer fails → that is a new alert (immediate). +- Recovery → one recovery notice, then the state is cleared and alerts stop. +- State is persisted in a Docker volume, so a watchdog restart neither re-spams + nor forgets an ongoing outage. + +First alerts use tag `APPRISE_TAG` (`sam_ntfy`); daily repeats escalate to +`APPRISE_TAG_REPEAT` (`sam`). Both live in the Apprise config on `.35`. + +## Files + +``` +watchdog/ + watchdog.py the service (paho + stdlib only) + Dockerfile python:3.12-slim + paho-mqtt==2.1.0 + docker-compose.yml host network, read-only Docker socket, state volume + start_watchdog.sh sources ~/.config/environment.d/10-secrets.conf, then compose up + tests/test_watchdog.py unit tests for classification, repeats, messages +``` + +## Deploy + +```bash +cd /home/sam/paseo/projects/hub_fixes/watchdog +./start_watchdog.sh # normal operation +DRY_RUN=1 ./start_watchdog.sh # log alerts, send nothing +SILENCE_TIMEOUT_SECONDS=60 CHECK_INTERVAL_SECONDS=5 ./start_watchdog.sh # test mode +``` + +It runs as container `voice_intake_watchdog`, `network_mode: host`, no listening +port, `restart: unless-stopped`. The Docker socket is mounted **read-only** and +only ever read (`/containers/voice_bridge/json` and its logs). + +## Test + +```bash +python3 -m unittest discover -s watchdog/tests -v +``` + +## Verify (evidence) + +All commands below were run on `.13` on 2026-10-14. + +**1. Stop `voice_bridge` → alert arrives within 15 minutes** + +``` +$ docker stop voice_bridge +$ # watchdog, real Apprise, 5 s checks +WARNING ALERT kind=alert layer=bridge -> tag=sam_ntfy +INFO apprise notify OK HTTP 200 tag=sam_ntfy: Voice intake DOWN — voice_bridge is not running +``` + +The bridge layer is checked directly, so the alert fires on the next check +(seconds), well inside the 15-minute budget. + +**2. Start it again → alerts stop** + +``` +$ docker start voice_bridge +WARNING ALERT kind=recovery layer=bridge -> tag=sam_ntfy +INFO apprise notify OK HTTP 200 tag=sam_ntfy: Voice intake recovered +$ # no further alerts once healthy +``` + +**3. Broker down → names the broker, not the ESP32** + +``` +$ # watchdog pointed at an unreachable broker address +WARNING ALERT kind=alert layer=broker -> tag=sam_ntfy +TITLE: Voice intake DOWN — .13 MQTT broker unreachable +... Do NOT power-cycle the ESP32 — the broker layer failed, not the device. +``` + +**4. Daily repeat (shortened interval)** + +``` +$ # SILENCE_TIMEOUT=1 REPEAT_INTERVAL=6 CHECK_INTERVAL=2, dry-run +ALERT kind=alert layer=publisher -> tag=sam_ntfy (first) +ALERT kind=repeat layer=publisher -> tag=sam (after the interval) +``` + +The unit tests also drive three consecutive daily repeats +(`test_daily_repeats_keep_firing`). + +## Not in scope / boundaries + +- No auto-heal of any kind. +- The `.13:1883` plane is untouched (plain, anonymous). The watchdog connects as + a normal subscriber and publishes nothing. +- The broker was **not** stopped to prove layer attribution; an unreachable + broker address exercises the same code path without disturbing n8n or the + kids' 8883 devices. diff --git a/watchdog/docker-compose.yml b/watchdog/docker-compose.yml new file mode 100644 index 0000000..c3f1138 --- /dev/null +++ b/watchdog/docker-compose.yml @@ -0,0 +1,43 @@ +# voice intake watchdog — Hub Fixes Directive 1 +# +# Network mode host: the watchdog must reach the .13 broker on localhost:1883 +# and the Apprise API on 192.168.20.35:8210. No new listening port is opened. +# +# The Docker socket is mounted read-only so the watchdog can check whether the +# `voice_bridge` container is running and subscribed. It never writes to it. +# +# Bring it up with ./start_watchdog.sh (sources the lab secrets first) or: +# DRY_RUN=1 docker compose up -d --build # no real notifications + +services: + voice-intake-watchdog: + build: . + image: voice_intake_watchdog:local + container_name: voice_intake_watchdog + restart: unless-stopped + network_mode: host + environment: + - MQTT_HOST=192.168.20.13 + - MQTT_PORT=1883 + - MQTT_USER + - MQTT_PASS + - AUDIO_TOPIC=voice/audio_stream + - STATUS_TOPIC=voice/status + - SILENCE_TIMEOUT_SECONDS=${SILENCE_TIMEOUT_SECONDS:-900} + - CHECK_INTERVAL_SECONDS=${CHECK_INTERVAL_SECONDS:-30} + - REPEAT_INTERVAL_SECONDS=${REPEAT_INTERVAL_SECONDS:-86400} + - APPRISE_NOTIFY_URL=http://192.168.20.35:8210/notify/apprise/ + - APPRISE_TAG=sam_ntfy + - APPRISE_TAG_REPEAT=sam + - BRIDGE_CONTAINER=voice_bridge + - DOCKER_SOCKET=/var/run/docker.sock + - STATE_FILE=/var/lib/voice_intake_watchdog/state.json + - DRY_RUN=${DRY_RUN:-0} + - REQUIRE_PUBLISHER_EVIDENCE=${REQUIRE_PUBLISHER_EVIDENCE:-1} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + volumes: + - /var/run/docker.sock:/var/run/docker.sock:ro + - watchdog_state:/var/lib/voice_intake_watchdog + +volumes: + watchdog_state: diff --git a/watchdog/requirements.txt b/watchdog/requirements.txt new file mode 100644 index 0000000..96a2716 --- /dev/null +++ b/watchdog/requirements.txt @@ -0,0 +1 @@ +paho-mqtt==2.1.0 diff --git a/watchdog/start_watchdog.sh b/watchdog/start_watchdog.sh new file mode 100755 index 0000000..3aa00e3 --- /dev/null +++ b/watchdog/start_watchdog.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# Start the voice intake watchdog. +# +# environment.d reaches the systemd user manager, not a fresh SSH shell, so the +# secrets file is sourced here the same way ~/voice_bridge/start_bridge.sh does. +# MQTT_USER/MQTT_PASS are optional: the .13:1883 plane allows anonymous clients. +set -euo pipefail + +cd "$(dirname "$0")" + +SECRETS_FILE="${SECRETS_FILE:-$HOME/.config/environment.d/10-secrets.conf}" +if [[ -f "$SECRETS_FILE" ]]; then + set -a + # shellcheck disable=SC1090 + source "$SECRETS_FILE" + set +a +fi + +exec docker compose up -d --build diff --git a/watchdog/tests/test_watchdog.py b/watchdog/tests/test_watchdog.py new file mode 100644 index 0000000..fd1b9bf --- /dev/null +++ b/watchdog/tests/test_watchdog.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Unit tests for the voice intake watchdog. + +Run with: python3 -m unittest discover -s tests -v +(No third-party packages needed; paho is imported lazily by watchdog.py.) +""" + +import os +import sys +import time +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from watchdog import ( # noqa: E402 + AlertState, + Config, + Layer, + Snapshot, + advance, + build_message, + classify, +) + +NOW = 1_700_000_000.0 + + +def snap(**kw): + base = dict( + now=NOW, + broker_reachable=True, + bridge_running=True, + bridge_subscribed=True, + last_audio_at=NOW - 10, + started_at=NOW - 3600, + presence=None, + ) + base.update(kw) + return Snapshot(**base) + + +class ClassifyTests(unittest.TestCase): + def test_healthy(self): + self.assertEqual(classify(snap(), 900), Layer.HEALTHY) + + def test_broker_down_wins_over_everything(self): + s = snap(broker_reachable=False, bridge_running=False, last_audio_at=NOW - 9999) + self.assertEqual(classify(s, 900), Layer.BROKER) + + def test_bridge_down(self): + s = snap(bridge_running=False) + self.assertEqual(classify(s, 900), Layer.BRIDGE) + + def test_bridge_running_but_not_subscribed(self): + s = snap(bridge_running=True, bridge_subscribed=False) + self.assertEqual(classify(s, 900), Layer.BRIDGE) + + def test_publisher_silent_after_timeout(self): + s = snap(last_audio_at=NOW - 901) + self.assertEqual(classify(s, 900), Layer.PUBLISHER) + + def test_just_inside_timeout_is_healthy(self): + s = snap(last_audio_at=NOW - 899) + self.assertEqual(classify(s, 900), Layer.HEALTHY) + + def test_presence_offline_is_immediate_publisher_failure(self): + s = snap(presence="offline", last_audio_at=NOW - 1) + self.assertEqual(classify(s, 900), Layer.PUBLISHER) + + def test_no_audio_since_start_uses_start_time(self): + s = snap(last_audio_at=None, started_at=NOW - 1000) + self.assertEqual(classify(s, 900, require_evidence=False), Layer.PUBLISHER) + + def test_unarmed_silence_is_not_a_failure(self): + # No heartbeat and no retained presence: the heartbeat firmware is not + # flashed yet, so silence must not be reported as a dead publisher. + s = snap(last_audio_at=None, presence=None, started_at=NOW - 100000) + self.assertEqual(classify(s, 900), Layer.HEALTHY) + + def test_presence_online_arms_the_silence_check(self): + s = snap(last_audio_at=None, presence="online", started_at=NOW - 1000) + self.assertEqual(classify(s, 900), Layer.PUBLISHER) + + +class AdvanceTests(unittest.TestCase): + def setUp(self): + self.cfg = Config(silence_timeout=900, repeat_interval=86400) + + def test_first_failure_alerts_once(self): + state = AlertState() + alerts = advance(state, snap(bridge_running=False), self.cfg) + self.assertEqual([a.kind for a in alerts], ["alert"]) + self.assertEqual(alerts[0].layer, "bridge") + self.assertEqual(state.layer, "bridge") + + def test_no_repeat_before_interval(self): + state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) + alerts = advance( + state, snap(now=NOW + 60, bridge_running=False), self.cfg + ) + self.assertEqual(alerts, []) + + def test_repeat_after_interval(self): + state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) + alerts = advance( + state, + snap(now=NOW + 86400, bridge_running=False), + self.cfg, + ) + self.assertEqual([a.kind for a in alerts], ["repeat"]) + self.assertEqual(state.last_alert_at, NOW + 86400) + + def test_daily_repeats_keep_firing(self): + state = AlertState(layer="publisher", first_seen_at=NOW, last_alert_at=NOW) + fires = [] + for day in (1, 2, 3): + alerts = advance( + state, + snap(now=NOW + day * 86400, last_audio_at=NOW - 5000), + self.cfg, + ) + fires.extend(a.kind for a in alerts) + self.assertEqual(fires, ["repeat", "repeat", "repeat"]) + + def test_recovery_clears_state_and_notifies_once(self): + state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) + alerts = advance(state, snap(now=NOW + 120), self.cfg) + self.assertEqual([a.kind for a in alerts], ["recovery"]) + self.assertIsNone(state.layer) + self.assertIsNone(state.last_alert_at) + # A second healthy cycle is silent. + self.assertEqual(advance(state, snap(now=NOW + 150), self.cfg), []) + + def test_layer_change_is_a_new_alert(self): + state = AlertState(layer="publisher", first_seen_at=NOW, last_alert_at=NOW) + alerts = advance( + state, + snap(now=NOW + 60, bridge_running=False), + self.cfg, + ) + self.assertEqual([a.kind for a in alerts], ["alert"]) + self.assertEqual(alerts[0].layer, "bridge") + + +class MessageTests(unittest.TestCase): + def setUp(self): + self.cfg = Config( + apprise_tag="sam_ntfy", + apprise_tag_repeat="sam", + silence_timeout=900, + ) + + def test_broker_names_broker_and_not_esp32(self): + title, body, tag = build_message( + __import__("watchdog").Alert("alert", "broker", NOW), self.cfg + ) + self.assertIn("broker", title.lower()) + self.assertIn("mosquitto", body.lower()) + self.assertIn("Do NOT power-cycle the ESP32", body) + self.assertEqual(tag, "sam_ntfy") + + def test_bridge_names_bridge_and_not_esp32(self): + title, body, _ = build_message( + __import__("watchdog").Alert("alert", "bridge", NOW), self.cfg + ) + self.assertIn("voice_bridge", title) + self.assertIn("docker compose up -d", body) + self.assertIn("Do NOT power-cycle the ESP32", body) + + def test_publisher_says_power_cycle(self): + title, body, _ = build_message( + __import__("watchdog").Alert("alert", "publisher", NOW), self.cfg + ) + self.assertIn("ESP32", title) + self.assertIn("try a power-cycle", body.lower()) + + def test_repeat_escalates_tag(self): + _, _, tag = build_message( + __import__("watchdog").Alert("repeat", "publisher", NOW), self.cfg + ) + self.assertEqual(tag, "sam") + + def test_recovery_uses_primary_tag(self): + _, _, tag = build_message( + __import__("watchdog").Alert("recovery", "publisher", NOW), self.cfg + ) + self.assertEqual(tag, "sam_ntfy") + + +class DockerTimeTests(unittest.TestCase): + def test_parses_rfc3339_nano(self): + import datetime as dt + + from watchdog import _parse_docker_time + + expected = dt.datetime( + 2026, 10, 14, 9, 3, 11, 123456, tzinfo=dt.timezone.utc + ).timestamp() + ts = _parse_docker_time("2026-10-14T09:03:11.123456789Z") + self.assertAlmostEqual(ts, expected, places=3) + + def test_bad_value_falls_back_to_now(self): + from watchdog import _parse_docker_time + + self.assertAlmostEqual(_parse_docker_time("not-a-time"), time.time(), delta=5) + + +if __name__ == "__main__": + unittest.main() diff --git a/watchdog/watchdog.py b/watchdog/watchdog.py new file mode 100644 index 0000000..784c610 --- /dev/null +++ b/watchdog/watchdog.py @@ -0,0 +1,692 @@ +#!/usr/bin/env python3 +"""voice intake watchdog — Hub Fixes Directive 1. + +Watches the house voice path and alerts Sam through Apprise when intake dies. +It NEVER heals anything: no smart plug, no power-cycle, no restart. Notification +only. Sam fixes the layer by hand. + +What it watches +--------------- +"Voice intake alive" means all three layers are healthy, checked in this order: + + 1. BROKER the .13 Mosquitto broker (:1883) is reachable. + 2. BRIDGE the `voice_bridge` container is running AND subscribed to + `voice/audio_stream` (checked through the Docker socket). + 3. PUBLISHER the ESP32 is publishing. Since the Directive 3 firmware, the + device sends a heartbeat on `voice/audio_stream` every 5 minutes, + so 15 minutes of topic silence = three missed heartbeats. + A retained `voice/status == "offline"` Last Will is a second, + immediate signal. + +The layer attribution rule from the spec: if something else broke, the ESP32 is +NOT blamed. The alert names the layer that actually failed and what to do about +it. + +Repeat policy +------------- +When a layer fails, an alert goes out immediately. While the same layer stays +down, the alert repeats once per `REPEAT_INTERVAL_SECONDS` (default: one day). +When a different layer fails, that is a new alert. When intake recovers, the +state is cleared and a single recovery notice is sent. State is persisted so a +watchdog restart does not re-spam or silently forget an ongoing outage. + +Environment (all optional, sane defaults for the lab) +------------------------------------------------------ + MQTT_HOST=192.168.20.13 + MQTT_PORT=1883 + MQTT_USER / MQTT_PASS (optional; the 1883 plane is anonymous) + AUDIO_TOPIC=voice/audio_stream + STATUS_TOPIC=voice/status + SILENCE_TIMEOUT_SECONDS=900 (15 minutes) + CHECK_INTERVAL_SECONDS=30 + REPEAT_INTERVAL_SECONDS=86400 (one day) + APPRISE_NOTIFY_URL=http://192.168.20.35:8210/notify/apprise/ + APPRISE_TAG=sam_ntfy (first alert) + APPRISE_TAG_REPEAT=sam (daily repeat escalates to all Sam channels) + APPRISE_TIMEOUT_SECONDS=20 + BRIDGE_CONTAINER=voice_bridge + DOCKER_SOCKET=/var/run/docker.sock + DOCKER_TIMEOUT_SECONDS=8 + STATE_FILE=/var/lib/voice_intake_watchdog/state.json + DRY_RUN=0 (1 => print alerts instead of sending them) + REQUIRE_PUBLISHER_EVIDENCE=1 (1 => do not treat silence as a publisher + failure until a heartbeat or presence has + been seen; prevents false alarms before the + Directive 3 heartbeat firmware is flashed) + LOG_LEVEL=INFO +""" + +from __future__ import annotations + +import argparse +import dataclasses +import datetime as dt +import http.client +import json +import logging +import os +import socket +import sys +import threading +import time +import urllib.request +from enum import Enum +from typing import Callable, Optional + +LOG = logging.getLogger("voice_intake_watchdog") + +try: + import paho.mqtt.client as mqtt +except ImportError: # pragma: no cover - only hit when paho is missing + mqtt = None + + +# --------------------------------------------------------------------------- +# Layers +# --------------------------------------------------------------------------- + + +class Layer(str, Enum): + BROKER = "broker" + BRIDGE = "bridge" + PUBLISHER = "publisher" + HEALTHY = "healthy" + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + + +def _env_int(name: str, default: int) -> int: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return int(raw) + + +def _env_float(name: str, default: float) -> float: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return float(raw) + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return raw.strip().lower() in ("1", "true", "yes", "on") + + +@dataclasses.dataclass +class Config: + mqtt_host: str = "192.168.20.13" + mqtt_port: int = 1883 + mqtt_user: Optional[str] = None + mqtt_pass: Optional[str] = None + audio_topic: str = "voice/audio_stream" + status_topic: str = "voice/status" + silence_timeout: float = 900.0 + check_interval: float = 30.0 + repeat_interval: float = 86400.0 + apprise_notify_url: str = "http://192.168.20.35:8210/notify/apprise/" + apprise_tag: str = "sam_ntfy" + apprise_tag_repeat: str = "sam" + apprise_timeout: float = 20.0 + bridge_container: str = "voice_bridge" + docker_socket: str = "/var/run/docker.sock" + docker_timeout: float = 8.0 + state_file: str = "/var/lib/voice_intake_watchdog/state.json" + dry_run: bool = False + require_publisher_evidence: bool = True + + @classmethod + def from_env(cls) -> "Config": + return cls( + mqtt_host=os.environ.get("MQTT_HOST", cls.mqtt_host), + mqtt_port=_env_int("MQTT_PORT", cls.mqtt_port), + mqtt_user=os.environ.get("MQTT_USER") or None, + mqtt_pass=os.environ.get("MQTT_PASS") or None, + audio_topic=os.environ.get("AUDIO_TOPIC", cls.audio_topic), + status_topic=os.environ.get("STATUS_TOPIC", cls.status_topic), + silence_timeout=_env_float("SILENCE_TIMEOUT_SECONDS", cls.silence_timeout), + check_interval=_env_float("CHECK_INTERVAL_SECONDS", cls.check_interval), + repeat_interval=_env_float("REPEAT_INTERVAL_SECONDS", cls.repeat_interval), + apprise_notify_url=os.environ.get("APPRISE_NOTIFY_URL", cls.apprise_notify_url), + apprise_tag=os.environ.get("APPRISE_TAG", cls.apprise_tag), + apprise_tag_repeat=os.environ.get("APPRISE_TAG_REPEAT", cls.apprise_tag_repeat), + apprise_timeout=_env_float("APPRISE_TIMEOUT_SECONDS", cls.apprise_timeout), + bridge_container=os.environ.get("BRIDGE_CONTAINER", cls.bridge_container), + docker_socket=os.environ.get("DOCKER_SOCKET", cls.docker_socket), + docker_timeout=_env_float("DOCKER_TIMEOUT_SECONDS", cls.docker_timeout), + state_file=os.environ.get("STATE_FILE", cls.state_file), + dry_run=_env_bool("DRY_RUN", cls.dry_run), + require_publisher_evidence=_env_bool( + "REQUIRE_PUBLISHER_EVIDENCE", cls.require_publisher_evidence + ), + ) + + +# --------------------------------------------------------------------------- +# Classification and repeat state machine (pure, unit-testable) +# --------------------------------------------------------------------------- + + +@dataclasses.dataclass +class Snapshot: + now: float + broker_reachable: bool + bridge_running: bool + bridge_subscribed: bool + last_audio_at: Optional[float] + started_at: float + presence: Optional[str] = None # ESP32 retained voice/status, or None + + +def publisher_armed(snap: Snapshot) -> bool: + """True once we have evidence that a publisher exists for this topic. + + The Directive 3 firmware publishes a retained `voice/status` presence + (online/offline) and a heartbeat on the audio topic. Until one of those has + been seen, silence on `voice/audio_stream` only means "nobody spoke" or + "the heartbeat firmware is not flashed yet" — not "the device is dead". + Alerting in that state would cry wolf every quiet night, which is the exact + failure this project exists to prevent. + """ + return snap.presence is not None or snap.last_audio_at is not None + + +def classify(snap: Snapshot, silence_timeout: float, require_evidence: bool = True) -> Layer: + """Name the layer that failed, in order: broker, bridge, publisher.""" + if not snap.broker_reachable: + return Layer.BROKER + if not (snap.bridge_running and snap.bridge_subscribed): + return Layer.BRIDGE + if snap.presence == "offline": + return Layer.PUBLISHER + if require_evidence and not publisher_armed(snap): + return Layer.HEALTHY + reference = snap.last_audio_at if snap.last_audio_at is not None else snap.started_at + if snap.now - reference > silence_timeout: + return Layer.PUBLISHER + return Layer.HEALTHY + + +@dataclasses.dataclass +class AlertState: + layer: Optional[str] = None # currently failing layer, or None + first_seen_at: Optional[float] = None + last_alert_at: Optional[float] = None + last_audio_at: Optional[float] = None + started_at: Optional[float] = None + + +@dataclasses.dataclass +class Alert: + kind: str # "alert" | "repeat" | "recovery" + layer: str + since: Optional[float] = None + + +def advance(state: AlertState, snap: Snapshot, cfg: Config) -> list[Alert]: + """Pure state transition. Returns the alerts that must be delivered now.""" + layer = classify(snap, cfg.silence_timeout, cfg.require_publisher_evidence) + + if layer is Layer.HEALTHY: + if state.layer is not None: + recovered_from = state.layer + since = state.first_seen_at + state.layer = None + state.first_seen_at = None + state.last_alert_at = None + return [Alert(kind="recovery", layer=recovered_from, since=since)] + return [] + + if state.layer != layer.value: + state.layer = layer.value + state.first_seen_at = snap.now + state.last_alert_at = snap.now + return [Alert(kind="alert", layer=layer.value, since=snap.now)] + + if state.last_alert_at is None or (snap.now - state.last_alert_at) >= cfg.repeat_interval: + state.last_alert_at = snap.now + return [Alert(kind="repeat", layer=layer.value, since=state.first_seen_at)] + + return [] + + +# --------------------------------------------------------------------------- +# Message text +# --------------------------------------------------------------------------- + + +def _hhmm(ts: Optional[float]) -> str: + if ts is None: + return "??:??" + return dt.datetime.fromtimestamp(ts).strftime("%H:%M") + + +def build_message(alert: Alert, cfg: Config) -> tuple[str, str, str]: + """Return (title, body, tag) for an alert. Names what failed and what to do.""" + silent_since = _hhmm(alert.since) + if alert.kind == "recovery": + title = "Voice intake recovered" + body = ( + f"Voice intake recovered at {_hhmm(time.time())}.\n\n" + "The watchdog cleared its alert state. It will alert again if intake dies." + ) + return title, body, cfg.apprise_tag + + tag = cfg.apprise_tag if alert.kind == "alert" else cfg.apprise_tag_repeat + repeat_note = { + "alert": "This alert repeats once a day while the fault stays.", + "repeat": "Repeat of an unresolved fault. It repeats again in a day.", + }[alert.kind] + + if alert.layer == Layer.BROKER.value: + title = "Voice intake DOWN — .13 MQTT broker unreachable" + body = ( + f"Voice intake is dead: the .13 MQTT broker ({cfg.mqtt_host}:{cfg.mqtt_port}) " + f"has been unreachable since {silent_since}.\n\n" + "Nothing can reach the voice path. n8n and the ESP32 also use this broker.\n\n" + "What to do: check the Mosquitto container on .13 " + "(docker ps; docker logs mosquitto).\n" + "Do NOT power-cycle the ESP32 — the broker layer failed, not the device.\n\n" + f"{repeat_note}" + ) + elif alert.layer == Layer.BRIDGE.value: + title = "Voice intake DOWN — voice_bridge is not running" + body = ( + f"Voice intake is dead: the broker is up, but `{cfg.bridge_container}` has not " + f"been running and subscribed since {silent_since}.\n\n" + "The ESP32 may still be publishing, but nothing is consuming the audio.\n\n" + "What to do: start the bridge on .13\n" + " cd ~/voice_bridge && docker compose up -d\n" + "Do NOT power-cycle the ESP32 — the bridge layer failed, not the device.\n\n" + f"{repeat_note}" + ) + else: # publisher + title = "Voice intake silent — no publisher (likely the ESP32)" + body = ( + f"Voice intake silent since {silent_since}: no message on " + f"`{cfg.audio_topic}` for {int(cfg.silence_timeout // 60)} minutes. " + "The broker and voice_bridge are healthy, so the publisher stopped.\n\n" + "Likely the ESP32 — try a power-cycle.\n\n" + f"{repeat_note}" + ) + return title, body, tag + + +# --------------------------------------------------------------------------- +# Notifier (Apprise) +# --------------------------------------------------------------------------- + + +class AppriseNotifier: + def __init__(self, cfg: Config): + self.cfg = cfg + self.sent: list[dict] = [] + + def send(self, title: str, body: str, tag: str) -> bool: + record = { + "title": title, + "tag": tag, + "at": time.time(), + "dry_run": self.cfg.dry_run, + } + if self.cfg.dry_run: + LOG.info("[dry-run] notify tag=%s title=%s", tag, title) + record["body"] = body + self.sent.append(record) + return True + payload = json.dumps({"title": title, "body": body, "tag": tag}).encode() + request = urllib.request.Request( + self.cfg.apprise_notify_url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=self.cfg.apprise_timeout) as resp: + ok = 200 <= resp.status < 300 + record["status"] = resp.status + self.sent.append(record) + if ok: + LOG.info("apprise notify OK HTTP %s tag=%s: %s", resp.status, tag, title) + else: + LOG.error("apprise returned HTTP %s for %r", resp.status, title) + return ok + except Exception as exc: # network/auth/HTTP error + record["error"] = str(exc) + self.sent.append(record) + LOG.error("apprise notify failed for %r: %s", title, exc) + return False + + +# --------------------------------------------------------------------------- +# Broker watcher (paho) +# --------------------------------------------------------------------------- + + +class MqttWatcher: + """Keeps a subscription on the audio topic and records the last message time.""" + + def __init__(self, cfg: Config): + if mqtt is None: + raise RuntimeError("paho-mqtt is not installed") + self.cfg = cfg + self.lock = threading.Lock() + self.last_audio_at: Optional[float] = None + self.presence: Optional[str] = None + self.connected = False + self.client = mqtt.Client( + mqtt.CallbackAPIVersion.VERSION2, + client_id="voice_intake_watchdog", + ) + if cfg.mqtt_user: + self.client.username_pw_set(cfg.mqtt_user, cfg.mqtt_pass) + self.client.on_connect = self._on_connect + self.client.on_disconnect = self._on_disconnect + self.client.on_message = self._on_message + self.client.reconnect_delay_set(min_delay=1, max_delay=15) + + # callbacks ------------------------------------------------------------- + def _on_connect(self, client, userdata, flags, reason_code, properties=None): + if getattr(reason_code, "is_failure", False): + LOG.warning("broker refused connection: %s", reason_code) + with self.lock: + self.connected = False + return + with self.lock: + self.connected = True + LOG.info( + "connected to %s:%s; subscribing %s + %s", + self.cfg.mqtt_host, + self.cfg.mqtt_port, + self.cfg.audio_topic, + self.cfg.status_topic, + ) + client.subscribe([(self.cfg.audio_topic, 0), (self.cfg.status_topic, 0)]) + + def _on_disconnect(self, client, userdata, flags, reason_code, properties=None): + with self.lock: + self.connected = False + LOG.warning("broker disconnected: %s", reason_code) + + def _on_message(self, client, userdata, msg): + now = time.time() + with self.lock: + if msg.topic == self.cfg.audio_topic: + self.last_audio_at = now + elif msg.topic == self.cfg.status_topic: + value = msg.payload.decode("utf-8", "replace").strip().lower() + self.presence = value + # A retained "offline" proves the device's Last Will fired. + LOG.info("voice/status presence: %s", value) + + # lifecycle ------------------------------------------------------------- + def start(self): + # connect_async + loop_start so a broker outage cannot block startup. + try: + self.client.connect_async(self.cfg.mqtt_host, self.cfg.mqtt_port, keepalive=30) + except Exception as exc: # pragma: no cover - defensive + LOG.error("mqtt connect_async failed: %s", exc) + self.client.loop_start() + + def stop(self): + self.client.loop_stop() + try: + self.client.disconnect() + except Exception: + pass + + +def tcp_probe(host: str, port: int, timeout: float = 3.0) -> bool: + try: + with socket.create_connection((host, port), timeout=timeout): + return True + except OSError: + return False + + +# --------------------------------------------------------------------------- +# Docker bridge check (unix socket, stdlib only) +# --------------------------------------------------------------------------- + + +class DockerClient: + def __init__(self, socket_path: str, timeout: float): + self.socket_path = socket_path + self.timeout = timeout + + def _get(self, path: str) -> tuple[int, bytes]: + conn = http.client.HTTPConnection("localhost", timeout=self.timeout) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + sock.connect(self.socket_path) + conn.sock = sock + try: + conn.request("GET", path) + resp = conn.getresponse() + return resp.status, resp.read() + finally: + conn.close() + + def container_json(self, name: str) -> Optional[dict]: + try: + status, body = self._get(f"/containers/{name}/json") + except OSError: + return None + if status == 404: + return {} + if status != 200: + return None + try: + return json.loads(body) + except json.JSONDecodeError: + return None + + def logs_since(self, name: str, since_epoch: float) -> bytes: + try: + status, body = self._get( + f"/containers/{name}/logs?stdout=1&stderr=1&since={int(since_epoch)}" + ) + except OSError: + return b"" + if status != 200: + return b"" + return body + + +def _parse_docker_time(value: str) -> float: + # e.g. 2026-10-14T09:03:11.123456789Z + value = value.replace("Z", "+00:00") + if "." in value: + head, _, tail = value.partition(".") + digits = "" + for ch in tail: + if ch.isdigit(): + digits += ch + else: + break + rest = tail[len(digits):] + # keep at most 6 fractional digits for fromisoformat + value = f"{head}.{digits[:6]}{rest}" + try: + return dt.datetime.fromisoformat(value).timestamp() + except ValueError: + return time.time() + + +@dataclasses.dataclass +class BridgeState: + checked: bool # False when the Docker socket is unavailable + running: bool + subscribed: bool + detail: str + + +def check_bridge(docker: DockerClient, container: str) -> BridgeState: + obj = docker.container_json(container) + if obj is None: + # Docker socket unavailable: do not invent a bridge failure. + return BridgeState(False, True, True, "docker socket unavailable; bridge check skipped") + if obj == {}: + return BridgeState(True, False, False, "container does not exist") + state = obj.get("State", {}) + running = bool(state.get("Running")) + if not running: + return BridgeState(True, False, False, f"container not running (status={state.get('Status')})") + started = state.get("StartedAt") or "" + since = _parse_docker_time(started) if started else time.time() + logs = docker.logs_since(container, since) + subscribed = b"subscribed voice/audio_stream" in logs + detail = "running and subscribed" if subscribed else "running but no subscribe line in logs" + return BridgeState(True, running, subscribed, detail) + + +# --------------------------------------------------------------------------- +# State persistence +# --------------------------------------------------------------------------- + + +def load_state(path: str, started_at: float) -> AlertState: + state = AlertState(started_at=started_at) + try: + with open(path) as handle: + raw = json.load(handle) + state.layer = raw.get("layer") + state.first_seen_at = raw.get("first_seen_at") + state.last_alert_at = raw.get("last_alert_at") + state.last_audio_at = raw.get("last_audio_at") + if raw.get("started_at"): + state.started_at = raw["started_at"] + except (OSError, json.JSONDecodeError): + pass + return state + + +def save_state(path: str, state: AlertState): + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + tmp = f"{path}.tmp" + with open(tmp, "w") as handle: + json.dump(dataclasses.asdict(state), handle, indent=2) + os.replace(tmp, path) + + +# --------------------------------------------------------------------------- +# Main loop +# --------------------------------------------------------------------------- + + +def build_snapshot( + cfg: Config, + watcher: MqttWatcher, + docker: DockerClient, + started_at: float, + now: Optional[float] = None, + broker_probe: Callable[[str, int, float], bool] = tcp_probe, +) -> Snapshot: + now = now if now is not None else time.time() + with watcher.lock: + connected = watcher.connected + last_audio_at = watcher.last_audio_at + presence = watcher.presence + broker_reachable = connected or broker_probe(cfg.mqtt_host, cfg.mqtt_port, 3.0) + bridge = check_bridge(docker, cfg.bridge_container) + return Snapshot( + now=now, + broker_reachable=broker_reachable, + bridge_running=bridge.running, + bridge_subscribed=bridge.subscribed, + last_audio_at=last_audio_at, + started_at=started_at, + presence=presence, + ) + + +def run(cfg: Config, cycles: Optional[int] = None, notifier: Optional[AppriseNotifier] = None) -> int: + started_at = time.time() + state = load_state(cfg.state_file, started_at) + watcher = MqttWatcher(cfg) + if state.last_audio_at is not None: + watcher.last_audio_at = state.last_audio_at # survive watchdog restarts + watcher.start() + docker = DockerClient(cfg.docker_socket, cfg.docker_timeout) + notifier = notifier or AppriseNotifier(cfg) + + LOG.info( + "watchdog starting: broker=%s:%s topic=%s silence=%ss repeat=%ss check=%ss dry_run=%s", + cfg.mqtt_host, + cfg.mqtt_port, + cfg.audio_topic, + cfg.silence_timeout, + cfg.repeat_interval, + cfg.check_interval, + cfg.dry_run, + ) + + delivered = 0 + warned_unarmed = False + try: + count = 0 + while True: + snap = build_snapshot(cfg, watcher, docker, started_at) + if snap.last_audio_at is not None: + state.last_audio_at = snap.last_audio_at + if ( + cfg.require_publisher_evidence + and not warned_unarmed + and not publisher_armed(snap) + and (snap.now - snap.started_at) > cfg.silence_timeout + ): + warned_unarmed = True + LOG.warning( + "no heartbeat or presence on %s/%s yet: silence is NOT " + "treated as a publisher failure. Flash the Directive 3 " + "firmware (heartbeat + presence) to arm this check.", + cfg.audio_topic, + cfg.status_topic, + ) + alerts = advance(state, snap, cfg) + for alert in alerts: + title, body, tag = build_message(alert, cfg) + LOG.warning( + "ALERT kind=%s layer=%s -> tag=%s", alert.kind, alert.layer, tag + ) + notifier.send(title, body, tag) + delivered += 1 + save_state(cfg.state_file, state) + count += 1 + if cycles is not None and count >= cycles: + break + time.sleep(cfg.check_interval) + finally: + watcher.stop() + save_state(cfg.state_file, state) + return delivered + + +def main(argv: Optional[list[str]] = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--cycles", type=int, default=None, help="run N checks then exit") + parser.add_argument("--once", action="store_true", help="run one check then exit") + parser.add_argument("--dry-run", action="store_true", help="print alerts instead of sending") + parser.add_argument("--log-level", default=os.environ.get("LOG_LEVEL", "INFO")) + args = parser.parse_args(argv) + + cfg = Config.from_env() + if args.dry_run: + cfg.dry_run = True + logging.basicConfig( + level=getattr(logging, args.log_level.upper(), logging.INFO), + format="%(asctime)s %(levelname)s %(message)s", + ) + cycles = 1 if args.once else args.cycles + run(cfg, cycles=cycles) + return 0 + + +if __name__ == "__main__": + sys.exit(main())