homelab-codex-ws/services/control-plane/src/supervisor.py

import os
import json
import time
import logging
import yaml
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


def _atomic_write_json(path: Path, data) -> None:
    """Write JSON atomically: write to a sibling .tmp, fsync, then os.replace."""
    tmp = path.with_suffix(".tmp")
    with open(tmp, "w") as f:
        json.dump(data, f, indent=2)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)

# Constants and Paths
RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab")
WORLD_DIR = Path(RUNTIME_PATH) / "world"
ACTIONS_DIR = Path(RUNTIME_PATH) / "actions"
EVENTS_DIR = Path(RUNTIME_PATH) / "events"
REPO_ROOT = Path(os.getenv("REPO_ROOT", "/repo"))

# Node alias map: maps alternative node names (as they appear in events/world state)
# to canonical topology node names (as they appear in hosts/*/services.yaml and topology.yaml).
# Override at runtime via NODE_ALIAS_MAP env var as a JSON string, e.g.:
#   NODE_ALIAS_MAP='{"node-2": "chelsty", "node-1": "piha"}'
_NODE_ALIAS_ENV = os.getenv("NODE_ALIAS_MAP", "{}")
try:
    NODE_ALIAS_MAP = json.loads(_NODE_ALIAS_ENV)
except Exception:
    NODE_ALIAS_MAP = {}

# Event trigger types that should result in a lightweight container_restart
# rather than a full redeploy. The container is present but not running,
# or a dependency (MQTT) is unreachable — a restart is the right first step.
CONTAINER_RESTART_TRIGGERS = {"containers_not_running", "mqtt_unreachable"}

# Nodes where automatic disk_cleanup actions must NOT be generated.
# On chelsty nodes disk fullness is overwhelmingly caused by Frigate recordings
# or the HA database — Docker cleanup will not help and the operator must
# decide explicitly (e.g. adjust Frigate retain policy or purge HA recorder).
NO_DISK_CLEANUP_NODES = {"chelsty-infra", "chelsty-ha"}

# ---------------------------------------------------------------------------
# HA diagnostic event routing (ha-diag-agent events)
# ---------------------------------------------------------------------------

# ha_websocket_dead: HA WebSocket unresponsive → restart the homeassistant container.
# Separate from CONTAINER_RESTART_TRIGGERS because these events are routed directly
# from the events dir (not via the world-state drift loop) to avoid conflicts with
# the stability-agent's independent container health tracking on the same service key.
HA_CONTAINER_RESTART_EVENTS = {"ha_websocket_dead"}

# Alert-only events — operator notification, no automated action.
HA_ALERT_ONLY_EVENTS = {
    "ha_integration_failed",
    "ha_entity_unavailable_long",
    "ha_automation_failing",
    "ha_update_available",
    "ha_recorder_lag",
    "ha_system_health_degraded",
}

# Stable action-ID suffix for each alert-only type
_HA_ALERT_ID_SUFFIX = {
    "ha_integration_failed":       "integration-failed",
    "ha_entity_unavailable_long":  "entity-unavailable",
    "ha_automation_failing":       "automation-failing",
    "ha_update_available":         "update-available",
    "ha_recorder_lag":             "recorder-lag",
    "ha_system_health_degraded":   "system-health-degraded",
}

# 30-min cooldown after a container_restart completes; prevents restart loops
# when HA repeatedly fails to connect (e.g. bad config, slow startup).
HA_WEBSOCKET_RESTART_COOLDOWN = 1800

# 1-hour cooldown for alert-only events; avoids repeated Telegram noise for
# persistent conditions (e.g. an entity that stays unavailable for hours).
HA_ALERT_COOLDOWN = 3600

# Suppress ha_* events if homeassistant had a containers_not_running incident
# within this window — HA is in a planned restart/update and alerts would be noise.
HA_TRANSITION_WINDOW = 300  # 5 minutes

# ---------------------------------------------------------------------------
# Node liveness event routing (observer-emitted node_offline/node_stale/node_online)
# ---------------------------------------------------------------------------
# A node we cannot reach cannot be auto-remediated (you can't docker-restart a
# host that is offline), so these are alert-only — they exist to make a silent
# outage loud.  node_online is the recovery notice.
NODE_ALERT_EVENTS = {"node_offline", "node_stale", "node_online"}
NODE_ALERT_COOLDOWN = 3600  # 1-hour cooldown to avoid repeated Telegram noise

# When True, events that would generate container_restart are downgraded to alert_only
# with a "[SHADOW MODE]" note. Safe default for initial deployment; set
# HA_DIAG_SHADOW_MODE=false on the control-plane node when ready for live actions.
HA_DIAG_SHADOW_MODE = os.getenv("HA_DIAG_SHADOW_MODE", "true").lower() == "true"

# ---------------------------------------------------------------------------
# Loop resilience
# ---------------------------------------------------------------------------
# A single reconcile() cycle must never be able to freeze the process forever.
# reconcile() does only synchronous local filesystem I/O (open/fsync/os.replace,
# directory globs) — none of it has a language-level timeout — so the cycle is
# run in a single-worker executor and bounded by RECONCILE_TIMEOUT. If a cycle
# blocks past the timeout (e.g. a stalled fsync on the /opt/homelab bind mount,
# or an ever-growing EVENTS_DIR walk), the main loop logs it and keeps ticking
# instead of hanging silently; the stuck worker thread is abandoned in place
# (Python cannot forcibly cancel a blocked syscall) and the executor's single
# worker naturally serializes the next cycle behind it, so two cycles can never
# write the same action file concurrently.
RECONCILE_TIMEOUT = float(os.getenv("SUPERVISOR_RECONCILE_TIMEOUT", "90"))

# Every Nth cycle logs an INFO "tick" line even when nothing actionable
# happened, so silence in `docker logs` is itself a meaningful signal rather
# than being indistinguishable from a healthy, quiet loop.
TICK_LOG_EVERY = int(os.getenv("SUPERVISOR_TICK_LOG_EVERY", "10"))

# Logging setup
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("supervisor")


class Supervisor:
    def __init__(self):
        self.desired_state = {"services": {}}
        self.actual_state = {"services": {}, "nodes": {}, "incidents": {}}
        # In-memory set of already-routed HA event IDs; prevents re-processing
        # on each reconcile cycle. Grows to at most ~hundreds of entries/day.
        self._ha_processed_event_ids: set = set()
        self._ensure_dirs()
        logger.info(
            "shadow_mode=%s — HA container_restart actions %s",
            HA_DIAG_SHADOW_MODE,
            "downgraded to alert_only" if HA_DIAG_SHADOW_MODE else "enabled",
        )

    def _ensure_dirs(self):
        ACTIONS_DIR.mkdir(parents=True, exist_ok=True)
        (ACTIONS_DIR / "pending").mkdir(parents=True, exist_ok=True)

    # ------------------------------------------------------------------
    # Node name resolution
    # ------------------------------------------------------------------

    def _resolve_node(self, name):
        """Resolve an event/world-state node name to its canonical topology name."""
        return NODE_ALIAS_MAP.get(name, name)

    # ------------------------------------------------------------------
    # Container name lookup
    # ------------------------------------------------------------------

    def _get_container_name(self, service):
        """
        Determine the Docker container name for a service.
        Parses container_name from the service's docker-compose.yml.
        Falls back to the service name if not found.
        """
        compose_path = REPO_ROOT / "services" / service / "docker-compose.yml"
        if compose_path.exists():
            try:
                with open(compose_path, "r") as f:
                    compose = yaml.safe_load(f)
                for svc_block in compose.get("services", {}).values():
                    cname = svc_block.get("container_name")
                    if cname:
                        return cname
            except Exception as e:
                logger.warning(f"Could not parse docker-compose for {service}: {e}")
        # Convention: container name matches service name
        return service

    # ------------------------------------------------------------------
    # State loading
    # ------------------------------------------------------------------

    def _load_desired_state(self):
        services = {}
        hosts_dir = REPO_ROOT / "hosts"
        if not hosts_dir.exists():
            logger.warning(f"Hosts directory {hosts_dir} does not exist")
            return

        for host_dir in hosts_dir.iterdir():
            if host_dir.is_dir():
                svc_file = host_dir / "services.yaml"
                if svc_file.exists():
                    try:
                        with open(svc_file, "r") as f:
                            data = yaml.safe_load(f)
                            host_name = data.get("host")
                            for svc_name, svc_info in data.get("services", {}).items():
                                svc_info = svc_info or {}
                                # monitor: false — service is documented as desired but
                                # intentionally excluded from supervisor action generation.
                                # Use this when a service is not yet bootstrapped on an
                                # offline/LTE node so the queue stays clean until it is.
                                if svc_info.get("monitor") is False:
                                    logger.debug(
                                        f"Skipping {host_name}/{svc_name}: monitor=false"
                                    )
                                    continue
                                svc_key = f"{host_name}/{svc_name}"
                                services[svc_key] = {
                                    "node": host_name,
                                    "service": svc_name,
                                    "desired": "running"
                                }
                    except Exception as e:
                        logger.error(f"Failed to load {svc_file}: {e}")
        self.desired_state["services"] = services

    def _load_actual_state(self) -> bool:
        """Load world state from disk.  Returns False if any file is unreadable
        (empty / mid-write truncation), in which case actual_state is NOT updated
        so the caller can skip this reconcile cycle rather than treating missing
        data as a real drift signal."""
        files = {
            "services": WORLD_DIR / "services.json",
            "nodes": WORLD_DIR / "nodes.json",
            "incidents": WORLD_DIR / "incidents.json"
        }
        raw = {}
        for key, path in files.items():
            if path.exists():
                try:
                    with open(path, "r") as f:
                        raw[key] = json.load(f)
                except Exception as e:
                    logger.warning(
                        f"World state {path.name} unreadable (truncated write?): {e} "
                        f"— skipping reconcile cycle, keeping last known state"
                    )
                    return False
            else:
                raw[key] = {}

        # Normalize node names in services using alias map so that
        # event-sourced names (e.g. "node-2") resolve to canonical
        # topology names (e.g. "chelsty") before comparison with desired state.
        normalized_services = {}
        for svc_key, svc_info in raw.get("services", {}).items():
            svc_info = dict(svc_info)
            raw_node = svc_info.get("node", "")
            canonical_node = self._resolve_node(raw_node)
            if canonical_node != raw_node:
                logger.debug(f"Resolved node alias: {raw_node} → {canonical_node}")
                svc_info["node"] = canonical_node
                svc_name = svc_info.get("service") or svc_key.split("/", 1)[-1]
                svc_key = f"{canonical_node}/{svc_name}"
            normalized_services[svc_key] = svc_info

        # Normalize node names in incidents as well
        normalized_incidents = {}
        for inc_id, inc in raw.get("incidents", {}).items():
            inc = dict(inc)
            raw_node = inc.get("node", "")
            inc["node"] = self._resolve_node(raw_node)
            normalized_incidents[inc_id] = inc

        self.actual_state["services"] = normalized_services
        self.actual_state["nodes"] = raw.get("nodes", {})
        self.actual_state["incidents"] = normalized_incidents
        return True

    # ------------------------------------------------------------------
    # Incident helpers
    # ------------------------------------------------------------------

    def _get_incident_trigger(self, svc_key):
        """
        Return the trigger_type of the active incident for a service, or None.
        trigger_type is set by the observer when it creates an incident from
        a specific event type (e.g. 'containers_not_running', 'mqtt_unreachable').
        """
        svc_info = self.actual_state["services"].get(svc_key, {})
        incident_id = svc_info.get("incident_id")
        if not incident_id:
            return None
        incident = self.actual_state["incidents"].get(incident_id, {})
        if incident.get("status") == "active":
            return incident.get("trigger_type")
        return None

    # ------------------------------------------------------------------
    # Reconciliation loop
    # ------------------------------------------------------------------

    def reconcile(self):
        # Update heartbeat
        heartbeat_file = WORLD_DIR.parent / "state" / "supervisor.heartbeat"
        try:
            heartbeat_file.touch()
        except Exception as e:
            logger.error(f"Failed to touch heartbeat file: {e}")

        self._load_desired_state()
        if not self._load_actual_state():
            return  # world state unreadable this cycle — skip to avoid false drift

        drifts = []

        # 1. Check for missing or unhealthy services
        for svc_key, desired_info in self.desired_state["services"].items():
            actual_info = self.actual_state["services"].get(svc_key)

            if not actual_info:
                drifts.append({
                    "type": "missing_service",
                    "svc_key": svc_key,
                    "node": desired_info["node"],
                    "service": desired_info["service"],
                    "trigger_type": None,
                })
            elif actual_info.get("status") != "healthy":
                trigger_type = self._get_incident_trigger(svc_key)
                drifts.append({
                    "type": "unhealthy_service",
                    "svc_key": svc_key,
                    "node": desired_info["node"],
                    "service": desired_info["service"],
                    "status": actual_info.get("status"),
                    "trigger_type": trigger_type,
                })

        # 2. Generate service-level recommendations
        for drift in drifts:
            self._generate_recommendation(drift)

        # 3. Generate node-level recommendations (disk pressure)
        for node_name, node_info in self.actual_state["nodes"].items():
            if node_name in NO_DISK_CLEANUP_NODES:
                continue
            if node_info.get("disk_pressure") == "high":
                self._generate_disk_cleanup_recommendation(node_name)

        # 4. Cancel pending actions whose drift has been resolved.
        #    When a service becomes healthy again (because node-agent emits
        #    service_healthy and the observer updates services.json), any
        #    previously queued redeploy/container_restart action for that
        #    service is no longer needed. Move it to "cancelled/" so the
        #    operator can see it was auto-resolved rather than silently dropped.
        self._cancel_resolved_pending_actions()

        # 5. Route HA diagnostic events emitted by ha-diag-agent.
        #    Processed directly from the events directory — not via the world-state
        #    drift loop — to avoid conflicts with stability-agent's independent
        #    container health tracking for the homeassistant service.
        self._process_ha_events()

    # ------------------------------------------------------------------
    # Recommendation generation
    # ------------------------------------------------------------------

    def _generate_recommendation(self, drift):
        node = drift["node"]
        service = drift["service"]
        trigger_type = drift.get("trigger_type")

        # Choose action type first so we can build the stable, deterministic ID.
        # Stable IDs mean reconcile is truly idempotent: the same drift always
        # produces the same filename, so we never create duplicates even across
        # restarts of the supervisor.
        if trigger_type in CONTAINER_RESTART_TRIGGERS:
            action_id = f"container-restart-{node}-{service}"
        else:
            action_id = f"redeploy-{node}-{service}"

        # Skip if an action for this ID is already live in any active state
        # (pending → approved → running).  This prevents re-creation after
        # a human approves an action that hasn't executed yet.
        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        if trigger_type in CONTAINER_RESTART_TRIGGERS:
            # Lightweight remediation: the container exists but is not running
            # (containers_not_running) or its MQTT dependency is unreachable
            # (mqtt_unreachable). A docker restart is sufficient and low-risk.
            container_name = self._get_container_name(service)
            action = {
                "action_id": action_id,
                "timestamp": time.time(),
                "type": "container_restart",
                "node": node,
                "service": service,
                "container_name": container_name,
                "risk_level": "low",
                "confidence": 0.95,
                "description": (
                    f"Restart container '{container_name}' on {node} "
                    f"(service: {service}, reason: {trigger_type})"
                ),
                "status": "pending",
                "payload": {
                    "reason": trigger_type,
                    "svc_key": drift["svc_key"],
                },
            }
        else:
            # Full redeploy: container is running but service is broken,
            # or the cause is unknown / not a simple restart candidate.
            action = {
                "action_id": action_id,
                "timestamp": time.time(),
                "type": "redeploy",
                "node": node,
                "service": service,
                "risk_level": "guarded",
                "confidence": 0.9,
                "description": f"Redeploy {service} on {node} due to {drift['type']}",
                "status": "pending",
                "payload": {
                    "reason": drift["type"],
                    "svc_key": drift["svc_key"],
                },
            }

        action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
        try:
            _atomic_write_json(action_path, action)
            logger.info(
                f"Generated recommendation: {action_id} "
                f"(type={action['type']}, risk={action['risk_level']})"
            )
        except Exception as e:
            logger.error(f"Failed to save recommendation {action_id}: {e}")

    def _generate_disk_cleanup_recommendation(self, node: str):
        """
        Generate a disk_cleanup action when node-agent reports critical disk
        pressure (>85 %) on a node that supports automated Docker cleanup.

        This is an OPERATOR-APPROVED action (risk=guarded): it runs
        `docker image prune -a -f` and `docker volume prune -f`, which are
        more aggressive than the safe auto-cleanup the node-agent runs itself.

        Nodes in NO_DISK_CLEANUP_NODES never reach this method (filtered in
        reconcile) because their disk fullness is caused by application data
        (Frigate, HA) that the operator must handle manually.
        """
        action_id = f"disk-cleanup-{node}"

        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        action = {
            "action_id":   action_id,
            "timestamp":   time.time(),
            "type":        "disk_cleanup",
            "node":        node,
            "service":     "",
            "risk_level":  "guarded",
            "confidence":  0.85,
            "description": (
                f"Aggressive disk cleanup on {node}: docker image prune -a "
                f"and docker volume prune (requires operator approval)"
            ),
            "status": "pending",
            "payload": {
                "reason": "disk_pressure",
                "commands": [
                    "docker image prune -a -f",
                    "docker volume prune -f",
                ],
            },
        }

        action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
        try:
            _atomic_write_json(action_path, action)
            logger.info(
                f"Generated disk cleanup recommendation: {action_id} "
                f"(node={node}, risk=guarded)"
            )
        except Exception as e:
            logger.error(f"Failed to save disk cleanup recommendation {action_id}: {e}")

    def _cancel_resolved_pending_actions(self):
        """
        Auto-cancel pending service actions (redeploy / container_restart) whose
        target service is now healthy in the actual state.

        This keeps the action queue clean: when node-agent starts reporting
        service_healthy for a container that previously had no world-state entry,
        the pending 'missing_service' redeploy action that was generated before
        the first health confirmation should be removed automatically rather than
        sitting in the queue until an operator manually rejects it.

        Only pending actions are considered — approved/running actions have already
        been committed to by the operator and must not be cancelled automatically.
        """
        cancelled_dir = ACTIONS_DIR / "cancelled"
        cancelled_dir.mkdir(parents=True, exist_ok=True)
        pending_dir = ACTIONS_DIR / "pending"
        if not pending_dir.exists():
            return

        for action_file in list(pending_dir.glob("*.json")):
            try:
                with open(action_file, "r") as f:
                    action = json.load(f)
            except Exception as e:
                logger.error(f"Failed to read action {action_file.name}: {e}")
                continue

            action_type = action.get("type")
            node = action.get("node")
            service = action.get("service")

            # Only auto-cancel service-level actions (not disk_cleanup)
            if action_type not in ("redeploy", "container_restart"):
                continue
            if not node or not service:
                continue

            svc_key = f"{node}/{service}"

            cancel_reason = None

            # Case 1: service is no longer in desired state (removed from services.yaml
            # or marked monitor:false). The action was generated under old config.
            if svc_key not in self.desired_state["services"]:
                cancel_reason = "service_removed_from_desired_state"

            # Case 2: drift resolved — service is now healthy in actual state.
            elif self.actual_state["services"].get(svc_key, {}).get("status") == "healthy":
                cancel_reason = "drift_resolved_auto"

            if cancel_reason:
                dest = cancelled_dir / action_file.name
                try:
                    action["status"] = "cancelled"
                    action["cancelled_reason"] = cancel_reason
                    action["cancelled_at"] = time.time()
                    _atomic_write_json(dest, action)
                    action_file.unlink()
                    logger.info(
                        f"Auto-cancelled {action_file.name}: "
                        f"{svc_key} — {cancel_reason}"
                    )
                except Exception as e:
                    logger.error(f"Failed to cancel action {action_file.name}: {e}")

    # ------------------------------------------------------------------
    # HA diagnostic event routing
    # ------------------------------------------------------------------

    def _process_ha_events(self):
        """Scan the events directory for unprocessed ha_* events and route them."""
        if not EVENTS_DIR.exists():
            return
        for event_file in sorted(EVENTS_DIR.glob("**/*.json")):
            event_id = event_file.stem
            if event_id in self._ha_processed_event_ids:
                continue
            self._ha_processed_event_ids.add(event_id)
            try:
                with open(event_file) as f:
                    event = json.load(f)
            except Exception as e:
                logger.debug(f"Could not read event {event_file}: {e}")
                continue
            etype = event.get("type", "")
            if etype in NODE_ALERT_EVENTS:
                self._route_node_event(event)
                continue
            if not etype.startswith("ha_"):
                continue
            self._route_ha_event(event)

    def _route_ha_event(self, event: dict):
        event_type = event.get("type", "")
        node = event.get("node", "")
        if not node:
            return

        if event_type in HA_CONTAINER_RESTART_EVENTS:
            if self._is_ha_in_transition(node):
                logger.debug(
                    f"Suppressing {event_type} on {node}: homeassistant in transition"
                )
                return
            if HA_DIAG_SHADOW_MODE:
                logger.info(
                    "shadow_mode: suppressed container_restart for %s", event_type
                )
                self._generate_ha_shadow_alert(node, event)
            else:
                self._generate_ha_container_restart(node, event)

        elif event_type == "ha_websocket_recovered":
            self._cancel_ha_container_restart(node)

        elif event_type in HA_ALERT_ONLY_EVENTS:
            if self._is_ha_in_transition(node):
                logger.debug(
                    f"Suppressing {event_type} on {node}: homeassistant in transition"
                )
                return
            self._generate_ha_alert_only(node, event)

    def _is_ha_in_transition(self, node: str) -> bool:
        """Return True if homeassistant container had a recent containers_not_running incident.

        Suppresses ha_* alerts during planned HA restarts/updates to avoid
        flooding the operator with secondary diagnostic alerts.
        """
        svc_key = f"{node}/homeassistant"
        svc_info = self.actual_state["services"].get(svc_key, {})
        incident_id = svc_info.get("incident_id")
        if not incident_id:
            return False
        incident = self.actual_state["incidents"].get(incident_id, {})
        return (
            incident.get("status") == "active"
            and incident.get("trigger_type") == "containers_not_running"
            and time.time() - (incident.get("last_occurrence") or 0) < HA_TRANSITION_WINDOW
        )

    def _ha_action_recently_completed(self, action_id: str, cooldown: int) -> bool:
        """Return True if action completed/rejected/cancelled within the cooldown window."""
        for state in ("completed", "rejected", "cancelled"):
            path = ACTIONS_DIR / state / f"{action_id}.json"
            if path.exists():
                try:
                    with open(path) as f:
                        data = json.load(f)
                    finished = (
                        data.get("finished_at")
                        or data.get("cancelled_at")
                        or data.get("updated_at")
                        or 0
                    )
                    if time.time() - finished < cooldown:
                        return True
                except Exception:
                    pass
        return False

    def _generate_ha_container_restart(self, node: str, event: dict):
        service = "homeassistant"
        action_id = f"container-restart-{node}-{service}"

        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        if self._ha_action_recently_completed(action_id, HA_WEBSOCKET_RESTART_COOLDOWN):
            logger.debug(
                f"Skipping {action_id}: within {HA_WEBSOCKET_RESTART_COOLDOWN}s cooldown"
            )
            return

        payload = dict(event.get("payload", {}))
        payload["reason"] = "ha_websocket_dead"
        payload["svc_key"] = f"{node}/{service}"

        container_name = self._get_container_name(service)
        action = {
            "action_id": action_id,
            "timestamp": time.time(),
            "type": "container_restart",
            "node": node,
            "service": service,
            "container_name": container_name,
            "risk_level": "low",
            "confidence": 0.9,
            "description": (
                f"Restart '{container_name}' on {node}: HA WebSocket unresponsive"
            ),
            "status": "pending",
            "payload": payload,
        }
        self._write_pending_action(action)

    def _generate_ha_shadow_alert(self, node: str, event: dict):
        """Shadow-mode downgrade: emit alert_only instead of container_restart.

        Uses the same action_id and cooldown as the real restart so that
        cooldown semantics are identical regardless of shadow mode state.
        """
        service = "homeassistant"
        action_id = f"container-restart-{node}-{service}"

        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        if self._ha_action_recently_completed(action_id, HA_WEBSOCKET_RESTART_COOLDOWN):
            logger.debug(
                f"Skipping {action_id}: within {HA_WEBSOCKET_RESTART_COOLDOWN}s cooldown"
            )
            return

        payload = dict(event.get("payload", {}))
        payload["reason"] = "ha_websocket_dead"
        payload["svc_key"] = f"{node}/{service}"
        payload["shadow_mode"] = True

        action = {
            "action_id": action_id,
            "timestamp": time.time(),
            "type": "alert_only",
            "node": node,
            "service": service,
            "risk_level": "info",
            "confidence": 0.9,
            "description": (
                f"[SHADOW MODE] would have triggered container_restart "
                f"for {service} on {node}: HA WebSocket unresponsive"
            ),
            "status": "pending",
            "payload": payload,
        }
        self._write_pending_action(action)

    def _generate_ha_alert_only(self, node: str, event: dict):
        event_type = event.get("type", "")
        suffix = _HA_ALERT_ID_SUFFIX.get(event_type, event_type.replace("_", "-"))
        action_id = f"alert-ha-{suffix}-{node}"

        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        if self._ha_action_recently_completed(action_id, HA_ALERT_COOLDOWN):
            logger.debug(
                f"Skipping {action_id}: within {HA_ALERT_COOLDOWN}s cooldown"
            )
            return

        payload = dict(event.get("payload", {}))
        payload["reason"] = event_type

        action = {
            "action_id": action_id,
            "timestamp": time.time(),
            "type": "alert_only",
            "node": node,
            "service": event.get("service", "homeassistant"),
            "risk_level": "info",
            "confidence": 1.0,
            "description": event.get(
                "message", f"HA diagnostic alert: {event_type} on {node}"
            ),
            "status": "pending",
            "payload": payload,
        }
        self._write_pending_action(action)

    def _route_node_event(self, event: dict):
        """Route an observer-emitted node liveness event to an alert_only action.

        node_offline / node_stale make a silent outage loud; node_online is the
        recovery notice.  No auto-remediation — an unreachable node can't be
        restarted from here.  Dedup via stable action_id + cooldown, mirroring
        the HA alert path.
        """
        event_type = event.get("type", "")
        node = event.get("node") or event.get("payload", {}).get("affected_node")
        if not node:
            return

        action_id = f"alert-{event_type.replace('_', '-')}-{node}"

        for state in ("pending", "approved", "running"):
            if (ACTIONS_DIR / state / f"{action_id}.json").exists():
                logger.debug(f"Skipping {action_id}: already in state '{state}'")
                return

        if self._ha_action_recently_completed(action_id, NODE_ALERT_COOLDOWN):
            logger.debug(f"Skipping {action_id}: within {NODE_ALERT_COOLDOWN}s cooldown")
            return

        payload = dict(event.get("payload", {}))
        payload["reason"] = event_type

        action = {
            "action_id": action_id,
            "timestamp": time.time(),
            "type": "alert_only",
            "node": node,
            "service": None,
            "risk_level": "info",
            "confidence": 1.0,
            "description": event.get(
                "message", f"Node liveness alert: {event_type} on {node}"
            ),
            "status": "pending",
            "payload": payload,
        }
        self._write_pending_action(action)

    def _cancel_ha_container_restart(self, node: str):
        """Move a pending ha_websocket_dead container_restart to cancelled on recovery."""
        action_id = f"container-restart-{node}-homeassistant"
        pending_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
        if not pending_path.exists():
            return
        cancelled_dir = ACTIONS_DIR / "cancelled"
        cancelled_dir.mkdir(parents=True, exist_ok=True)
        dest = cancelled_dir / f"{action_id}.json"
        try:
            with open(pending_path) as f:
                action = json.load(f)
            action["status"] = "cancelled"
            action["cancelled_reason"] = "ha_websocket_recovered"
            action["cancelled_at"] = time.time()
            _atomic_write_json(dest, action)
            pending_path.unlink()
            logger.info(f"Cancelled {action_id}: ha_websocket_recovered on {node}")
        except Exception as e:
            logger.error(f"Failed to cancel {action_id}: {e}")

    def _write_pending_action(self, action: dict):
        action_id = action["action_id"]
        action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
        try:
            _atomic_write_json(action_path, action)
            logger.info(
                f"Generated HA action: {action_id} "
                f"(type={action['type']}, risk={action['risk_level']})"
            )
        except Exception as e:
            logger.error(f"Failed to save action {action_id}: {e}")

    def _run_cycle_safely(self):
        """Run one reconcile() cycle, never letting an exception escape.

        An uncaught exception here would previously propagate out of loop()
        and kill the process outright — a crash is at least visible (the
        container exits and restart:unless-stopped brings it back). This
        makes that failure mode explicit and non-fatal: log the full
        traceback and let the loop continue on the next cycle.
        """
        try:
            self.reconcile()
        except Exception:
            logger.exception(
                "reconcile cycle raised an unhandled exception — logging and "
                "continuing to the next cycle"
            )

    def loop(self, interval=30, max_cycles=None, reconcile_timeout=None):
        """Run reconcile() every `interval` seconds, forever.

        max_cycles: stop after N cycles instead of looping forever (tests only).
        reconcile_timeout: override RECONCILE_TIMEOUT (tests only).
        """
        logger.info("Starting supervisor loop")
        timeout = RECONCILE_TIMEOUT if reconcile_timeout is None else reconcile_timeout
        cycle = 0
        # max_workers=1 serializes cycles: if one is abandoned after a timeout,
        # the next submit() queues behind it rather than running concurrently
        # and racing on the same action files.
        executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="supervisor-reconcile")
        try:
            while max_cycles is None or cycle < max_cycles:
                cycle += 1
                future = executor.submit(self._run_cycle_safely)
                try:
                    future.result(timeout=timeout)
                except TimeoutError:
                    logger.error(
                        "reconcile cycle #%d did not complete within %ss — "
                        "likely blocked on I/O (fsync/glob/file open) with no "
                        "language-level timeout. Abandoning this cycle; the loop "
                        "continues. The stuck worker thread keeps running in the "
                        "background and the next cycle will queue behind it.",
                        cycle, timeout,
                    )
                if cycle % TICK_LOG_EVERY == 0:
                    logger.info("tick: supervisor loop alive, cycle #%d", cycle)
                time.sleep(interval)
        finally:
            executor.shutdown(wait=False, cancel_futures=True)


if __name__ == "__main__":
    supervisor = Supervisor()
    supervisor.loop()