fix(events): service_healthy tylko na transition + cleanup 358k backlog + retencja — glob zalewu paralizowal reconcile supervisora

- node-agent: service_healthy emitowany tylko przy przejsciu w stan zdrowy
  (per-service in-memory state), nie co cykl dla kazdego zdrowego serwisu.
  To samo dla control-plane HTTP probe. healthcheck_failed/containers_not_running/
  incydenty pozostaja emitowane bez zmian (realne sygnaly).
- node-agent: naprawiono _cleanup_control_plane_fs — istniejaca retencja
  eventow byla martwa od migracji observer_checkpoint.json na epoch-int
  (str(path) <= int rzucal TypeError, lapane cicho przez broad except).
  Teraz porownuje epoch-do-epoch i czysci wylacznie service_healthy/node_health
  starsze niz checkpoint + 3-dniowy bufor; healthcheck_failed/incydenty/ha_*
  zachowane bezterminowo.
- scripts/maintenance/cleanup_event_backlog.py: jednorazowy skrypt do
  bezpiecznego czyszczenia backlogu na VPS (dry-run domyslnie, --apply
  do usuniecia). Ten sam warunek: typ szumu + starsze niz checkpoint + 1h bufor.

scripts/maintenance/cleanup_event_backlog.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""
+cleanup_event_backlog.py — safe backlog cleanup for /opt/homelab/events/.
+
+Removes ONLY processed noise events (service_healthy, node_health by default)
+that are both:
+  1. Strictly older than the observer's per-node checkpoint timestamp
+     (observer_checkpoint.json) — i.e. guaranteed already processed into
+     world_state, per the same ordering guarantee observer.py itself relies on
+     (oldest-first, monotonically-advancing checkpoint).
+  2. Older than a minimum safety age (default 3600s / 1h) — defense in depth
+     against clock skew or a checkpoint write racing this script, mirroring
+     the dual-condition pattern already used in node_agent.py's
+     _cleanup_control_plane_fs.
+
+Everything else — healthcheck_failed, containers_not_running, incidents,
+ha_* events, node liveness transitions, disk/memory/cpu pressure, and any
+event newer than the checkpoint — is never touched.
+
+Defaults to DRY RUN: prints counts per node/type and does not delete
+anything. Pass --apply to actually delete.
+
+Usage:
+    python3 cleanup_event_backlog.py                          # dry run, defaults
+    python3 cleanup_event_backlog.py --apply                   # actually delete
+    python3 cleanup_event_backlog.py --types service_healthy,node_health
+    python3 cleanup_event_backlog.py --min-age-secs 86400       # extra-cautious
+"""
+import argparse
+import json
+import os
+import re
+import sys
+import time
+from pathlib import Path
+
+RUNTIME_PATH = Path(os.getenv("RUNTIME_PATH", "/opt/homelab"))
+EVENTS_DIR = RUNTIME_PATH / "events"
+CHECKPOINT_FILE = RUNTIME_PATH / "state" / "observer_checkpoint.json"
+
+# Matches observer.py's _EVENT_TS_RE: evt-<node>-<unixts>-<type>-<svc>.json
+_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")
+
+DEFAULT_TYPES = {"service_healthy", "node_health"}
+
+
+def _ts_from_name(name: str):
+    m = _EVENT_TS_RE.search(Path(name).stem)
+    return int(m.group(1)) if m else None
+
+
+def _type_from_name(name: str):
+    """Best-effort event type extraction from evt-<node>-<ts>-<type>-<svc>.json.
+
+    Mirrors the informal convention node-agent/observer use when building
+    filenames; falls back to 'unknown' rather than guessing wrong.
+    """
+    stem = Path(name).stem
+    m = re.match(r"evt-.+?-\d{9,11}-(.+)$", stem)
+    if not m:
+        return "unknown"
+    rest = m.group(1)
+    # rest is "<type>-<svc_slug>"; type itself may contain underscores but not
+    # dashes in this codebase's event vocabulary, so split on first dash after
+    # the known type token is ambiguous — instead check against DEFAULT_TYPES
+    # and a couple of other well-known multi-word-free types directly.
+    for t in KNOWN_TYPES:
+        if rest == t or rest.startswith(t + "-"):
+            return t
+    return rest.split("-")[0]
+
+
+# Every event type this repo's emitters are known to produce (node_agent.py,
+# ha_diag/event_emitter.py, observer.py's own _emit_node_transition), longest
+# first so startswith() matching in _type_from_name doesn't shadow a longer
+# type with a shorter prefix (e.g. "node_health" vs "node_offline" share no
+# prefix, but keep this list explicit rather than relying on luck).
+KNOWN_TYPES = sorted(
+    {
+        "service_healthy", "service_unhealthy", "service_recovered",
+        "node_health", "node_online", "node_offline", "node_stale",
+        "healthcheck_failed", "containers_not_running", "container_restarting",
+        "container_state_unexpected", "disk_pressure", "high_memory", "high_cpu",
+        "mqtt_unreachable",
+        "ha_websocket_dead", "ha_websocket_recovered", "ha_integration_failed",
+        "ha_entity_unavailable_long", "ha_automation_failing", "ha_update_available",
+        "ha_recorder_lag", "ha_system_health_degraded",
+        "deployment_started", "deployment_completed", "deployment_failed",
+        "remediation_started", "remediation_completed",
+    },
+    key=len, reverse=True,
+)
+
+
+def load_checkpoints(checkpoint_file: Path = CHECKPOINT_FILE):
+    if not checkpoint_file.exists():
+        print(f"WARNING: no checkpoint file at {checkpoint_file} — refusing to "
+              f"guess; nothing is safe to delete without it.", file=sys.stderr)
+        return {}
+    data = json.loads(checkpoint_file.read_text())
+    raw = data.get("node_checkpoints", {})
+    out = {}
+    for node, val in raw.items():
+        if isinstance(val, bool):
+            continue
+        if isinstance(val, (int, float)):
+            out[node] = int(val)
+    return out
+
+
+def scan(types_to_clean, min_age_secs, checkpoints, now, events_dir: Path = EVENTS_DIR):
+    """Return (candidates: list[Path], counts: dict[node][type] -> (delete, keep))."""
+    candidates = []
+    counts = {}
+    if not events_dir.exists():
+        return candidates, counts
+
+    for node_dir in sorted(p for p in events_dir.iterdir() if p.is_dir()):
+        node = node_dir.name
+        checkpoint = checkpoints.get(node)
+        counts.setdefault(node, {})
+        for f in node_dir.glob("*.json"):
+            etype = _type_from_name(f.name)
+            ts = _ts_from_name(f.name)
+            if ts is None:
+                try:
+                    ts = int(f.stat().st_mtime)
+                except OSError:
+                    continue
+
+            bucket = counts[node].setdefault(etype, {"delete": 0, "keep": 0})
+
+            if etype not in types_to_clean:
+                bucket["keep"] += 1
+                continue
+            if checkpoint is None:
+                # No checkpoint for this node dir at all (e.g. a ghost/foreign
+                # dir) — never guess; keep everything.
+                bucket["keep"] += 1
+                continue
+            if ts >= checkpoint:
+                # Not guaranteed processed yet — never delete.
+                bucket["keep"] += 1
+                continue
+            if now - ts < min_age_secs:
+                # Within the safety buffer even though past checkpoint.
+                bucket["keep"] += 1
+                continue
+
+            bucket["delete"] += 1
+            candidates.append(f)
+
+    return candidates, counts
+
+
+def print_report(counts, min_age_secs, checkpoints):
+    total_delete = 0
+    total_keep = 0
+    for node in sorted(counts):
+        cp = checkpoints.get(node)
+        cp_str = f"checkpoint={cp}" if cp is not None else "NO CHECKPOINT (nothing eligible)"
+        print(f"\n=== {node} ({cp_str}) ===")
+        for etype, c in sorted(counts[node].items(), key=lambda kv: -sum(kv[1].values())):
+            if c["delete"] or c["keep"]:
+                print(f"  {etype:32s} delete={c['delete']:>7d}  keep={c['keep']:>7d}")
+            total_delete += c["delete"]
+            total_keep += c["keep"]
+    print(f"\nTOTAL: delete={total_delete}  keep={total_keep}")
+    print(f"(min_age_secs={min_age_secs}, i.e. only files older than checkpoint AND "
+          f"more than {min_age_secs}s old are ever deleted)")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--apply", action="store_true",
+                     help="Actually delete. Without this flag, dry-run only.")
+    ap.add_argument("--types", default=",".join(sorted(DEFAULT_TYPES)),
+                     help=f"Comma-separated event types to clean (default: {DEFAULT_TYPES})")
+    ap.add_argument("--min-age-secs", type=int, default=3600,
+                     help="Extra safety buffer past the checkpoint (default 3600s = 1h)")
+    args = ap.parse_args()
+
+    types_to_clean = {t.strip() for t in args.types.split(",") if t.strip()}
+    checkpoints = load_checkpoints(CHECKPOINT_FILE)
+    now = int(time.time())
+
+    candidates, counts = scan(types_to_clean, args.min_age_secs, checkpoints, now, EVENTS_DIR)
+    print_report(counts, args.min_age_secs, checkpoints)
+
+    if not args.apply:
+        print(f"\nDRY RUN — no files deleted. Re-run with --apply to delete "
+              f"the {len(candidates)} files listed above as 'delete'.")
+        return
+
+    deleted = 0
+    for f in candidates:
+        try:
+            f.unlink(missing_ok=True)
+            deleted += 1
+        except Exception as exc:
+            print(f"Failed to remove {f}: {exc}", file=sys.stderr)
+    print(f"\nDeleted {deleted}/{len(candidates)} files.")
+
+
+if __name__ == "__main__":
+    main()

scripts/maintenance/tests/test_cleanup_event_backlog.py

@@ -0,0 +1,111 @@
+"""Tests for cleanup_event_backlog.py — the event-flood backlog cleanup script.
+
+Covers the safety invariant the whole cleanup depends on: only events that are
+(a) a noise type, (b) strictly older than the observer's per-node checkpoint,
+and (c) past the extra min-age safety buffer are ever deleted. Everything
+else — unprocessed events, real signals, foreign/ghost node dirs — survives.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import cleanup_event_backlog as cleanup  # noqa: E402
+
+
+def _write(node_dir: Path, ts: int, etype: str, svc: str = "svc") -> Path:
+    p = node_dir / f"evt-{node_dir.name}-{ts}-{etype}-{svc}.json"
+    p.write_text("{}")
+    return p
+
+
+def test_deletes_only_processed_noise_past_buffer(tmp_path):
+    events_dir = tmp_path / "events"
+    node_dir = events_dir / "piha"
+    node_dir.mkdir(parents=True)
+    now = int(time.time())
+    checkpoint = now - 100
+
+    old_healthy = _write(node_dir, now - 7200, "service_healthy")
+    old_node_health = _write(node_dir, now - 7200, "node_health")
+    recent_healthy = _write(node_dir, now - 30, "service_healthy", svc="svc2")  # within buffer
+    unprocessed = _write(node_dir, now + 50, "service_healthy", svc="svc3")     # after checkpoint
+    real_signal = _write(node_dir, now - 999999, "healthcheck_failed")
+
+    checkpoints = {"piha": checkpoint}
+    candidates, counts = cleanup.scan(
+        {"service_healthy", "node_health"}, 3600, checkpoints, now, events_dir
+    )
+
+    assert set(candidates) == {old_healthy, old_node_health}
+    for f in (recent_healthy, unprocessed, real_signal):
+        assert f.exists()
+    assert counts["piha"]["healthcheck_failed"]["delete"] == 0
+    assert counts["piha"]["healthcheck_failed"]["keep"] == 1
+
+
+def test_never_deletes_without_checkpoint_for_node(tmp_path):
+    """A node dir with no checkpoint entry (ghost/foreign dir) must be left alone."""
+    events_dir = tmp_path / "events"
+    node_dir = events_dir / "be17cb6eb0f6"
+    node_dir.mkdir(parents=True)
+    now = int(time.time())
+    _write(node_dir, now - 999999, "service_healthy")
+
+    candidates, counts = cleanup.scan(
+        {"service_healthy", "node_health"}, 3600, {}, now, events_dir
+    )
+    assert candidates == []
+    assert counts["be17cb6eb0f6"]["service_healthy"]["keep"] == 1
+
+
+def test_never_deletes_incidents_or_unknown_types(tmp_path):
+    events_dir = tmp_path / "events"
+    node_dir = events_dir / "vps"
+    node_dir.mkdir(parents=True)
+    now = int(time.time())
+    checkpoint = now - 100
+    ha_event = _write(node_dir, now - 999999, "ha_entity_unavailable_long")
+    incident_like = _write(node_dir, now - 999999, "containers_not_running")
+
+    candidates, _counts = cleanup.scan(
+        {"service_healthy", "node_health"}, 3600, {"vps": checkpoint}, now, events_dir
+    )
+    assert candidates == []
+    assert ha_event.exists()
+    assert incident_like.exists()
+
+
+def test_apply_actually_deletes_and_dry_run_does_not(tmp_path, capsys, monkeypatch):
+    events_dir = tmp_path / "events"
+    state_dir = tmp_path / "state"
+    node_dir = events_dir / "piha"
+    node_dir.mkdir(parents=True)
+    state_dir.mkdir(parents=True)
+    now = int(time.time())
+    checkpoint = now - 100
+    old_healthy = _write(node_dir, now - 7200, "service_healthy")
+    (state_dir / "observer_checkpoint.json").write_text(
+        json.dumps({"node_checkpoints": {"piha": checkpoint}})
+    )
+
+    monkeypatch.setattr(cleanup, "EVENTS_DIR", events_dir)
+    monkeypatch.setattr(cleanup, "CHECKPOINT_FILE", state_dir / "observer_checkpoint.json")
+    monkeypatch.setattr(sys, "argv", ["cleanup_event_backlog.py"])
+    cleanup.main()
+    assert old_healthy.exists()  # dry run: untouched
+
+    monkeypatch.setattr(sys, "argv", ["cleanup_event_backlog.py", "--apply"])
+    cleanup.main()
+    assert not old_healthy.exists()  # applied: deleted
+
+
+def test_type_extraction_matches_node_agent_filename_convention():
+    assert cleanup._type_from_name("evt-piha-1784215978-service_healthy-mosquitto.json") == "service_healthy"
+    assert cleanup._type_from_name("evt-vps-1784215978-node_health-node.json") == "node_health"
+    assert cleanup._type_from_name("evt-piha-1784215978-healthcheck_failed-z2m.json") == "healthcheck_failed"
+    assert cleanup._type_from_name("evt-piha-1784215978-ha_entity_unavailable_long-node.json") == \
+        "ha_entity_unavailable_long"

services/node-agent/src/node_agent.py

@@ -128,6 +128,66 @@ def _utc_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+# Matches evt-<node>-<unixts>-<type>-<svc>.json (this module's own emit_event
+# naming, mirrored by observer.py's _EVENT_TS_RE / _ts_from_event_name so both
+# sides of the checkpoint agree on how to read a filename's embedded epoch).
+_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")
+_EVENT_TYPE_RE = re.compile(r"^evt-.+?-\d{9,11}-(.+)$")
+
+
+def _event_ts_from_filename(name: str):
+    """Return the embedded <unixts> from an event filename, or None if absent."""
+    m = _EVENT_TS_RE.search(Path(name).stem)
+    return int(m.group(1)) if m else None
+
+
+def _event_type_from_filename(name: str) -> str:
+    """Best-effort event type from evt-<node>-<ts>-<type>-<svc>.json.
+
+    Used only to decide retention eligibility (never for dispatch), so an
+    imprecise match on an unrecognized type is safe: it simply keeps the file.
+    """
+    stem = Path(name).stem
+    m = _EVENT_TYPE_RE.match(stem)
+    if not m:
+        return "unknown"
+    rest = m.group(1)
+    for t in _RETENTION_NOISE_TYPES:
+        if rest == t or rest.startswith(t + "-"):
+            return t
+    return rest
+
+
+def _checkpoint_ts_from_value(value) -> int:
+    """Coerce a stored observer_checkpoint.json value into an int epoch.
+
+    Mirrors observer.py's own _checkpoint_ts_from_value: current format is an
+    int/float epoch; older observer builds stored a lexical path string, from
+    which the embedded <unixts> is extracted. Unparseable/absent -> 0, which
+    compares as "nothing processed yet" and safely keeps every file for that
+    node rather than guessing a checkpoint too high and deleting unprocessed
+    events.
+    """
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, (int, float)):
+        return int(value)
+    if isinstance(value, str) and value:
+        ts = _event_ts_from_filename(value)
+        if ts is not None:
+            return ts
+    return 0
+
+
+# Event types eligible for the ongoing retention sweep in
+# _cleanup_control_plane_fs: pure positive-confirmation noise, never the
+# actionable signals (healthcheck_failed, containers_not_running, ha_*,
+# incidents, disk/memory/cpu pressure, deployment/remediation records, node
+# liveness transitions), which are kept indefinitely by this method — an
+# operator or the panel event feed may need to look back at those.
+_RETENTION_NOISE_TYPES = frozenset({"service_healthy", "node_health"})
+
+
 # ---------------------------------------------------------------------------
 # NodeAgent
 # ---------------------------------------------------------------------------
@@ -138,6 +198,16 @@ class NodeAgent:
         self.node_type = _resolve_node_type()
         self._ensure_dirs()
 
+        # Tracks the last known health classification per service (by canonical
+        # name — container service names AND the synthetic "control-plane" key
+        # used by _check_control_plane_health share this dict) so service_healthy
+        # is emitted only on the unhealthy->healthy transition, not once per cycle
+        # for a service that is already healthy. In-memory only: reset on process
+        # restart is acceptable — it just re-emits one confirmation event, not a
+        # flood, and the observer treats that re-confirmation as idempotent (see
+        # process_event).
+        self._service_health_state: dict = {}
+
         self.docker_client = None
         if DOCKER_AVAILABLE:
             try:
@@ -361,6 +431,7 @@ class NodeAgent:
 
                 # Exited container that carries an auto-restart policy
                 if status in ("exited", "dead"):
+                    self._service_health_state[name] = False
                     logger.warning(f"Container exited: {name} (restart={restart_policy})")
                     self.emit_event(
                         "containers_not_running", "high", name,
@@ -378,6 +449,7 @@ class NodeAgent:
                 # alarm, while a real crash-loop escalates to the same actionable
                 # signal as an exited container.
                 elif status == "restarting":
+                    self._service_health_state[name] = False
                     if restart_count >= CRASH_LOOP_RESTART_THRESHOLD:
                         # Genuine crash-loop: reuse containers_not_running so it
                         # rides the existing, supervisor-wired remediation path
@@ -414,6 +486,7 @@ class NodeAgent:
 
                 # Running container with a failing health check
                 elif status == "running" and health_status == "unhealthy":
+                    self._service_health_state[name] = False
                     logger.warning(f"Container unhealthy: {name}")
                     self.emit_event(
                         "healthcheck_failed", "high", name,
@@ -425,18 +498,31 @@ class NodeAgent:
                 # services.json stays populated for the supervisor's drift detection.
                 # Without this, the supervisor sees services.json as empty and treats
                 # all desired services as "missing", flooding the action queue.
+                #
+                # Emitted ONLY on the unhealthy/unknown -> healthy transition, not
+                # every cycle: a healthy service otherwise re-confirms itself once
+                # per health-check interval forever, which is exactly the flood
+                # that paralyzed the supervisor's reconcile() glob (358k backlog
+                # files, 91% service_healthy). The positive "still healthy" state is
+                # already carried by the presence of a `healthy` services.json entry
+                # (written once here) plus the node_health heartbeat every cycle —
+                # so a stuck/frozen node_agent is still visible via node liveness,
+                # without needing a fresh service_healthy file per service per cycle.
                 elif status == "running":
+                    if self._service_health_state.get(name) is not True:
                         self.emit_event(
                             "service_healthy", "info", name,
                             f"Container '{name}' is running",
                             {**base_payload, "health_status": health_status or "none"},
                         )
+                        self._service_health_state[name] = True
 
                 # Deliberately paused. docker pause is always an operator/tool
                 # action (Docker never auto-pauses), so it is not a fault — but a
                 # managed service should not normally sit paused, so surface it
                 # observationally rather than swallowing it. Non-actionable.
                 elif status == "paused":
+                    self._service_health_state[name] = False
                     logger.info(f"Container paused: {name}")
                     self.emit_event(
                         "container_state_unexpected", "medium", name,
@@ -456,6 +542,7 @@ class NodeAgent:
                 # silently: emit a diagnostic so a new/unknown state becomes visible
                 # instead of quietly recreating the very blind spot this fix closes.
                 else:
+                    self._service_health_state[name] = False
                     logger.warning(
                         f"Container in unhandled Docker state: {name} (state='{status}')"
                     )
@@ -587,23 +674,42 @@ class NodeAgent:
                 except Exception as exc:
                     logger.error(f"Failed to remove {f}: {exc}")
 
-        # 3. Event files older than 3 days AND already past observer checkpoint.
+        # 3. Noise event files (service_healthy / node_health) that are BOTH
-        #    The dual condition guarantees we never delete an unprocessed event.
+        #    already past the observer's per-node checkpoint (guaranteed
-        #    Checkpoint format: {"node_checkpoints": {"piha": "/path/last", ...}}
+        #    processed into world_state — see _checkpoint_ts_from_value) AND
+        #    older than 3 days (safety buffer past the checkpoint). The dual
+        #    condition guarantees we never delete an unprocessed event, and
+        #    the type restriction guarantees healthcheck_failed / incidents /
+        #    ha_* / node-liveness events are kept indefinitely by this method.
+        #
+        #    BUG FIX (event-flood cleanup): this previously compared str(f)
+        #    (a full file path) against the checkpoint value with <=, which
+        #    only ever worked when the checkpoint stored the pre-fix lexical
+        #    path format. Since observer.py migrated node_checkpoints to int
+        #    epoch timestamps, every comparison here raised
+        #    "'<=' not supported between instances of 'str' and 'int'",
+        #    silently caught by the broad except below — so this cleanup has
+        #    been a no-op ever since, which is the direct cause of the 358k-file
+        #    backlog that paralyzed the supervisor's reconcile(). Fixed by
+        #    comparing epoch-to-epoch, same as observer.py's own checkpoint
+        #    logic.
         checkpoint_file = STATE_DIR / "observer_checkpoint.json"
         node_checkpoints: dict = {}
         if checkpoint_file.exists():
             try:
                 cp = json.loads(checkpoint_file.read_text())
                 if "node_checkpoints" in cp:
-                    node_checkpoints = cp["node_checkpoints"]
+                    node_checkpoints = {
+                        node: _checkpoint_ts_from_value(val)
+                        for node, val in (cp["node_checkpoints"] or {}).items()
+                    }
                 elif "last_processed_file" in cp:
                     # Migrate old single-file format
                     old = cp.get("last_processed_file", "")
                     if old:
                         try:
                             node_dir = Path(old).relative_to(EVENTS_DIR).parts[0]
-                            node_checkpoints = {node_dir: old}
+                            node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)}
                         except Exception:
                             pass
             except Exception as exc:
@@ -612,13 +718,20 @@ class NodeAgent:
         if node_checkpoints:
             for f in EVENTS_DIR.glob("**/*.json"):
                 try:
+                    etype = _event_type_from_filename(f.name)
+                    if etype not in _RETENTION_NOISE_TYPES:
+                        continue
                     # Determine which node directory this event belongs to
                     rel = Path(f).relative_to(EVENTS_DIR)
                     node_dir = str(rel.parts[0]) if rel.parts else "__unknown__"
-                    last_for_node = node_checkpoints.get(node_dir, "")
+                    checkpoint_ts = node_checkpoints.get(node_dir)
-                    if (last_for_node
+                    if not checkpoint_ts:
-                            and now - f.stat().st_mtime > three_days
+                        continue
-                            and str(f) <= last_for_node):
+                    event_ts = _event_ts_from_filename(f.name)
+                    if event_ts is None:
+                        event_ts = int(f.stat().st_mtime)
+                    if (event_ts < checkpoint_ts
+                            and now - f.stat().st_mtime > three_days):
                         f.unlink(missing_ok=True)
                         logger.info(f"Cleaned old event: {f.name}")
                 except Exception as exc:
@@ -711,18 +824,25 @@ class NodeAgent:
         try:
             resp = urllib.request.urlopen(endpoint, timeout=5)
             if resp.status == 200:
+                # Transition-only, same rationale as check_containers(): without
+                # this gate, control-plane re-confirmed itself every cycle
+                # forever, contributing to the same service_healthy flood.
+                if self._service_health_state.get("control-plane") is not True:
                     self.emit_event(
                         "service_healthy", "info", "control-plane",
                         "Control-plane HTTP endpoint is reachable",
                         {"endpoint": endpoint},
                     )
+                    self._service_health_state["control-plane"] = True
             else:
+                self._service_health_state["control-plane"] = False
                 self.emit_event(
                     "service_unhealthy", "high", "control-plane",
                     f"Control-plane HTTP endpoint returned HTTP {resp.status}",
                     {"endpoint": endpoint, "http_status": resp.status},
                 )
         except Exception as exc:
+            self._service_health_state["control-plane"] = False
             self.emit_event(
                 "service_unhealthy", "high", "control-plane",
                 f"Control-plane HTTP endpoint unreachable: {exc}",

services/node-agent/tests/test_check_containers.py

@@ -127,6 +127,59 @@ def test_running_healthy_emits_service_healthy(agent, monkeypatch):
     assert events[0][1] == "info"
 
 
+def test_running_healthy_second_cycle_is_silent(agent, monkeypatch):
+    """Regression for the event-flood fix: a service that stays healthy across
+    cycles must NOT re-emit service_healthy every cycle — only on the
+    transition into healthy."""
+    c = make_container("running")
+    run_check(agent, [c], monkeypatch)  # first cycle: transition, emits once
+    events = run_check(agent, [c], monkeypatch)  # second cycle: still healthy
+    assert events == []
+
+
+def test_running_healthy_after_unhealthy_reemits_service_healthy(agent, monkeypatch):
+    """Recovery (unhealthy -> healthy) must still be visible exactly once."""
+    unhealthy = make_container("running", health="unhealthy", name="svc")
+    events = run_check(agent, [unhealthy], monkeypatch)
+    assert events[0][0] == "healthcheck_failed"
+
+    healthy = make_container("running", name="svc")
+    events = run_check(agent, [healthy], monkeypatch)
+    assert len(events) == 1
+    assert events[0][0] == "service_healthy"
+
+    # Third cycle, still healthy: silent again.
+    events = run_check(agent, [healthy], monkeypatch)
+    assert events == []
+
+
+def test_running_healthy_after_crash_loop_reemits_service_healthy(agent, monkeypatch):
+    """Recovery from a crash-loop (containers_not_running) is also a real
+    transition and must re-confirm health exactly once."""
+    crashing = make_container("restarting", restart_count=7, name="svc")
+    events = run_check(agent, [crashing], monkeypatch)
+    assert events[0][0] == "containers_not_running"
+
+    recovered = make_container("running", name="svc")
+    events = run_check(agent, [recovered], monkeypatch)
+    assert len(events) == 1
+    assert events[0][0] == "service_healthy"
+
+
+def test_different_services_tracked_independently(agent, monkeypatch):
+    """One service's health transition must not gate another service's emission."""
+    svc_a = make_container("running", name="svc-a", compose_service="svc-a")
+    svc_b = make_container("running", name="svc-b", compose_service="svc-b")
+
+    events = run_check(agent, [svc_a, svc_b], monkeypatch)
+    assert {e[0] for e in events} == {"service_healthy"}
+    assert len(events) == 2
+
+    # Both still healthy next cycle: both silent.
+    events = run_check(agent, [svc_a, svc_b], monkeypatch)
+    assert events == []
+
+
 def test_running_unhealthy_emits_healthcheck_failed(agent, monkeypatch):
     c = make_container("running", health="unhealthy")
     events = run_check(agent, [c], monkeypatch)

services/node-agent/tests/test_cleanup_control_plane_fs.py

@@ -0,0 +1,152 @@
+"""Tests for NodeAgent._cleanup_control_plane_fs() event-retention sweep.
+
+Regression coverage for the event-flood root cause: this method's event
+cleanup previously compared str(file_path) against the observer's checkpoint
+value with <=, which only worked for the old lexical-path checkpoint format.
+Once observer.py migrated node_checkpoints to int epoch timestamps, every
+comparison raised a TypeError (caught silently by the broad except), so this
+retention sweep was a no-op — the direct cause of the 358k-file backlog that
+paralyzed the supervisor's reconcile().
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import time
+
+import pytest
+
+import node_agent
+
+
+@pytest.fixture(autouse=True)
+def _isolated_runtime_state():
+    """EVENTS_DIR/STATE_DIR are a single temp dir shared for the whole test
+    session (conftest.py sets RUNTIME_PATH once at import time). Clear them
+    before and after each test here so tests that write real event/checkpoint
+    files don't leak state into each other or into other test modules."""
+    def _clear():
+        if node_agent.EVENTS_DIR.exists():
+            shutil.rmtree(node_agent.EVENTS_DIR)
+        node_agent.EVENTS_DIR.mkdir(parents=True, exist_ok=True)
+        checkpoint = node_agent.STATE_DIR / "observer_checkpoint.json"
+        checkpoint.unlink(missing_ok=True)
+
+    _clear()
+    yield
+    _clear()
+
+
+def _write_event(events_dir, node, ts, etype, svc="svc"):
+    """Write an event file whose filename AND on-disk mtime both reflect `ts`.
+
+    _cleanup_control_plane_fs's age check reads f.stat().st_mtime (real
+    filesystem mtime), not the filename-embedded timestamp — matching
+    production, where a file's mtime is set once at the moment node-agent
+    writes it. Tests must set mtime explicitly to simulate an "old" event.
+    """
+    node_dir = events_dir / node
+    node_dir.mkdir(parents=True, exist_ok=True)
+    p = node_dir / f"evt-{node}-{ts}-{etype}-{svc}.json"
+    p.write_text("{}")
+    os.utime(p, (ts, ts))
+    return p
+
+
+def _write_checkpoint(state_dir, node_checkpoints):
+    (state_dir / "observer_checkpoint.json").write_text(
+        json.dumps({"node_checkpoints": node_checkpoints})
+    )
+
+
+def test_int_checkpoint_deletes_old_processed_noise(agent):
+    """The bug: this must NOT raise, and must actually delete eligible files."""
+    events_dir = node_agent.EVENTS_DIR
+    state_dir = node_agent.STATE_DIR
+    now = int(time.time())
+    checkpoint = now - 100
+
+    old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy")
+    old_node_health = _write_event(events_dir, "piha", now - 4 * 86_400, "node_health")
+    _write_checkpoint(state_dir, {"piha": checkpoint})
+
+    agent._cleanup_control_plane_fs()
+
+    assert not old_healthy.exists()
+    assert not old_node_health.exists()
+
+
+def test_keeps_real_signals_regardless_of_age(agent):
+    """healthcheck_failed / incidents must survive this sweep indefinitely."""
+    events_dir = node_agent.EVENTS_DIR
+    state_dir = node_agent.STATE_DIR
+    now = int(time.time())
+    checkpoint = now - 100
+
+    healthcheck = _write_event(events_dir, "piha", now - 30 * 86_400, "healthcheck_failed")
+    ha_event = _write_event(events_dir, "piha", now - 30 * 86_400, "ha_entity_unavailable_long")
+    _write_checkpoint(state_dir, {"piha": checkpoint})
+
+    agent._cleanup_control_plane_fs()
+
+    assert healthcheck.exists()
+    assert ha_event.exists()
+
+
+def test_keeps_unprocessed_noise_even_if_old(agent):
+    """An event newer than the checkpoint is not guaranteed processed — keep it."""
+    events_dir = node_agent.EVENTS_DIR
+    state_dir = node_agent.STATE_DIR
+    now = int(time.time())
+    checkpoint = now - 100
+
+    unprocessed = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy", svc="future")
+    # Force it to sort after the checkpoint despite an old mtime by using a
+    # timestamp greater than checkpoint in the filename itself.
+    unprocessed.rename(events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json")
+    _write_checkpoint(state_dir, {"piha": checkpoint})
+
+    agent._cleanup_control_plane_fs()
+
+    assert (events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json").exists()
+
+
+def test_keeps_recent_processed_noise_within_three_day_buffer(agent):
+    """Past checkpoint but within the 3-day safety buffer: not yet eligible."""
+    events_dir = node_agent.EVENTS_DIR
+    state_dir = node_agent.STATE_DIR
+    now = int(time.time())
+    checkpoint = now - 100
+
+    recent = _write_event(events_dir, "piha", now - 3600, "service_healthy")
+    _write_checkpoint(state_dir, {"piha": checkpoint})
+
+    agent._cleanup_control_plane_fs()
+
+    assert recent.exists()
+
+
+def test_legacy_path_string_checkpoint_still_migrates_correctly(agent):
+    """Old observer builds stored a lexical path string per node; the embedded
+    timestamp must still be extracted so retention works during a rolling
+    upgrade instead of silently no-op'ing again."""
+    events_dir = node_agent.EVENTS_DIR
+    state_dir = node_agent.STATE_DIR
+    now = int(time.time())
+    checkpoint_ts = now - 100
+    legacy_path_value = f"/opt/homelab/events/piha/evt-piha-{checkpoint_ts}-node_health-node.json"
+
+    old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy")
+    _write_checkpoint(state_dir, {"piha": legacy_path_value})
+
+    agent._cleanup_control_plane_fs()
+
+    assert not old_healthy.exists()
+
+
+def test_no_checkpoint_file_is_noop_not_crash(agent):
+    events_dir = node_agent.EVENTS_DIR
+    _write_event(events_dir, "piha", int(time.time()) - 999999, "service_healthy")
+    # No checkpoint file written at all.
+    agent._cleanup_control_plane_fs()  # must not raise

services/node-agent/tests/test_control_plane_health.py

@@ -0,0 +1,64 @@
+"""Tests for NodeAgent._check_control_plane_health() transition-only emission.
+
+Regression coverage for the event-flood fix: the VPS control-plane HTTP probe
+used to emit service_healthy every single cycle it was reachable, which was
+part of the same flood as check_containers()'s per-service confirmation.
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import node_agent
+
+
+def run_probe(agent, monkeypatch, *, status=200, raise_exc=None):
+    """Run _check_control_plane_health() with urlopen mocked, capturing emitted events."""
+    emitted = []
+
+    def fake_emit(event_type, severity, service, message, payload=None):
+        emitted.append((event_type, severity, service, message, payload or {}))
+
+    monkeypatch.setattr(agent, "emit_event", fake_emit)
+
+    if raise_exc is not None:
+        def fake_urlopen(*a, **kw):
+            raise raise_exc
+    else:
+        resp = MagicMock()
+        resp.status = status
+
+        def fake_urlopen(*a, **kw):
+            return resp
+
+    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+    agent._check_control_plane_health()
+    return emitted
+
+
+def test_reachable_emits_service_healthy(agent, monkeypatch):
+    events = run_probe(agent, monkeypatch, status=200)
+    assert len(events) == 1
+    assert events[0][0] == "service_healthy"
+    assert events[0][2] == "control-plane"
+
+
+def test_reachable_second_cycle_is_silent(agent, monkeypatch):
+    run_probe(agent, monkeypatch, status=200)
+    events = run_probe(agent, monkeypatch, status=200)
+    assert events == []
+
+
+def test_unreachable_reemits_service_unhealthy_every_cycle(agent, monkeypatch):
+    """service_unhealthy is a real fault signal — must stay unconditional."""
+    run_probe(agent, monkeypatch, status=200)
+    events = run_probe(agent, monkeypatch, raise_exc=OSError("connection refused"))
+    assert events[0][0] == "service_unhealthy"
+    events = run_probe(agent, monkeypatch, raise_exc=OSError("connection refused"))
+    assert events[0][0] == "service_unhealthy"
+
+
+def test_recovery_after_unreachable_reemits_service_healthy(agent, monkeypatch):
+    run_probe(agent, monkeypatch, raise_exc=OSError("connection refused"))
+    events = run_probe(agent, monkeypatch, status=200)
+    assert len(events) == 1
+    assert events[0][0] == "service_healthy"