ces_monitoring.py

"""
CES Monitoring — Health context, rotating logger, and HTTP dashboard.

Three monitoring layers:

1. ``health_context()`` — Natural language string describing system state
   for prompt injection (e.g. "NeuroGraph: 1,234 nodes, 89% accuracy").
2. ``CESLogger`` — Rotating file logger to ``~/.neurograph/logs/ces.log``.
3. ``MonitoringDashboard`` — HTTP server (port 8847) with JSON endpoints
   for external monitoring tools.

Usage::

    from ces_monitoring import CESMonitor
    monitor = CESMonitor(ng_memory, ces_config)
    monitor.start()
    print(monitor.health_context())
    monitor.stop()

# ---- Changelog ----
# [2026-05-05] Claude (Sonnet 4.6) — #237 Async stats cache for MonitoringDashboard
# What: Added _StatsCache (write-behind cache, refreshed every health_interval by
#       CESMonitor._health_check_tick). HTTP handlers in _DashboardHandler now serve
#       from the cache instead of calling ng_memory.stats() on each request.
# Why:  _health_data() called ng_memory.stats() synchronously on every /health request.
#       During Tonic ticks the HTTP handler thread blocked (step_lock contention or GIL
#       saturation from the Qwen forward pass). Gateway watchdog got BrokenPipeErrors
#       every 30-60s, generated 15-line tracebacks, kept Node.js event loop at ~80% CPU.
#       Fix: HTTP handlers never block — they serve the last-known snapshot immediately.
# How:  _StatsCache thread-safe wrapper around a dict + timestamp. CESMonitor.get_health()
#       feeds it. MonitoringDashboard receives it at init. _DashboardHandler class attrs
#       updated; _health_data() and _stats_data() serve cache with cache_age_seconds field.
# [2026-02-22] Claude (Opus 4.6) — Initial implementation.
#   What: CESMonitor with health_context, CESLogger with rotating file
#         handler, MonitoringDashboard HTTP server on port 8847.
#   Why:  Provides observability into CES subsystems for both AI
#         (natural language context) and human operators (HTTP + logs).
# -------------------
"""

from __future__ import annotations

import json
import logging
import logging.handlers
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any, Dict, Optional

from ces_config import CESConfig

logger = logging.getLogger("neurograph.ces.monitoring")


# ── Stats cache (write-behind, feeds HTTP handlers) ───────────────────


class _StatsCache:
    """Thread-safe write-behind cache for ng_memory.stats() snapshots.

    Refreshed by CESMonitor._health_check_tick every health_interval seconds.
    HTTP handlers serve from this cache instantly without blocking on the GIL
    or step_lock during Tonic ticks.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._data: Dict[str, Any] = {}
        self._updated_at: float = 0.0

    def set(self, stats: Dict[str, Any]) -> None:
        with self._lock:
            self._data = dict(stats)
            self._updated_at = time.time()

    def get(self) -> tuple:
        """Return (stats_dict, age_seconds). age is -1 if never populated."""
        with self._lock:
            age = (time.time() - self._updated_at) if self._updated_at else -1.0
            return dict(self._data), age


# ── Health context (Layer 1) ───────────────────────────────────────────


def health_context(ng_memory: Any) -> str:
    """Generate a natural language health summary for prompt injection.

    Args:
        ng_memory: ``NeuroGraphMemory`` instance.

    Returns:
        Human-readable status string.
    """
    try:
        stats = ng_memory.stats()
        parts = [
            f"NeuroGraph: {stats.get('nodes', 0):,} nodes",
            f"{stats.get('synapses', 0):,} synapses",
        ]

        accuracy = stats.get("prediction_accuracy", 0)
        if accuracy > 0:
            parts.append(f"{accuracy:.0%} prediction accuracy")

        # CES subsystem status
        ces = stats.get("ces", {})
        sp = ces.get("stream_parser")
        if sp:
            if sp.get("is_running"):
                parts.append("stream parser active")
            else:
                parts.append("stream parser paused")

        surf = ces.get("surfacing")
        if surf:
            depth = surf.get("queue_depth", 0)
            if depth > 0:
                parts.append(f"{depth} concepts surfaced")

        return ", ".join(parts)
    except Exception as exc:
        return f"NeuroGraph: status unavailable ({exc})"


# ── Rotating logger (Layer 2) ─────────────────────────────────────────


class CESLogger:
    """Rotating file logger for CES events.

    Writes structured JSON-line events to ``ces.log`` with automatic
    rotation based on file size.

    Args:
        ces_config: ``CESConfig`` with monitoring parameters.
    """

    def __init__(self, ces_config: CESConfig) -> None:
        self._cfg = ces_config.monitoring
        self._logger = logging.getLogger("neurograph.ces.events")
        self._setup_handler()

    def _setup_handler(self) -> None:
        """Configure rotating file handler."""
        log_dir = Path(self._cfg.log_dir).expanduser()
        log_dir.mkdir(parents=True, exist_ok=True)
        log_path = log_dir / "ces.log"

        handler = logging.handlers.RotatingFileHandler(
            str(log_path),
            maxBytes=self._cfg.max_log_size_mb * 1024 * 1024,
            backupCount=self._cfg.backup_count,
        )
        handler.setFormatter(logging.Formatter("%(message)s"))
        self._logger.addHandler(handler)
        self._logger.setLevel(logging.INFO)

    def log_event(self, event_type: str, data: Dict[str, Any]) -> None:
        """Write a structured event to the CES log."""
        event = {
            "timestamp": time.time(),
            "event": event_type,
            "data": data,
        }
        self._logger.info(json.dumps(event, default=str))


# ── HTTP dashboard (Layer 3) ──────────────────────────────────────────


class _DashboardHandler(BaseHTTPRequestHandler):
    """HTTP request handler for the monitoring dashboard."""

    # Set by MonitoringDashboard before server starts
    ng_memory: Any = None
    ces_monitor: Any = None
    stats_cache: Optional["_StatsCache"] = None

    def do_GET(self) -> None:
        """Handle GET requests."""
        if self.path == "/health":
            self._json_response(self._health_data())
        elif self.path == "/stats":
            self._json_response(self._stats_data())
        elif self.path == "/surfaced":
            self._json_response(self._surfaced_data())
        else:
            self.send_error(404, "Not Found")

    def _health_data(self) -> Dict[str, Any]:
        """Minimal health check — served from cache, never blocks."""
        result: Dict[str, Any] = {"status": "ok", "timestamp": time.time()}
        if self.stats_cache is not None:
            cached, age = self.stats_cache.get()
            result["cache_age_seconds"] = round(age, 1)
            if cached:
                result["nodes"] = cached.get("nodes", 0)
                result["synapses"] = cached.get("synapses", 0)
            else:
                result["status"] = "initializing"
        return result

    def _stats_data(self) -> Dict[str, Any]:
        """Full telemetry — served from cache, never blocks."""
        if self.stats_cache is not None:
            cached, age = self.stats_cache.get()
            if cached:
                cached["cache_age_seconds"] = round(age, 1)
                return cached
            return {"status": "initializing", "cache_age_seconds": -1}
        return {"error": "not initialized"}

    def _surfaced_data(self) -> Dict[str, Any]:
        """Current surfaced concepts."""
        if self.ces_monitor is not None and self.ces_monitor._surfacing_monitor is not None:
            try:
                items = self.ces_monitor._surfacing_monitor.get_surfaced()
                return {"surfaced": items, "count": len(items)}
            except Exception as exc:
                return {"error": str(exc)}
        return {"surfaced": [], "count": 0}

    def _json_response(self, data: Dict[str, Any]) -> None:
        """Send a JSON response."""
        body = json.dumps(data, default=str).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: Any) -> None:
        """Suppress default stderr logging."""
        pass


class MonitoringDashboard:
    """HTTP monitoring server running in a daemon thread.

    Args:
        ces_config: ``CESConfig`` with monitoring parameters.
        ng_memory: ``NeuroGraphMemory`` instance for stats.
        ces_monitor: ``CESMonitor`` parent for surfacing access.
    """

    def __init__(
        self,
        ces_config: CESConfig,
        ng_memory: Any = None,
        ces_monitor: Any = None,
        stats_cache: Optional[_StatsCache] = None,
    ) -> None:
        self._cfg = ces_config.monitoring
        self._server: Optional[HTTPServer] = None
        self._thread: Optional[threading.Thread] = None

        # Share refs with the handler class
        _DashboardHandler.ng_memory = ng_memory
        _DashboardHandler.ces_monitor = ces_monitor
        _DashboardHandler.stats_cache = stats_cache

    def start(self) -> None:
        """Start the HTTP server in a daemon thread."""
        if not self._cfg.http_enabled:
            logger.info("HTTP dashboard disabled by config")
            return

        try:
            self._server = HTTPServer(
                ("0.0.0.0", self._cfg.http_port), _DashboardHandler
            )
            self._thread = threading.Thread(
                target=self._server.serve_forever,
                daemon=True,
                name="ces-dashboard",
            )
            self._thread.start()
            logger.info("CES dashboard started on port %d", self._cfg.http_port)
        except Exception as exc:
            logger.warning("Failed to start CES dashboard: %s", exc)
            self._server = None

    def stop(self) -> None:
        """Shut down the HTTP server."""
        if self._server is not None:
            self._server.shutdown()
            self._server.server_close()  # Release the socket
            self._server = None
        if self._thread is not None:
            self._thread.join(timeout=5.0)
            self._thread = None

    @property
    def is_running(self) -> bool:
        """True if the server thread is alive."""
        return self._thread is not None and self._thread.is_alive()


# ── CESMonitor coordinator ────────────────────────────────────────────


class CESMonitor:
    """Coordinator for all CES monitoring infrastructure.

    Args:
        ng_memory: ``NeuroGraphMemory`` instance.
        ces_config: ``CESConfig`` with monitoring parameters.
    """

    def __init__(self, ng_memory: Any, ces_config: CESConfig) -> None:
        self._ng_memory = ng_memory
        self._cfg = ces_config
        self._ces_logger = CESLogger(ces_config)
        self._stats_cache = _StatsCache()
        self._dashboard = MonitoringDashboard(
            ces_config, ng_memory=ng_memory, ces_monitor=self,
            stats_cache=self._stats_cache,
        )

        # Access to surfacing monitor (set by openclaw_hook after init)
        self._surfacing_monitor: Any = None

        # Health check timer
        self._health_timer: Optional[threading.Timer] = None

    def start(self) -> None:
        """Start the dashboard and periodic health logging."""
        self._dashboard.start()
        self._schedule_health_check()

    def stop(self) -> None:
        """Stop all monitoring."""
        self._dashboard.stop()
        if self._health_timer is not None:
            self._health_timer.cancel()
            self._health_timer = None

    def get_health(self) -> Dict[str, Any]:
        """Return comprehensive health status and refresh the stats cache.

        # ---- Changelog ----
        # [2026-05-31] Claude Code (Opus 4.7, 1M) — Fix infinite mutual recursion
        # What: Read substrate directly (graph.nodes, graph.synapses, telemetry, vector_db.count)
        #       instead of calling self._ng_memory.stats(). Also rebuilt the stats cache
        #       payload locally from the same direct reads.
        # Why:  openclaw_hook.stats() builds its "monitor" key by calling self._ces_monitor.get_health(),
        #       which previously called back into stats() — INFINITE MUTUAL RECURSION. Each /health
        #       request to the CES dashboard (port 8847) pegged a whole CPU core indefinitely.
        #       Root cause of the "VPS 100% CPU" incident 2026-05-31 (misdiagnosed earlier as GSG
        #       Phase 4 / synaptic-delay perf regression).
        # How:  Replace `self._ng_memory.stats()` call with direct attribute reads — same data,
        #       no recursion path. Cache payload reduced to the 4 fields actually consumed.
        # Discovered via py-spy dump showing stats → get_health → stats → get_health stack alternation.
        # -------------------
        """
        result: Dict[str, Any] = {
            "timestamp": time.time(),
            "dashboard_running": self._dashboard.is_running,
            "dashboard_port": self._cfg.monitoring.http_port,
        }

        try:
            # Read substrate directly — calling self._ng_memory.stats() causes mutual
            # recursion with openclaw_hook.stats() (its "monitor" key calls back here).
            graph = self._ng_memory.graph
            nodes_cnt = len(graph.nodes)
            syn_cnt = len(graph.synapses)
            try:
                tel = graph.get_telemetry()
                pred_acc = round(tel.prediction_accuracy, 4)
            except Exception:
                pred_acc = 0
            try:
                vdb_cnt = self._ng_memory.vector_db.count()
            except Exception:
                vdb_cnt = 0

            result["nodes"] = nodes_cnt
            result["synapses"] = syn_cnt
            result["prediction_accuracy"] = pred_acc
            result["vector_db_count"] = vdb_cnt

            # Refresh stats cache with the 4 fields actually consumed (no recursion path).
            self._stats_cache.set({
                "nodes": nodes_cnt,
                "synapses": syn_cnt,
                "prediction_accuracy": pred_acc,
                "vector_db_count": vdb_cnt,
            })
        except Exception:
            result["status"] = "stats_unavailable"

        return result

    def health_context(self) -> str:
        """Generate a natural language health string."""
        return health_context(self._ng_memory)

    def log_event(self, event_type: str, data: Dict[str, Any]) -> None:
        """Write an event to the CES rotating log."""
        self._ces_logger.log_event(event_type, data)

    # ── Internal ───────────────────────────────────────────────────────

    def _schedule_health_check(self) -> None:
        """Schedule periodic health check logging."""
        self._health_timer = threading.Timer(
            self._cfg.monitoring.health_interval,
            self._health_check_tick,
        )
        self._health_timer.daemon = True
        self._health_timer.start()

    def _health_check_tick(self) -> None:
        """Periodic health check callback."""
        # Guard against orphan timer chains after stop() sets _health_timer
        # to None — without this, a pending callback can reschedule itself
        # indefinitely.
        if self._health_timer is None:
            return
        try:
            health = self.get_health()
            self._ces_logger.log_event("health_check", health)
        except Exception as exc:
            logger.debug("Health check failed: %s", exc)
        self._schedule_health_check()