oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity
homelab-codex-ws/services/control-plane/src/supervisor.py
oskar 5f1528e4ab feat(observer): 3-state node liveness (fresh/stale/dead) + transitions + read-time net 
Fixes the "dead node shown NOMINAL" silent outage: node status was set only by
events and never expired, so a node that crashed/lost connectivity stayed
"online" forever (chelsty-infra was online for 16d, piha ~6d). The only thing
that flipped status to offline was a node_offline event, which an unreachable
node can never emit.

Now node status is derived from freshness (now - last_seen), recomputed every
observer cycle (incl. cycles with no new events):
  - always-on: fresh <=180s, stale 180-600s, dead >600s (3x the 60s heartbeat)
  - remote/LTE (chelsty-*): fresh <=900s, stale 900-3600s, dead >3600s

Thresholds + tier logic live in ONE shared helper, services/control-plane/src/
liveness.py, imported by the observer and both operator UIs (bind-mounted into
the agent-system webui image). No 3x copy.

Transitions are not silent: the observer emits node_stale / node_offline /
node_online (recovery) events tagged source=observer (skipped on re-ingest so
they never reset last_seen), routed by the supervisor to alert_only actions.

Read-time safety net: both UIs recompute liveness from last_seen at request
time, so a stalled observer still surfaces dead nodes. Services inherit their
node's liveness (cascade, variant B) without mutating services.json.

Replaces the earlier binary NODE_OFFLINE_TTL_SECS flip.

Tests: liveness unit tests, observer 3-state + transitions/recovery/baseline +
self-event skip, operator_ui read-time net + cascade, supervisor node-event
routing. 89 passed. docker compose config valid for both stacks.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 20:07:25 +02:00

828 lines
34 KiB
Python
Raw Blame History

import os
import json
import time
import logging
import yaml
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"
# 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 loop(self, interval=30):
logger.info("Starting supervisor loop")
while True:
self.reconcile()
time.sleep(interval)
if __name__ == "__main__":
supervisor = Supervisor()
supervisor.loop()
Reference in a new issue View git blame Copy permalink