reyna-cli/backup/macmini-automation-baseline-2026-08-03/hermes-scripts/deco_network_watchdog.py

#!/usr/bin/env python3
"""Collect Deco clients, persist snapshots to MongoDB, and maintain local network history.

This script is intended to be run by Hermes cron via the companion shell wrapper
`deco_network_watchdog.sh` every 5 minutes. It is deliberately non-agentic and
prints nothing on success so no_agent cron runs stay silent.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

REYNA_DIR = Path("/home/adolforeyna/reyna-cli")
STATE_DIR = Path.home() / ".hermes" / "deco_network"
LAST_STATE_FILE = STATE_DIR / "last_state.json"
HISTOGRAM_FILE = STATE_DIR / "histogram_1h.json"
DECO_CHANGES_FILE = STATE_DIR / "linked_deco_changes.jsonl"
NEW_DEVICES_FILE = STATE_DIR / "new_devices.jsonl"
FETCH_ERRORS_FILE = STATE_DIR / "fetch_errors.jsonl"
SNAPSHOT_FILE = STATE_DIR / "latest_snapshot.json"
MONGO_DB = os.environ.get("DECO_NETWORK_MONGO_DB", "reyna_home")
MONGO_COLLECTION = os.environ.get("DECO_NETWORK_MONGO_COLLECTION", "deco_client_snapshots")
WINDOW = timedelta(hours=1)
FETCH_TIMEOUT_SECONDS = 45


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def iso(dt: datetime) -> str:
    return dt.isoformat().replace("+00:00", "Z")


def parse_iso(value: str) -> datetime:
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)


def load_json(path: Path, default: Any) -> Any:
    try:
        if path.exists():
            return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        # Keep the collector running even if a local state file is corrupt.
        backup = path.with_suffix(path.suffix + f".corrupt-{int(utc_now().timestamp())}")
        try:
            path.replace(backup)
        except Exception:
            pass
    return default


def atomic_write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(json.dumps(data, indent=2, sort_keys=True, default=str), encoding="utf-8")
    os.replace(tmp, path)


def append_jsonl(path: Path, records: list[dict[str, Any]]) -> None:
    if not records:
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        for record in records:
            handle.write(json.dumps(record, sort_keys=True, default=str) + "\n")


def fetch_deco_clients() -> tuple[list[dict[str, Any]], dict[str, Any]]:
    # The shell wrapper runs this under `uv run python` from REYNA_DIR, so the
    # project dependencies are available. Query Deco directly from reyna-cli
    # instead of depending on the Mac mini MCP endpoint.
    sys.path.insert(0, str(REYNA_DIR / "src"))
    from reyna_cli.deco_direct import DecoDirectClient

    data = DecoDirectClient().list_clients()
    clients = data.get("clients") if isinstance(data, dict) else None
    if not isinstance(clients, list):
        raise RuntimeError("deco clients result did not contain clients list")
    return clients, data


def device_key(client: dict[str, Any]) -> str:
    return str(client.get("mac") or client.get("ip") or client.get("hostname") or "unknown")


def compact_device(client: dict[str, Any]) -> dict[str, Any]:
    keys = [
        "hostname",
        "mac",
        "ip",
        "connection",
        "interface",
        "active",
        "linkedDecoMac",
        "linkedDecoName",
        "linkedDecoRole",
    ]
    return {key: client.get(key) for key in keys if client.get(key) is not None}


def maintain_local_files(now: datetime, clients: list[dict[str, Any]], raw_result: dict[str, Any]) -> None:
    now_s = iso(now)
    current: dict[str, dict[str, Any]] = {device_key(client): compact_device(client) for client in clients}
    last_state = load_json(LAST_STATE_FILE, {})
    if not isinstance(last_state, dict):
        last_state = {}

    deco_changes: list[dict[str, Any]] = []
    new_devices: list[dict[str, Any]] = []
    for key, current_device in current.items():
        previous = last_state.get(key)
        if not previous:
            new_devices.append({"timestamp": now_s, "device_key": key, "device": current_device})
            continue
        old_deco = previous.get("linkedDecoName")
        new_deco = current_device.get("linkedDecoName")
        if old_deco != new_deco:
            deco_changes.append(
                {
                    "timestamp": now_s,
                    "device_key": key,
                    "hostname": current_device.get("hostname"),
                    "mac": current_device.get("mac"),
                    "ip": current_device.get("ip"),
                    "old_linkedDecoName": old_deco,
                    "new_linkedDecoName": new_deco,
                    "old_linkedDecoMac": previous.get("linkedDecoMac"),
                    "new_linkedDecoMac": current_device.get("linkedDecoMac"),
                }
            )

    append_jsonl(DECO_CHANGES_FILE, deco_changes)
    append_jsonl(NEW_DEVICES_FILE, new_devices)
    atomic_write_json(LAST_STATE_FILE, current)

    histogram = load_json(HISTOGRAM_FILE, {"buckets": []})
    buckets = histogram.get("buckets", []) if isinstance(histogram, dict) else []
    cutoff = now - WINDOW
    kept_buckets = []
    for bucket in buckets:
        try:
            if parse_iso(bucket["timestamp"]) >= cutoff:
                kept_buckets.append(bucket)
        except Exception:
            continue

    bucket_devices = sorted(current)
    kept_buckets.append(
        {
            "timestamp": now_s,
            "client_count": len(clients),
            "devices": bucket_devices,
            "linkedDecoCounts": dict(sorted(_counts(c.get("linkedDecoName") for c in current.values()).items())),
        }
    )

    device_summary: dict[str, dict[str, Any]] = {}
    per_device_buckets: dict[str, list[str]] = defaultdict(list)
    for bucket in kept_buckets:
        for key in bucket.get("devices", []):
            per_device_buckets[key].append(bucket.get("timestamp"))
    for key, seen_timestamps in sorted(per_device_buckets.items()):
        device_summary[key] = {
            "seen_count": len(seen_timestamps),
            "first_seen": seen_timestamps[0],
            "last_seen": seen_timestamps[-1],
            "currently_present": key in current,
            "device": current.get(key) or last_state.get(key) or {},
        }

    atomic_write_json(
        HISTOGRAM_FILE,
        {
            "generated_at": now_s,
            "window_minutes": 60,
            "bucket_interval_minutes": 5,
            "bucket_count": len(kept_buckets),
            "current_client_count": len(clients),
            "current_devices": bucket_devices,
            "new_devices_this_run": [record["device"] for record in new_devices],
            "linkedDeco_changes_this_run": deco_changes,
            "devices": device_summary,
            "buckets": kept_buckets,
        },
    )
    atomic_write_json(
        SNAPSHOT_FILE,
        {
            "timestamp": now_s,
            "client_count": len(clients),
            "clients": clients,
            "decos": raw_result.get("decos", []),
        },
    )


def _counts(values: Any) -> dict[str, int]:
    counts: dict[str, int] = defaultdict(int)
    for value in values:
        counts[str(value or "unknown")] += 1
    return counts


def insert_mongo_snapshot(now: datetime, clients: list[dict[str, Any]], raw_result: dict[str, Any]) -> None:
    sys.path.insert(0, str(REYNA_DIR / "src"))
    from reyna_cli.mongo_direct import MongoDirectClient

    now_s = iso(now)
    mongo = MongoDirectClient(timeout_ms=5000)
    collection = mongo.client[MONGO_DB][MONGO_COLLECTION]
    collection.insert_one(
        {
            "timestamp": now,
            "timestamp_iso": now_s,
            "source": "reyna-cli macmini deco clients",
            "client_count": len(clients),
            "clients": clients,
            "decos": raw_result.get("decos", []),
        }
    )


def main() -> int:
    try:
        now = utc_now()
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        try:
            clients, raw_result = fetch_deco_clients()
        except Exception as exc:
            # Treat a temporarily unreachable Mac mini / MCP endpoint as a data
            # collection miss, not as a cron failure. no_agent cron jobs alert on
            # non-zero exits, and Wi-Fi/LAN outages would otherwise spam WhatsApp.
            append_jsonl(
                FETCH_ERRORS_FILE,
                [
                    {
                        "timestamp": iso(now),
                        "stage": "fetch_deco_clients",
                        "error_type": type(exc).__name__,
                        "error": str(exc),
                    }
                ],
            )
            return 0
        insert_mongo_snapshot(now, clients, raw_result)
        maintain_local_files(now, clients, raw_result)
        return 0
    except Exception as exc:
        print(f"deco_network_watchdog error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())