homelab-codex-ws/scripts/observer/observer.py

import os
import re
import sys
import json
import time
import glob
import logging
from logging.handlers import RotatingFileHandler
import urllib.request
import urllib.parse
import yaml
from datetime import datetime, timezone
from pathlib import Path

# Shared liveness logic (thresholds + fresh/stale/dead state machine) lives in
# the control-plane src so the observer and both operator UIs agree on what
# "dead" means.  Added to sys.path relative to this file (works both in the
# container, where the repo is mounted at /repo, and in the test tree).
_CP_SRC = Path(__file__).resolve().parent.parent.parent / "services" / "control-plane" / "src"
if str(_CP_SRC) not in sys.path:
    sys.path.insert(0, str(_CP_SRC))
from liveness import (  # noqa: E402
    compute_liveness,
    ttls_for,
    liveness_to_status,
    FRESH,
    STALE,
    DEAD,
    UNKNOWN,
)


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)


def _parse_ts(ts) -> float:
    """Return a Unix timestamp float from ts, which may be int/float or an ISO-8601 string.

    Events from node-agent use int(time.time()); events from stability-agent / events.py
    use ISO format ('2026-06-03T10:30:00Z').  Both appear in incident fields such as
    last_occurrence and resolved_at, so any arithmetic on them must go through here.
    Returns 0.0 on None or unparseable input so callers can use plain comparisons.
    """
    if ts is None:
        return 0.0
    if isinstance(ts, (int, float)):
        return float(ts)
    try:
        return datetime.fromisoformat(str(ts).replace("Z", "+00:00")).timestamp()
    except Exception:
        return 0.0


# Event filenames follow evt-<node>-<unixts>-<type>-<svc>.json (node_agent.py,
# ha_diag/event_emitter.py, and the observer's own _emit_node_transition).  The
# <unixts> is the authoritative ordering key for checkpointing — matched the same
# way operator_ui.py::_event_file_ts does (a 9–11 digit run flanked by dashes;
# real epochs are 10 digits and stay so until year 2286).
_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")


def _ts_from_event_name(name) -> "int | None":
    """Parse the embedded <unixts> from an event filename or full path.

    Returns the int epoch, or None when the name does not carry one (foreign
    prefix, events.py naming, junk file) so the caller can fall back to mtime.
    """
    m = _EVENT_TS_RE.search(Path(name).stem)
    return int(m.group(1)) if m else None


def _event_ts_from_path(file_path) -> int:
    """Ordering/checkpoint timestamp for an event file.

    Primary: the <unixts> embedded in the filename.  Fallback (name doesn't
    parse): the file's mtime.  NEVER returns 0 for an existing file — a value of
    0 would make the file compare as "older than the checkpoint" and be skipped
    forever, which is exactly the lexical-path poisoning this fix removes.  If
    even stat() fails, fall back to now() so the file is treated as new and gets
    a chance to be processed (and quarantined if truly unreadable).
    """
    ts = _ts_from_event_name(file_path)
    if ts is not None:
        return ts
    try:
        return int(os.stat(file_path).st_mtime)
    except OSError:
        return int(time.time())


def _checkpoint_ts_from_value(value) -> int:
    """Coerce a stored checkpoint value into an int epoch (migration helper).

    Historical formats of observer_checkpoint.json values:
      - int/float           → already a timestamp (current format) — kept as-is
      - path/name string    → the pre-fix lexical checkpoint; parse the embedded
                              <unixts> out of it so the node resumes near where it
                              left off instead of reprocessing everything
      - anything else / unparseable string → 0 (reprocess all events for that
        node).  Reprocessing is safe: process_event is idempotent w.r.t.
        last_seen/world_state, so re-ingesting duplicates cannot corrupt state —
        whereas guessing too high a checkpoint could silently drop events (the
        failure mode being fixed).  Bias to reprocess, never to skip.
    """
    if isinstance(value, bool):  # bool is an int subclass — exclude explicitly
        return 0
    if isinstance(value, (int, float)):
        return int(value)
    if isinstance(value, str) and value:
        ts = _ts_from_event_name(value)
        if ts is not None:
            return ts
    return 0


# Constants and Paths
RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab")
EVENTS_DIR = Path(RUNTIME_PATH) / "events"
STATE_DIR = Path(RUNTIME_PATH) / "state"
LOGS_DIR = Path(RUNTIME_PATH) / "logs"
WORLD_DIR = Path(RUNTIME_PATH) / "world"
OBSERVER_STATE_FILE = STATE_DIR / "observer_checkpoint.json"
FAILED_EVENTS_DIR = STATE_DIR / "observer_failed_events"

# Manual incident-resolve flag drop: an operator (or a future UI action) touches
# world/resolve-requests/<incident-id> and the observer picks it up here, once
# per loop iteration — see _process_resolve_requests. Chosen over adding a new
# operator-ui/API mutation endpoint (see COMMIT 1b rationale in the commit
# message): /action/mutate only knows about actions/<status>/<id>.json, there is
# no equivalent for incidents.json, and incidents.json is written exclusively by
# the observer (world/*.json is observer-owned, overwritten wholesale every
# cycle by _save_world) — a second writer would race it. A flag file needs no
# new HTTP surface, no new writer of world/incidents.json, and follows the same
# "operator drops a file, the owning process consumes it" pattern the actions
# pending/approved queue already uses.
RESOLVE_REQUESTS_DIR = WORLD_DIR / "resolve-requests"
# mkdir(mode=...) is masked by the process umask, so the SSH operator (group
# aerbot) cannot drop a flag file into a dir created at the default 0o755 —
# same defect/fix as executor.py's INBOX_DIR_MODE (2026-08-06): an explicit,
# idempotent os.chmod after mkdir, applied on the same path the dir is used.
RESOLVE_REQUESTS_DIR_MODE = 0o775

# Time-based fallback for incidents that can never receive the service_healthy
# event _resolve_incident() waits for (service removed/renamed/decommissioned
# from services.yaml, or a one-off test service) — see _prune_stale_world Case 3.
INCIDENT_STALE_RESOLVE_SECS = int(os.getenv("INCIDENT_STALE_RESOLVE_SECS", str(24 * 3600)))

REPO_ROOT = Path(__file__).parent.parent.parent
INVENTORY_TOPOLOGY = REPO_ROOT / "inventory" / "topology.yaml"

# --- SHADOW-READ: Prometheus up{} liveness (cutover etap 1) ----------------
# Optional parallel-run source.  When PROM_SHADOW_URL is set, the observer ALSO
# queries Prometheus `up{}` each cycle and LOGS any disagreement with its own
# event-driven liveness — but NEVER acts on it.  The authoritative liveness stays
# 100% event-driven (compute_liveness in _prune_stale_world).  Empty/unset →
# shadow disabled, observer behaves exactly as before (graceful, fail-open).
# Target value (do NOT hardcode — set via env/host override): http://100.95.58.48:9090
PROM_SHADOW_URL = os.environ.get("PROM_SHADOW_URL", "").strip()
PROM_SHADOW_TIMEOUT = int(os.getenv("PROM_SHADOW_TIMEOUT", "5"))

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

# --- PERSISTENT shadow-mismatch log (cutover etap 2) -----------------------
# SHADOW_LIVENESS_MISMATCH lines are cutover EVIDENCE and must survive a
# `docker rm`/recreate of the observer container.  stdout (json-file) logs do
# NOT: an observer recreate on 2026-07-14 destroyed the 07-13/14 mismatch
# material mid-analysis.  We therefore ALSO write each mismatch to a
# RotatingFileHandler on a HOST-mounted path so the file outlives the container.
#
# Path is the repo-conventional logs/<service>/ location under RUNTIME_PATH,
# i.e. /opt/homelab/logs/observer/shadow-liveness.log.  No dedicated bind-mount
# is required: the base compose already mounts the whole of /opt/homelab into
# the observer, so this file is on the host by construction, and the observer
# (uid 1000) creates the dir itself — avoiding the root-owned bind-source
# ownership footgun a separate /var/log mount would introduce.  Overridable via
# SHADOW_LOG_DIR for tests / non-container runs.
SHADOW_LOG_DIR = Path(os.getenv("SHADOW_LOG_DIR", str(LOGS_DIR / "observer")))
SHADOW_LOG_FILENAME = "shadow-liveness.log"


def _make_shadow_logger():
    """Build the dedicated PERSISTENT logger for SHADOW_LIVENESS_MISMATCH lines.

    Returns a `logging.getLogger("observer.shadow")` wired to a
    RotatingFileHandler (5 MiB x 5 backups — mismatches are short lines, that is
    weeks of headroom) writing to SHADOW_LOG_DIR/shadow-liveness.log with the
    same timestamped format as the main observer log.

    propagate=False keeps these records OUT of root/stdout: the call site still
    logs the mismatch to stdout via the ordinary `logger` (unchanged), so
    without this each mismatch would appear twice in `docker logs`.

    FAIL-SAFE: if the directory/file cannot be created or opened (permission,
    read-only mount, …) we log ONE warning on the main observer logger and
    return a handler-less logger.  `.info()` on a handler-less, non-propagating
    logger is a silent no-op, so a broken persistent log NEVER raises and NEVER
    takes the observer down — the mismatch still reaches stdout at the call site.
    Idempotent: any handler from a previous call is dropped first, so
    re-instantiation (tests, re-import) never double-writes or pins a stale path.
    """
    sl = logging.getLogger("observer.shadow")
    sl.setLevel(logging.INFO)
    sl.propagate = False
    for h in list(sl.handlers):
        sl.removeHandler(h)
        try:
            h.close()
        except Exception:
            pass
    try:
        os.makedirs(SHADOW_LOG_DIR, exist_ok=True)
        handler = RotatingFileHandler(
            SHADOW_LOG_DIR / SHADOW_LOG_FILENAME,
            maxBytes=5 * 1024 * 1024,
            backupCount=5,
        )
        handler.setFormatter(
            logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        )
        sl.addHandler(handler)
    except Exception as exc:
        logger.warning(
            "shadow-read: could not open persistent mismatch log at %s (%s) — "
            "falling back to stdout only, observer continues",
            SHADOW_LOG_DIR / SHADOW_LOG_FILENAME, exc,
        )
    return sl


class Observer:
    def __init__(self):
        # Per-node-directory checkpoint keyed on the last-processed event
        # TIMESTAMP (int epoch): {"vps": 1784000000, "piha": 1784000123}.
        # A file is "new" iff its event timestamp > the node's checkpoint.
        # This replaces the earlier lexical-PATH comparison, which permanently
        # poisoned a node the moment a single file with a lexically-larger name
        # landed in its dir (e.g. evt-unknown-… > evt-piha-…): every genuinely
        # newer event then sorted "before" the checkpoint and was skipped forever.
        self.node_checkpoints: dict = {}
        self.world_state = {
            "nodes": {},
            "services": {},
            "deployments": {},
            "incidents": {},
            "summary": {
                "last_update": datetime.now(timezone.utc).isoformat(),
                "status": "initializing",
                "active_incidents_count": 0
            }
        }
        self.inventory = self._load_inventory()
        # Nodes declared `status: dormant` in inventory/topology.yaml: expected
        # offline (e.g. chelsty site, hardware down since ~2026-06-01).  The
        # observer keeps their last-known world state but never reclassifies
        # liveness for them, so no node_offline/node_stale/node_online events
        # are emitted while a node is dormant.  See kb/decisions/architektura-2026-07-28.md.
        self.dormant_nodes = {
            name for name, info in self.inventory["nodes"].items()
            if info.get("status") == "dormant"
        }
        if self.dormant_nodes:
            logger.info("Dormant nodes (no liveness tracking): %s",
                        sorted(self.dormant_nodes))
        self._ensure_dirs()
        self._load_checkpoint()
        # Persistent SHADOW_LIVENESS_MISMATCH sink (survives container recreate).
        self.shadow_logger = _make_shadow_logger()

    def _ensure_dirs(self):
        WORLD_DIR.mkdir(parents=True, exist_ok=True)
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        EVENTS_DIR.mkdir(parents=True, exist_ok=True)
        LOGS_DIR.mkdir(parents=True, exist_ok=True)
        FAILED_EVENTS_DIR.mkdir(parents=True, exist_ok=True)
        RESOLVE_REQUESTS_DIR.mkdir(parents=True, exist_ok=True)
        try:
            os.chmod(RESOLVE_REQUESTS_DIR, RESOLVE_REQUESTS_DIR_MODE)
        except OSError as e:
            logger.warning(
                f"Could not set mode {oct(RESOLVE_REQUESTS_DIR_MODE)} on "
                f"{RESOLVE_REQUESTS_DIR}: {e}"
            )

    def _quarantine_event_file(self, file_path: str, node_dir: str, exc: Exception) -> None:
        """Move an unreadable/unprocessable event out of the hot path."""
        src = Path(file_path)
        dest_dir = FAILED_EVENTS_DIR / node_dir
        dest_dir.mkdir(parents=True, exist_ok=True)
        dest = dest_dir / src.name
        if dest.exists():
            dest = dest_dir / f"{src.stem}-{int(time.time())}{src.suffix}"
        try:
            os.replace(src, dest)
            logger.error(
                "Quarantined bad event for node_dir=%s: %s -> %s (%s: %s)",
                node_dir, src, dest, type(exc).__name__, exc,
            )
        except Exception as move_exc:
            logger.error(
                "Failed to quarantine bad event for node_dir=%s: %s (%s: %s); move error=%s: %s",
                node_dir, src, type(exc).__name__, exc, type(move_exc).__name__, move_exc,
            )

    def _load_inventory(self):
        inventory = {"nodes": {}, "services": {}}
        try:
            if INVENTORY_TOPOLOGY.exists():
                with open(INVENTORY_TOPOLOGY, "r") as f:
                    topo = yaml.safe_load(f)
                    for node_name, node_info in topo.get("nodes", {}).items():
                        node_info = node_info or {}
                        inventory["nodes"][node_name] = {
                            "roles": node_info.get("roles", []),
                            "connectivity": node_info.get("connectivity", {}),
                            # topology node status: active (default) | dormant
                            "status": node_info.get("status", "active"),
                        }
            
            # Load service assignments from hosts files
            hosts_dir = REPO_ROOT / "hosts"
            for host_dir in hosts_dir.iterdir():
                if host_dir.is_dir():
                    svc_file = host_dir / "services.yaml"
                    if svc_file.exists():
                        with open(svc_file, "r") as f:
                            svc_data = yaml.safe_load(f)
                            host_name = svc_data.get("host")
                            for svc_name, svc_info in svc_data.get("services", {}).items():
                                if host_name not in inventory["services"]:
                                    inventory["services"][host_name] = {}
                                inventory["services"][host_name][svc_name] = {
                                    "role": svc_info.get("role"),
                                    "exposure": svc_info.get("exposure")
                                }
        except Exception as e:
            logger.error(f"Failed to load inventory: {e}")
        return inventory

    def _load_checkpoint(self):
        if OBSERVER_STATE_FILE.exists():
            try:
                with open(OBSERVER_STATE_FILE, "r") as f:
                    checkpoint = json.load(f)

                if "node_checkpoints" in checkpoint:
                    # Per-directory checkpoints.  Values may be int epochs (current
                    # format) OR pre-fix path strings — coerce every value to an
                    # int timestamp so a checkpoint file written by the old
                    # lexical-path observer migrates transparently on first start.
                    raw = checkpoint["node_checkpoints"] or {}
                    self.node_checkpoints = {
                        node: _checkpoint_ts_from_value(val)
                        for node, val in raw.items()
                    }
                    if any(not isinstance(v, (int, float)) for v in raw.values()):
                        logger.info(
                            "Migrated path-based node_checkpoints → timestamps: %s",
                            self.node_checkpoints,
                        )
                elif "last_processed_file" in checkpoint:
                    # Migrate the very old single-file checkpoint: extract node dir
                    # from the path and the timestamp from the filename.
                    old = checkpoint["last_processed_file"]
                    if old:
                        try:
                            node_dir = Path(old).relative_to(EVENTS_DIR).parts[0]
                            self.node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)}
                            logger.info(f"Migrated old checkpoint → node_checkpoints: {self.node_checkpoints}")
                        except Exception:
                            pass  # Bad path — start fresh

                self._load_world_from_disk()
            except Exception as e:
                logger.error(f"Failed to load checkpoint: {e}")

    def _load_world_from_disk(self):
        # Optional: Load existing state to resume faster
        files = {
            "nodes": WORLD_DIR / "nodes.json",
            "services": WORLD_DIR / "services.json",
            "deployments": WORLD_DIR / "deployments.json",
            "incidents": WORLD_DIR / "incidents.json",
            "summary": WORLD_DIR / "runtime-summary.json"
        }
        for key, path in files.items():
            if path.exists():
                try:
                    with open(path, "r") as f:
                        self.world_state[key] = json.load(f)
                except Exception as e:
                    logger.error(f"Failed to load {key} state: {e}")

    def _save_checkpoint(self):
        try:
            _atomic_write_json(OBSERVER_STATE_FILE, {"node_checkpoints": self.node_checkpoints})
        except Exception as e:
            logger.error(f"Failed to save checkpoint: {e}")

    def _emit_node_transition(self, node_name, prev, new, node_info, now):
        """Write an event when a node crosses a liveness boundary.

        Makes liveness transitions visible (panel event feed) and actionable
        (the supervisor routes node_offline/node_stale to an alert).  Without
        this the outage would be silent — exactly the failure mode being fixed.

        Recovery (stale/dead → fresh) emits node_online so a node coming back is
        as visible as it going down.

        The affected node is the top-level ``node`` (so the supervisor can route
        on it) AND ``payload.affected_node`` — note events.py::emit_event would
        tag node=gethostname() (the observer host, vps), so the observer writes
        the event directly instead of using that helper.

        ``source: "observer"`` marks the event so run_once skips re-ingesting it
        (re-ingestion would reset the node's last_seen and resurrect it).
        """
        if new == DEAD:
            etype, severity = "node_offline", "high"
        elif new == STALE:
            etype, severity = "node_stale", "warning"
        elif new == FRESH and prev in (STALE, DEAD):
            etype, severity = "node_online", "info"
        else:
            return  # not a transition we alert on

        ts = int(now)
        event_id = f"evt-{node_name}-{ts}-{etype}-node"
        age = int(now - _parse_ts(node_info.get("last_seen")))
        event = {
            "id": event_id,
            "timestamp": ts,
            "date": datetime.now(timezone.utc).isoformat(),
            "type": etype,
            "severity": severity,
            "node": node_name,
            "service": None,
            "source": "observer",
            "message": f"Node {node_name} liveness {prev} -> {new} (last_seen {age}s ago)",
            "payload": {
                "affected_node": node_name,
                "from": prev,
                "to": new,
                "last_seen": node_info.get("last_seen"),
                "age_secs": age,
            },
        }
        try:
            node_dir = EVENTS_DIR / node_name
            node_dir.mkdir(parents=True, exist_ok=True)
            _atomic_write_json(node_dir / f"{event_id}.json", event)
            logger.warning("Node %s transition %s -> %s (emitted %s)",
                           node_name, prev, new, etype)
        except Exception as exc:
            logger.error("Failed to emit node transition for %s: %s", node_name, exc)

    def _query_prometheus_liveness(self):
        """SHADOW-READ (cutover etap 1): Prometheus up{} → {node_name: up_bool}.

        Parallel-run only: the result is compared against — and logged next to —
        the authoritative event-driven liveness, but NEVER changes it.  See
        PROM_SHADOW_URL.  Returns {} when shadow is disabled OR on ANY error, so
        the observer can never crash or change behaviour because of this call:
        every failure is fail-open (info/warning, never error, never raised).

        Each `up` series is keyed by its `node` label (Prometheus fleet-node
        targets carry node:vps/piha/solaria/lustro).  Series without a node label
        (e.g. the prometheus self-scrape up{job="prometheus"}) are ignored.
        """
        url = PROM_SHADOW_URL
        if not url:
            return {}  # shadow disabled — graceful no-op, observer unchanged
        query_url = url.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode({"query": "up"})
        try:
            with urllib.request.urlopen(query_url, timeout=PROM_SHADOW_TIMEOUT) as resp:
                payload = json.loads(resp.read().decode("utf-8"))
        except Exception as exc:
            # Fail-open: Prometheus down / timeout / bad JSON → no shadow data.
            # Deliberately NOT logger.error — shadow-read must never look like a
            # critical observer failure.
            logger.warning(
                "shadow-read: Prometheus query failed (%s: %s) — fail-open, "
                "event liveness unaffected", type(exc).__name__, exc,
            )
            return {}
        result: dict = {}
        try:
            for series in payload.get("data", {}).get("result", []):
                node = series.get("metric", {}).get("node")
                if not node:
                    continue  # e.g. up{job="prometheus"} — no node label to map
                value = series.get("value", [None, None])[1]
                result[node] = (value == "1")
        except Exception as exc:
            logger.warning(
                "shadow-read: could not parse Prometheus response (%s: %s) — "
                "fail-open", type(exc).__name__, exc,
            )
            return {}
        return result

    def _shadow_compare_liveness(self, node_name, event_liveness, node_info, prom_map, now):
        """SHADOW-READ comparison (cutover etap 1): log event-vs-Prometheus
        liveness disagreement WITHOUT changing anything.

        Maps both sources onto a shared up/down axis:
          - event DEAD           ≈ prom down
          - event FRESH or STALE ≈ prom up   (STALE = "seen recently, just
            ageing" — still counts as up for this comparison; documented
            assumption of etap 1)

        A node Prometheus doesn't know (no series, e.g. chelsty-infra — not
        scraped) is NOT a mismatch: missing data is not disagreement (debug only).
        This method is read-only w.r.t. liveness/status and must never raise.
        """
        if node_name not in prom_map:
            logger.debug("shadow-read: no prom data for node=%s", node_name)
            return
        prom_up = prom_map[node_name]
        event_up = event_liveness in (FRESH, STALE)
        age = now - _parse_ts(node_info.get("last_seen"))
        if event_up != prom_up:
            logger.info(
                "SHADOW_LIVENESS_MISMATCH node=%s event=%s prom=%s last_seen_age=%.0fs",
                node_name, event_liveness, "up" if prom_up else "down", age,
            )
            # ALSO write to the persistent, host-mounted log so this evidence
            # survives a container recreate (stdout json-file logs do not).
            # Handler-less (fail-safe) → silent no-op; never affects the above.
            self.shadow_logger.info(
                "SHADOW_LIVENESS_MISMATCH node=%s event=%s prom=%s last_seen_age=%.0fs",
                node_name, event_liveness, "up" if prom_up else "down", age,
            )
        else:
            logger.debug(
                "shadow-read agree node=%s event=%s prom=%s last_seen_age=%.0fs",
                node_name, event_liveness, "up" if prom_up else "down", age,
            )

    def _process_resolve_requests(self, now):
        """Manual incident-resolve path (COMMIT 1b): consume flag files dropped
        at world/resolve-requests/<incident-id>.

        There is no operator-ui/API mutation endpoint for incidents (only for
        actions, via /action/mutate — see RESOLVE_REQUESTS_DIR comment), and
        incidents.json is exclusively observer-owned, so a flag file the
        observer itself polls is the smallest addition that lets an operator
        force-resolve a wedged incident without editing world/incidents.json
        by hand.

        The flag is always removed, even when the incident_id is unknown or
        already resolved, so a stale/mistyped/duplicate flag cannot sit
        forever looking like it hasn't been picked up yet.
        """
        if not RESOLVE_REQUESTS_DIR.exists():
            return
        for flag_path in RESOLVE_REQUESTS_DIR.iterdir():
            if not flag_path.is_file():
                continue
            incident_id = flag_path.name
            inc = self.world_state["incidents"].get(incident_id)
            if inc and inc.get("status") == "active":
                logger.info(
                    f"Manually resolving incident {incident_id} "
                    f"(service={inc.get('service')}, node={inc.get('node')}) "
                    f"via resolve-request flag"
                )
                inc["status"] = "resolved"
                inc["resolved_at"] = now
                inc["resolved_reason"] = "manual_operator"
                svc_key = f"{inc.get('node')}/{inc.get('service')}"
                svc = self.world_state["services"].get(svc_key)
                if svc and svc.get("incident_id") == incident_id:
                    svc["incident_id"] = None
            elif inc:
                logger.info(
                    f"Resolve-request flag for {incident_id} ignored "
                    f"(status already '{inc.get('status')}') — removing flag"
                )
            else:
                logger.warning(
                    f"Resolve-request flag for unknown incident {incident_id} — removing flag"
                )
            try:
                flag_path.unlink()
            except OSError as exc:
                logger.error(f"Failed to remove resolve-request flag {flag_path}: {exc}")

    def _prune_stale_world(self):
        """Remove world-state entries for nodes absent from the topology inventory.

        Root cause this guards against: when NODE_NAME env var is unset, node_agent.py
        falls back to socket.gethostname(), which inside a Docker container returns the
        12-char hex container ID (e.g. 'be17cb6eb0f6') instead of the canonical host name
        ('vps').  The observer ingests those events and creates ghost entries that never
        expire on their own.

        Also ages out resolved incidents older than 7 days to keep world state lean.
        """
        known_nodes = set(self.inventory["nodes"].keys())
        if not known_nodes:
            # Inventory failed to load — don't prune to avoid wiping valid state.
            return

        stale_nodes = [n for n in list(self.world_state["nodes"].keys())
                       if n not in known_nodes]
        for n in stale_nodes:
            logger.info(f"Pruning stale node from world state: {n}")
            del self.world_state["nodes"][n]

        stale_svcs = [k for k in list(self.world_state["services"].keys())
                      if k.split("/")[0] in stale_nodes]
        for k in stale_svcs:
            logger.info(f"Pruning stale service from world state: {k}")
            del self.world_state["services"][k]

        # Prune ghost service keys whose service-name portion is a hash-prefixed
        # Docker stale-state artifact (e.g. "9e36297651e7_control-plane-observer").
        # These are created when node-agent incorrectly uses c.name instead of the
        # compose label, and accumulate on every container rebuild.
        # Pattern: <node>/<12hexchars>_<real-name>
        ghost_svcs = [
            k for k in list(self.world_state["services"].keys())
            if len(k.split("/", 1)) == 2
            and len(k.split("/", 1)[1]) > 13
            and k.split("/", 1)[1][12] == "_"
            and all(ch in "0123456789abcdef" for ch in k.split("/", 1)[1][:12])
        ]
        for k in ghost_svcs:
            logger.info(f"Pruning ghost (hash-prefixed) service key from world state: {k}")
            del self.world_state["services"][k]

        now = time.time()

        # --- Authoritative node liveness (fresh / stale / dead) -------------
        # Status is derived from how long ago the node last reported, NOT from
        # the last status event.  This runs every cycle (including cycles with
        # no new events — see run_once), so a node that silently stops sending
        # heartbeats transitions on its own without any node_offline event ever
        # being emitted.  This is the fix for the "dead node shown NOMINAL"
        # class of silent outage.
        #
        # SHADOW-READ (cutover etap 1): fetch Prometheus up{} ONCE per cycle
        # (before the node loop, not per-node) for comparison-only logging.  {}
        # when disabled/unreachable — fail-open, event liveness is authoritative.
        prom_liveness_map = self._query_prometheus_liveness()
        for node_name, node_info in self.world_state["nodes"].items():
            if node_name in self.dormant_nodes:
                # Dormant node: keep last-known state frozen, emit nothing.
                continue
            roles = (node_info.get("roles")
                     or self.inventory["nodes"].get(node_name, {}).get("roles", []))
            liveness = compute_liveness(
                node_info.get("last_seen"), now=now, ttls=ttls_for(node_name, roles)
            )
            if liveness == UNKNOWN:
                # No last_seen yet — don't guess a node dead. Leave status as-is.
                continue
            prev = node_info.get("liveness")
            node_info["liveness"] = liveness
            new_status = liveness_to_status(liveness)
            if new_status:
                node_info["status"] = new_status
            # Emit on a real transition only.  prev is None on the very first
            # classification after (re)start — that is a baseline, not an event.
            if prev is not None and prev != liveness:
                self._emit_node_transition(node_name, prev, liveness, node_info, now)
            # SHADOW-READ (cutover etap 1): compare (log-only) the event-derived
            # `liveness` computed above against Prometheus up{}.  Purely additive —
            # does NOT read back or mutate node_info["liveness"]/["status"], and
            # the authoritative decision above is already committed.
            self._shadow_compare_liveness(node_name, liveness, node_info, prom_liveness_map, now)

        # Manual resolve-request flags: independent of the linked/orphan
        # distinction below (an operator can force-resolve either kind), so
        # process before the auto-resolve cases. Own try/except: a bad flag
        # file must not block the auto-resolve cases below from running.
        try:
            self._process_resolve_requests(now)
        except Exception as exc:
            logger.error(f"Error processing resolve-request flags: {exc}")

        try:
            # Collect incident_ids currently referenced by any service entry.
            linked_ids: set = {
                svc.get("incident_id")
                for svc in self.world_state["services"].values()
                if svc.get("incident_id")
            }

            # Case 1 — service is healthy but still points at an active incident.
            # process_event already calls _resolve_incident on service_healthy events,
            # but if the observer restarted with on-disk state where the link was
            # intact (inconsistency from a pre-atomic-write crash), it may not get
            # resolved until the next service_healthy event is processed.  Resolve
            # immediately — a healthy service cannot have an ongoing incident.
            for svc_key, svc in self.world_state["services"].items():
                if svc.get("status") != "healthy":
                    continue
                inc_id = svc.get("incident_id")
                if not inc_id:
                    continue
                inc = self.world_state["incidents"].get(inc_id, {})
                if inc.get("status") == "active":
                    logger.info(
                        f"Auto-resolving incident {inc_id} for {svc_key}: "
                        f"service is healthy"
                    )
                    inc["status"] = "resolved"
                    inc["resolved_at"] = now
                    svc["incident_id"] = None
                    linked_ids.discard(inc_id)

            # Case 2 — orphaned active incident: no service entry links to it and
            # last_occurrence is older than 5 minutes (guard against creation races).
            # These are the stale records left behind when on-disk state was
            # inconsistent: the service entry had incident_id cleared but incidents.json
            # still had the record as "active".
            for inc_id, inc in self.world_state["incidents"].items():
                if inc.get("status") != "active":
                    continue
                if inc_id in linked_ids:
                    continue
                age = now - _parse_ts(inc.get("last_occurrence"))
                if age > 300:  # 5-minute guard
                    logger.info(
                        f"Auto-resolving orphaned incident {inc_id} "
                        f"(service={inc.get('service')}, node={inc.get('node')}): "
                        f"no service references it, age={int(age)}s"
                    )
                    inc["status"] = "resolved"
                    inc["resolved_at"] = now

            # Case 3 — long-idle active incident, time-based fallback (COMMIT 1a).
            # _resolve_incident() only fires on a service_healthy/service_recovered
            # event, which a removed/renamed/decommissioned or one-off test
            # service will never emit again — such an incident would otherwise
            # stay "active" forever, unlike Case 2 above it is NOT limited to
            # orphaned (unlinked) incidents: a service entry can keep pointing at
            # incident_id indefinitely too (that lingering link is exactly the
            # case _resolve_incident() never got a chance to clear), so this
            # checks every active incident regardless of linked_ids.
            for inc_id, inc in self.world_state["incidents"].items():
                if inc.get("status") != "active":
                    continue
                age = now - _parse_ts(inc.get("last_occurrence"))
                if age > INCIDENT_STALE_RESOLVE_SECS:
                    logger.info(
                        f"Auto-resolving stale incident {inc_id} "
                        f"(service={inc.get('service')}, node={inc.get('node')}): "
                        f"no events for {int(age)}s "
                        f"(threshold {INCIDENT_STALE_RESOLVE_SECS}s)"
                    )
                    inc["status"] = "resolved"
                    inc["resolved_at"] = now
                    inc["resolved_reason"] = "auto_stale_no_events_24h"
                    svc_key = f"{inc.get('node')}/{inc.get('service')}"
                    svc = self.world_state["services"].get(svc_key)
                    if svc and svc.get("incident_id") == inc_id:
                        svc["incident_id"] = None

        except Exception as exc:
            logger.error(f"Error during incident auto-resolve in _prune_stale_world: {exc}")

        # Remove resolved incidents older than 7 days.
        # Use _parse_ts so ISO-string resolved_at values are handled correctly.
        stale_incidents = [
            k for k, v in self.world_state["incidents"].items()
            if v.get("status") == "resolved"
            and now - _parse_ts(v.get("resolved_at")) > 7 * 86400
        ]
        for k in stale_incidents:
            del self.world_state["incidents"][k]

    def _save_world(self):
        self.world_state["summary"]["last_update"] = datetime.now(timezone.utc).isoformat()
        active_incidents = [
            k for k, v in self.world_state["incidents"].items() if v.get("status") == "active"
        ]
        self.world_state["summary"]["active_incidents_count"] = len(active_incidents)
        self.world_state["summary"]["node_count"] = len(self.world_state["nodes"])
        self.world_state["summary"]["service_count"] = len(self.world_state["services"])

        if active_incidents:
            self.world_state["summary"]["status"] = "degraded"
        else:
            self.world_state["summary"]["status"] = "nominal"

        files = {
            "nodes.json": self.world_state["nodes"],
            "services.json": self.world_state["services"],
            "deployments.json": self.world_state["deployments"],
            "incidents.json": self.world_state["incidents"],
            "recommendations.json": [],
            "runtime-summary.json": self.world_state["summary"]
        }
        for filename, data in files.items():
            try:
                _atomic_write_json(WORLD_DIR / filename, data)
            except Exception as e:
                logger.error(f"Failed to save {filename}: {e}")

    def process_event(self, event):
        etype = event.get("type")
        node = event.get("node")
        service = event.get("service")
        severity = event.get("severity")
        timestamp = event.get("timestamp")
        cid = event.get("correlation_id")
        payload = event.get("payload", {})

        # 1. Update Node State
        if node not in self.world_state["nodes"]:
            self.world_state["nodes"][node] = {
                "status": "unknown",
                "last_seen": None,
                "roles": self.inventory["nodes"].get(node, {}).get("roles", [])
            }
        self.world_state["nodes"][node]["last_seen"] = timestamp

        if etype == "node_online":
            self.world_state["nodes"][node]["status"] = "online"
        elif etype == "node_offline":
            self.world_state["nodes"][node]["status"] = "offline"

        elif etype == "node_health":
            # Regular heartbeat from node-agent; updates resource metrics.
            # Clears disk_pressure if disk is now healthy (< warn threshold).
            self.world_state["nodes"][node]["status"] = "online"
            self.world_state["nodes"][node].update({
                "disk_usage_pct": payload.get("disk_pct"),
                "mem_usage_pct":  payload.get("mem_pct"),
                "cpu_usage_pct":  payload.get("cpu_pct"),
            })
            if (payload.get("disk_pct") or 0) < 75:
                self.world_state["nodes"][node].pop("disk_pressure", None)

        elif etype == "disk_pressure":
            # Emitted when disk usage crosses 75 % (medium) or 85 % (high).
            # The supervisor reads disk_pressure to generate disk_cleanup actions.
            self.world_state["nodes"][node]["disk_pressure"] = severity
            self.world_state["nodes"][node]["disk_usage_pct"] = payload.get("usage_pct")

        elif etype == "high_memory":
            # Memory pressure observation; recorded on the node for correlation.
            # No automated action — operator decides if a container restart helps.
            self.world_state["nodes"][node]["memory_pressure"] = severity
            self.world_state["nodes"][node]["mem_usage_pct"] = payload.get("usage_pct")

        elif etype == "high_cpu":
            # CPU pressure observation; recorded for visibility.
            self.world_state["nodes"][node]["cpu_pressure"] = severity
            self.world_state["nodes"][node]["cpu_usage_pct"] = payload.get("usage_pct")

        # 2. Update Service State
        if service and service != "all":
            svc_key = f"{node}/{service}"
            if svc_key not in self.world_state["services"]:
                self.world_state["services"][svc_key] = {
                    "node": node,
                    "service": service,
                    "status": "unknown",
                    "last_check": None,
                    "incident_id": None
                }
            self.world_state["services"][svc_key]["last_check"] = timestamp

            if etype == "service_recovered":
                self.world_state["services"][svc_key]["status"] = "healthy"
                self._resolve_incident(svc_key, timestamp)
            elif etype == "service_healthy":
                # Positive confirmation from node-agent that a managed container
                # is running. This keeps services.json populated so the supervisor
                # can correctly detect drift (absent entry = never reported = unknown,
                # not the same as confirmed missing).
                # Also resolve any active incident — if a service that had been
                # unhealthy/crashing is now confirmed healthy, the incident is over.
                self.world_state["services"][svc_key]["status"] = "healthy"
                self._resolve_incident(svc_key, timestamp)
            elif etype in ["service_unhealthy", "healthcheck_failed", "containers_not_running"]:
                # containers_not_running: node-agent (node_agent.py) — and the
                # stability-agent — report a *managed* container that has exited /
                # dead or is crash-looping (restarting past RestartCount threshold).
                # Treat it exactly like the other hard-failure signals: mark the
                # service unhealthy and open an incident.  The incident's
                # trigger_type is the event type ("containers_not_running"), which
                # the supervisor already recognises in CONTAINER_RESTART_TRIGGERS and
                # remediates with a low-risk container_restart (vs. a full redeploy).
                #
                # Before this branch existed, containers_not_running fell through
                # this if/elif chain entirely: node-agent saw the dead container and
                # emitted the event, the observer bumped last_check but left status at
                # its last "healthy" value and created NO incident.  The supervisor
                # then saw no drift and generated no action — so a dead / crash-looping
                # container produced ZERO operator alerts.  The monitoring loop was
                # silently broken (matches the "action queue empty despite failures"
                # symptom).  (node_agent.py's own comment claims this event "rides the
                # existing, supervisor-wired remediation path" — that path only exists
                # once the observer opens the incident here.)
                self.world_state["services"][svc_key]["status"] = "unhealthy"
                self._handle_incident(svc_key, event)
            elif etype in ["container_restarting", "container_state_unexpected"]:
                # Intentionally observational — parity with node_agent.check_containers,
                # which emits these as low/medium severity and explicitly does NOT wire
                # them to any supervisor trigger.  Causes: a transient post-deploy
                # restart still below the crash-loop threshold, an operator `docker
                # pause`, or an unhandled-but-not-fault docker state.
                #
                # We deliberately do NOT set status=unhealthy (that would make the
                # supervisor generate remediation for a transient blip — a redeploy,
                # since there is no CONTAINER_RESTART_TRIGGERS incident) and do NOT
                # open an incident (noise).  But the event must not vanish, so we
                # record a lightweight observational trace on the service entry; the
                # raw event also stays visible in the /events feed.  If a restart is a
                # real crash-loop it escalates on its own: node-agent re-emits it as
                # containers_not_running once RestartCount crosses the threshold, and
                # the branch above then alarms.
                self.world_state["services"][svc_key]["last_observation"] = {
                    "type": etype,
                    "severity": severity,
                    "timestamp": timestamp,
                    "message": event.get("message"),
                }

        # 3. Update Deployment State
        if etype.startswith("deployment_") and cid:
            if cid not in self.world_state["deployments"]:
                self.world_state["deployments"][cid] = {
                    "node": node,
                    "service": service,
                    "status": "unknown",
                    "started_at": None,
                    "finished_at": None,
                    "events": []
                }
            self.world_state["deployments"][cid]["events"].append({
                "type": etype,
                "timestamp": timestamp,
                "payload": payload
            })
            if etype == "deployment_started":
                self.world_state["deployments"][cid]["status"] = "in_progress"
                self.world_state["deployments"][cid]["started_at"] = timestamp
            elif etype == "deployment_completed":
                self.world_state["deployments"][cid]["status"] = "completed"
                self.world_state["deployments"][cid]["finished_at"] = timestamp
            elif etype == "deployment_failed":
                self.world_state["deployments"][cid]["status"] = "failed"
                self.world_state["deployments"][cid]["finished_at"] = timestamp
                # Deployment failure often creates an incident
                self._handle_deployment_failure(event)

    def _handle_incident(self, svc_key, event):
        # Correlation: collapse repeated failures for the same service on the same node
        active_incident = self.world_state["services"][svc_key].get("incident_id")
        
        if active_incident and active_incident in self.world_state["incidents"]:
            incident = self.world_state["incidents"][active_incident]
            if incident["status"] == "active":
                incident["last_occurrence"] = event["timestamp"]
                incident["occurrence_count"] = incident.get("occurrence_count", 1) + 1
                incident["events"].append(event["timestamp"])
                return

        # Create new incident
        incident_id = f"inc-{int(time.time())}-{event.get('node')}-{event.get('service')}"
        self.world_state["incidents"][incident_id] = {
            "id": incident_id,
            "node": event.get("node"),
            "service": event.get("service"),
            "status": "active",
            "severity": event.get("severity"),
            # trigger_type records the event type that opened this incident so that
            # the supervisor can choose the appropriate remediation action
            # (container_restart for containers_not_running / healthcheck_failed
            # vs. a full redeploy for other causes).
            "trigger_type": event.get("type"),
            "started_at": event.get("timestamp"),
            "last_occurrence": event.get("timestamp"),
            "occurrence_count": 1,
            "events": [event["timestamp"]],
            "correlation_id": event.get("correlation_id")
        }
        self.world_state["services"][svc_key]["incident_id"] = incident_id

    def _resolve_incident(self, svc_key, timestamp):
        incident_id = self.world_state["services"][svc_key].get("incident_id")
        if incident_id and incident_id in self.world_state["incidents"]:
            if self.world_state["incidents"][incident_id]["status"] == "active":
                self.world_state["incidents"][incident_id]["status"] = "resolved"
                self.world_state["incidents"][incident_id]["resolved_at"] = timestamp
        self.world_state["services"][svc_key]["incident_id"] = None

    def _handle_deployment_failure(self, event):
        # Specific logic for deployment failures
        svc_key = f"{event.get('node')}/{event.get('service')}"
        self._handle_incident(svc_key, event)
        
        # Link diagnostics if available in payload
        incident_id = self.world_state["services"][svc_key].get("incident_id")
        if incident_id and incident_id in self.world_state["incidents"]:
            payload = event.get("payload", {})
            if "diagnostics_file" in payload:
                self.world_state["incidents"][incident_id]["diagnostics_ref"] = payload["diagnostics_file"]
            elif "error" in payload:
                self.world_state["incidents"][incident_id]["last_error"] = payload["error"]

    def run_once(self):
        # Update heartbeat
        heartbeat_file = STATE_DIR / "observer.heartbeat"
        try:
            heartbeat_file.touch()
        except Exception as e:
            logger.error(f"Failed to touch heartbeat file: {e}")

        # Collect all event files grouped by node directory.  A file is "new"
        # when its event TIMESTAMP (from the filename, mtime fallback) is greater
        # than the node's checkpoint timestamp — never a lexical path compare, so
        # a lexically-smaller-but-newer filename (evt-piha-… after a stray
        # evt-unknown-…) can no longer poison a node into skipping every event.
        all_files = glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True)

        new_files = []
        for file_path in all_files:
            try:
                node_dir = str(Path(file_path).relative_to(EVENTS_DIR).parts[0])
            except (IndexError, ValueError):
                node_dir = "__unknown__"
            ev_ts = _event_ts_from_path(file_path)
            last_for_node = self.node_checkpoints.get(node_dir, 0)
            if ev_ts > last_for_node:
                new_files.append((ev_ts, node_dir, file_path))

        # Process oldest-first (tie-break on path for determinism) so the
        # checkpoint advances monotonically in time and a mid-batch crash resumes
        # from the right place.
        new_files.sort(key=lambda t: (t[0], t[2]))

        if not new_files:
            # Even if no new events, prune stale entries and refresh summary freshness.
            self._prune_stale_world()
            self._save_world()
            return

        logger.info(f"Processing {len(new_files)} new events across "
                    f"{len({n for _, n, _ in new_files})} node(s)")
        for ev_ts, node_dir, file_path in new_files:
            try:
                with open(file_path, "r") as f:
                    event = json.load(f)
                # Skip events the observer emitted itself (node liveness
                # transitions).  Re-ingesting them would call process_event,
                # which sets last_seen = event timestamp and would resurrect a
                # node we just declared dead.  They are still consumed by the
                # supervisor (alerting) and the panel event feed.
                if event.get("source") == "observer":
                    if ev_ts > self.node_checkpoints.get(node_dir, 0):
                        self.node_checkpoints[node_dir] = ev_ts
                    continue
                self.process_event(event)
                # Advance per-node checkpoint by timestamp (only forward).
                if ev_ts > self.node_checkpoints.get(node_dir, 0):
                    self.node_checkpoints[node_dir] = ev_ts
            except Exception as e:
                logger.error(
                    "Error processing node_dir=%s file=%s (%s: %s)",
                    node_dir, file_path, type(e).__name__, e,
                )
                self._quarantine_event_file(file_path, node_dir, e)

        self._save_checkpoint()
        self._prune_stale_world()
        self._save_world()

    def loop(self, interval=5):
        logger.info("Starting observer loop")
        while True:
            self.run_once()
            time.sleep(interval)

if __name__ == "__main__":
    import sys
    observer = Observer()
    if "--run-once" in sys.argv:
        observer.run_once()
    else:
        observer.loop()