diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py index bdff680..7e07715 100644 --- a/meshchatx/meshchat.py +++ b/meshchatx/meshchat.py @@ -88,7 +88,7 @@ from meshchatx.src.backend.nomadnet_utils import ( convert_nomadnet_string_data_to_map, ) from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler -from meshchatx.src.backend.recovery import CrashRecovery +from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor from meshchatx.src.backend.rnprobe_handler import RNProbeHandler from meshchatx.src.backend.sideband_commands import SidebandCommands from meshchatx.src.backend.telemetry_utils import Telemeter @@ -744,6 +744,19 @@ class ReticulumMeshChat: # Link database to memory log handler memory_log_handler.set_database(context.database) + # Wire crash recovery with DB + log handler for adaptive diagnostics + if hasattr(self, "_crash_recovery") and self._crash_recovery: + self._crash_recovery.set_database(context.database) + self._crash_recovery.log_handler = memory_log_handler + + # Start health monitor if not already running + if not hasattr(self, "_health_monitor") or self._health_monitor is None: + self._health_monitor = HealthMonitor( + log_handler=memory_log_handler, + app=self, + ) + self._health_monitor.start() + def _checkpoint_and_close(self): # delegated to database instance self.database._checkpoint_and_close() @@ -12419,6 +12432,9 @@ def main(): docs_download_urls=args.docs_download_urls, ) + # store recovery on app for wiring with identity context + reticulum_meshchat._crash_recovery = recovery + # update recovery with known paths recovery.update_paths( storage_dir=reticulum_meshchat.storage_dir, diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py index 5d37e94..2ce37bf 100644 --- a/meshchatx/src/backend/database/__init__.py +++ b/meshchatx/src/backend/database/__init__.py @@ -8,6 +8,7 @@ from datetime import UTC, datetime from .announces import AnnounceDAO from .config import ConfigDAO from .contacts import ContactsDAO +from .crash_history import CrashHistoryDAO from .debug_logs import DebugLogsDAO from .legacy_migrator import LegacyMigrator from .map_drawings import MapDrawingsDAO @@ -41,6 +42,7 @@ class Database: self.contacts = ContactsDAO(self.provider) self.map_drawings = MapDrawingsDAO(self.provider) self.debug_logs = DebugLogsDAO(self.provider) + self.crash_history = CrashHistoryDAO(self.provider) def initialize(self): self._tune_sqlite_pragmas() diff --git a/meshchatx/src/backend/database/crash_history.py b/meshchatx/src/backend/database/crash_history.py new file mode 100644 index 0000000..1bd5b66 --- /dev/null +++ b/meshchatx/src/backend/database/crash_history.py @@ -0,0 +1,65 @@ +import json + +from .provider import DatabaseProvider + + +class CrashHistoryDAO: + def __init__(self, provider: DatabaseProvider): + self.provider = provider + + def insert_crash( + self, + timestamp, + error_type, + error_message, + diagnosed_cause, + symptoms, + probability, + entropy, + divergence, + ): + self.provider.execute( + """ + INSERT INTO crash_history + (timestamp, error_type, error_message, diagnosed_cause, + symptoms, probability, entropy, divergence) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + timestamp, + error_type, + error_message[:500] if error_message else None, + diagnosed_cause, + json.dumps(symptoms) if isinstance(symptoms, dict) else symptoms, + probability, + entropy, + divergence, + ), + ) + + def get_recent_crashes(self, limit=50): + return self.provider.fetchall( + "SELECT * FROM crash_history ORDER BY timestamp DESC LIMIT ?", + (limit,), + ) + + def get_cause_frequencies(self, limit=50): + return self.provider.fetchall( + """ + SELECT diagnosed_cause, COUNT(*) as count + FROM (SELECT * FROM crash_history ORDER BY timestamp DESC LIMIT ?) + GROUP BY diagnosed_cause + ORDER BY count DESC + """, + (limit,), + ) + + def cleanup_old(self, max_entries=200): + self.provider.execute( + """ + DELETE FROM crash_history WHERE id NOT IN ( + SELECT id FROM crash_history ORDER BY timestamp DESC LIMIT ? + ) + """, + (max_entries,), + ) diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py index eebca3f..38e8bef 100644 --- a/meshchatx/src/backend/database/schema.py +++ b/meshchatx/src/backend/database/schema.py @@ -13,7 +13,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str: class DatabaseSchema: - LATEST_VERSION = 39 + LATEST_VERSION = 40 def __init__(self, provider: DatabaseProvider): self.provider = provider @@ -1046,6 +1046,25 @@ class DatabaseSchema: "CREATE INDEX IF NOT EXISTS idx_lxmf_messages_state_peer ON lxmf_messages(state, peer_hash)", ) + if current_version < 40: + self._safe_execute(""" + CREATE TABLE IF NOT EXISTS crash_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL, + error_type TEXT, + error_message TEXT, + diagnosed_cause TEXT, + symptoms TEXT, + probability INTEGER, + entropy REAL, + divergence REAL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """) + self._safe_execute( + "CREATE INDEX IF NOT EXISTS idx_crash_history_timestamp ON crash_history(timestamp)", + ) + # Update version in config self._safe_execute( """ diff --git a/meshchatx/src/backend/persistent_log_handler.py b/meshchatx/src/backend/persistent_log_handler.py index 3dadde3..6260d63 100644 --- a/meshchatx/src/backend/persistent_log_handler.py +++ b/meshchatx/src/backend/persistent_log_handler.py @@ -1,10 +1,13 @@ import collections import logging +import math import re import threading import time from datetime import UTC, datetime +_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") + class PersistentLogHandler(logging.Handler): def __init__(self, database=None, capacity=5000, flush_interval=5): @@ -27,6 +30,10 @@ class PersistentLogHandler(logging.Handler): self.known_ips = set() self.known_uas = set() + # Entropy tracking — level counts over a sliding 60s window + self._level_events = collections.deque(maxlen=10000) + self._error_events = collections.deque(maxlen=10000) + def set_database(self, database): with self.lock: self.database = database @@ -49,6 +56,10 @@ class PersistentLogHandler(logging.Handler): with self.lock: self.logs_buffer.append(log_entry) + now_mono = time.monotonic() + self._level_events.append((now_mono, record.levelname)) + if record.levelno >= logging.ERROR: + self._error_events.append(now_mono) # Periodically flush to database if available if self.database and ( @@ -233,3 +244,34 @@ class PersistentLogHandler(logging.Handler): is_anomaly=is_anomaly, ) return len(self.logs_buffer) + + @property + def current_log_entropy(self): + """Shannon entropy over log-level distribution in the last 60 seconds.""" + cutoff = time.monotonic() - 60.0 + with self.lock: + counts = {lv: 0 for lv in _LOG_LEVELS} + total = 0 + for ts, level in self._level_events: + if ts >= cutoff: + counts[level] = counts.get(level, 0) + 1 + total += 1 + if total == 0: + return 0.0 + entropy = 0.0 + for c in counts.values(): + if c > 0: + p = c / total + entropy -= p * math.log2(p) + return entropy + + @property + def current_error_rate(self): + """Fraction of log events at ERROR or above in the last 60 seconds.""" + cutoff = time.monotonic() - 60.0 + with self.lock: + total = sum(1 for ts, _ in self._level_events if ts >= cutoff) + errors = sum(1 for ts in self._error_events if ts >= cutoff) + if total == 0: + return 0.0 + return errors / total diff --git a/meshchatx/src/backend/recovery/__init__.py b/meshchatx/src/backend/recovery/__init__.py index 5992cca..4a36ac2 100644 --- a/meshchatx/src/backend/recovery/__init__.py +++ b/meshchatx/src/backend/recovery/__init__.py @@ -1,3 +1,4 @@ from .crash_recovery import CrashRecovery +from .health_monitor import HealthMonitor -__all__ = ["CrashRecovery"] +__all__ = ["CrashRecovery", "HealthMonitor"] diff --git a/meshchatx/src/backend/recovery/crash_recovery.py b/meshchatx/src/backend/recovery/crash_recovery.py index 7cda9f8..709c46e 100644 --- a/meshchatx/src/backend/recovery/crash_recovery.py +++ b/meshchatx/src/backend/recovery/crash_recovery.py @@ -1,26 +1,45 @@ -"""CRASH RECOVERY & DIAGNOSTIC ENGINE ------------------------------------------- -This module implements a mathematically grounded diagnostic system for MeshChatX. -It utilizes Active Inference heuristics, Shannon Entropy, and KL-Divergence +"""CRASH RECOVERY & ADAPTIVE DIAGNOSTIC ENGINE +-------------------------------------------------- +Mathematically grounded diagnostic system for MeshChatX. + +Uses Shannon Entropy, KL-Divergence, and Bayesian weight learning to map application failures onto deterministic manifold constraints. +Crash history is persisted and priors are refined over time using +a conjugate Beta-Binomial model. """ import contextlib +import json import os import platform import re import shutil import sqlite3 import sys +import time import traceback import psutil import RNS +_DEFAULT_PRIORS = { + "DB_SYNC_FAILURE": 0.05, + "DB_CORRUPTION": 0.05, + "ASYNC_RACE": 0.10, + "OOM": 0.02, + "CONFIG_MISSING": 0.01, + "RNS_IDENTITY_FAILURE": 0.05, + "LXMF_STORAGE_FAILURE": 0.05, + "INTERFACE_OFFLINE": 0.05, + "UNSUPPORTED_PYTHON": 0.05, + "LEGACY_SYSTEM_LIMITATION": 0.05, +} + class CrashRecovery: """A diagnostic utility that intercepts application crashes and provides - meaningful error reports and system state analysis. + meaningful error reports and system state analysis. Learns from crash + history to refine root-cause probabilities over time. """ def __init__( @@ -29,12 +48,17 @@ class CrashRecovery: database_path=None, public_dir=None, reticulum_config_dir=None, + database=None, + log_handler=None, ): self.storage_dir = storage_dir self.database_path = database_path self.public_dir = public_dir self.reticulum_config_dir = reticulum_config_dir + self.database = database + self.log_handler = log_handler self.enabled = True + self._learned_priors = None # Check environment variable to allow disabling the recovery system env_val = os.environ.get("MESHCHAT_NO_CRASH_RECOVERY", "").lower() @@ -69,6 +93,95 @@ class CrashRecovery: if reticulum_config_dir: self.reticulum_config_dir = reticulum_config_dir + def set_database(self, database): + """Provide database access for crash persistence and weight learning.""" + self.database = database + self._load_learned_priors() + + def _load_learned_priors(self): + """Load learned Bayesian priors from the config table.""" + if not self.database: + return + try: + raw = self.database.config.get("diagnostic_weights") + if raw: + self._learned_priors = json.loads(raw) + except Exception: + self._learned_priors = None + + def _get_prior(self, cause_key): + """Return the learned prior for a cause, falling back to the hardcoded default.""" + if self._learned_priors and cause_key in self._learned_priors: + return self._learned_priors[cause_key] + return _DEFAULT_PRIORS.get(cause_key, 0.05) + + def _persist_crash( + self, error_type, error_msg, causes, symptoms, entropy, divergence + ): + """Store crash event in crash_history for future learning.""" + if not self.database: + return + try: + top_cause = causes[0]["description"] if causes else "Unknown" + top_prob = causes[0]["probability"] if causes else 0 + self.database.crash_history.insert_crash( + timestamp=time.time(), + error_type=error_type, + error_message=error_msg, + diagnosed_cause=top_cause, + symptoms=symptoms, + probability=top_prob, + entropy=entropy, + divergence=divergence, + ) + self.database.crash_history.cleanup_old(max_entries=200) + except Exception: + pass + + def _update_learned_weights(self): + """Bayesian weight update using Beta-Binomial conjugate model.""" + if not self.database: + return + try: + freq_rows = self.database.crash_history.get_cause_frequencies(limit=50) + if not freq_rows: + return + total = sum(r["count"] for r in freq_rows) + if total < 3: + return + + cause_counts = {r["diagnosed_cause"]: r["count"] for r in freq_rows} + weights = {} + for key, default_prior in _DEFAULT_PRIORS.items(): + desc = self._cause_key_to_description(key) + count = cause_counts.get(desc, 0) + alpha = 1.0 + count + beta = 1.0 + (total - count) + posterior = alpha / (alpha + beta) + weights[key] = max(0.01, min(0.99, round(posterior, 4))) + + self.database.config.set("diagnostic_weights", json.dumps(weights)) + self._learned_priors = weights + except Exception: + pass + + @staticmethod + def _cause_key_to_description(key): + """Map internal cause keys to their human-readable descriptions.""" + mapping = { + "DB_SYNC_FAILURE": "In-Memory Database Sync Failure", + "DB_CORRUPTION": "SQLite Database Corruption", + "ASYNC_RACE": "Asynchronous Initialization Race Condition", + "OOM": "System Resource Exhaustion (OOM)", + "CONFIG_MISSING": "Missing Reticulum Configuration", + "RNS_IDENTITY_FAILURE": "Reticulum Identity Load Failure", + "LXMF_STORAGE_FAILURE": "LXMF Router Storage Failure", + "INTERFACE_OFFLINE": "Reticulum Interface Initialization Failure", + "UNSUPPORTED_PYTHON": "Unsupported Python Environment", + "LEGACY_SYSTEM_LIMITATION": "Legacy System Resource Limitation", + } + return mapping.get(key, key) + def handle_exception(self, exc_type, exc_value, exc_traceback): """Intercepts unhandled exceptions to provide a detailed diagnosis report.""" # Let keyboard interrupts pass through normally @@ -113,6 +226,15 @@ class CrashRecovery: out.write(f" [Manifold Curvature: {curvature:.2f}κ]\n") out.write(" [Deterministic Manifold Constraints: V1,V4 Active]\n") + if self.log_handler: + log_ent = self.log_handler.current_log_entropy + err_rate = self.log_handler.current_error_rate + out.write(f" [Log Entropy (60s): {log_ent:.4f} bits]\n") + out.write(f" [Error Rate (60s): {err_rate:.2%}]\n") + + if self._learned_priors: + out.write(" [Bayesian Priors: Learned from crash history]\n") + for cause in causes: out.write( f" - [{cause['probability']}% Probability] {cause['description']}\n", @@ -146,6 +268,11 @@ class CrashRecovery: out.write("=" * 70 + "\n\n") out.flush() + # Persist crash and update weights (best-effort, never raise) + with contextlib.suppress(Exception): + self._persist_crash(error_type, error_msg, causes, {}, entropy, divergence) + self._update_learned_weights() + # Exit with error code sys.exit(1) @@ -157,10 +284,10 @@ class CrashRecovery: error_msg = str(exc_value).lower() error_type = exc_type.__name__.lower() - # Define potential root causes with prior probabilities + # Define potential root causes with prior probabilities (learned or default) potential_causes = { "DB_SYNC_FAILURE": { - "probability": 0.05, + "probability": self._get_prior("DB_SYNC_FAILURE"), "description": "In-Memory Database Sync Failure", "reasoning": "A background thread attempted to access an in-memory database that was not initialized in its local context.", "suggestions": [ @@ -169,7 +296,7 @@ class CrashRecovery: ], }, "DB_CORRUPTION": { - "probability": 0.05, + "probability": self._get_prior("DB_CORRUPTION"), "description": "SQLite Database Corruption", "reasoning": "The database file on disk has become physically or logically corrupted.", "suggestions": [ @@ -178,7 +305,7 @@ class CrashRecovery: ], }, "ASYNC_RACE": { - "probability": 0.10, + "probability": self._get_prior("ASYNC_RACE"), "description": "Asynchronous Initialization Race Condition", "reasoning": "A component tried to access the asyncio event loop before it was started.", "suggestions": [ @@ -187,7 +314,7 @@ class CrashRecovery: ], }, "OOM": { - "probability": 0.02, + "probability": self._get_prior("OOM"), "description": "System Resource Exhaustion (OOM)", "reasoning": "Available system memory is extremely low, leading to allocation failures.", "suggestions": [ @@ -196,7 +323,7 @@ class CrashRecovery: ], }, "CONFIG_MISSING": { - "probability": 0.01, + "probability": self._get_prior("CONFIG_MISSING"), "description": "Missing Reticulum Configuration", "reasoning": "The Reticulum Network Stack (RNS) could not find its configuration file.", "suggestions": [ @@ -204,7 +331,7 @@ class CrashRecovery: ], }, "RNS_IDENTITY_FAILURE": { - "probability": 0.05, + "probability": self._get_prior("RNS_IDENTITY_FAILURE"), "description": "Reticulum Identity Load Failure", "reasoning": "The Reticulum identity file is missing, corrupt, or unreadable.", "suggestions": [ @@ -213,7 +340,7 @@ class CrashRecovery: ], }, "LXMF_STORAGE_FAILURE": { - "probability": 0.05, + "probability": self._get_prior("LXMF_STORAGE_FAILURE"), "description": "LXMF Router Storage Failure", "reasoning": "The LXMF router could not access its message storage directory.", "suggestions": [ @@ -222,7 +349,7 @@ class CrashRecovery: ], }, "INTERFACE_OFFLINE": { - "probability": 0.05, + "probability": self._get_prior("INTERFACE_OFFLINE"), "description": "Reticulum Interface Initialization Failure", "reasoning": "No active communication interfaces could be established.", "suggestions": [ @@ -231,7 +358,7 @@ class CrashRecovery: ], }, "UNSUPPORTED_PYTHON": { - "probability": 0.05, + "probability": self._get_prior("UNSUPPORTED_PYTHON"), "description": "Unsupported Python Environment", "reasoning": "The application is running on an outdated or incompatible Python version.", "suggestions": [ @@ -240,7 +367,7 @@ class CrashRecovery: ], }, "LEGACY_SYSTEM_LIMITATION": { - "probability": 0.05, + "probability": self._get_prior("LEGACY_SYSTEM_LIMITATION"), "description": "Legacy System Resource Limitation", "reasoning": "The host system lacks modern kernel features or resource allocation capabilities required for high-performance mesh networking.", "suggestions": [ diff --git a/meshchatx/src/backend/recovery/health_monitor.py b/meshchatx/src/backend/recovery/health_monitor.py new file mode 100644 index 0000000..748143a --- /dev/null +++ b/meshchatx/src/backend/recovery/health_monitor.py @@ -0,0 +1,156 @@ +"""Lightweight predictive health monitor. + +Runs as a daemon thread on a 5-minute interval, reading in-memory +metrics (log entropy, error rate, memory) and emitting WebSocket +warnings when anomalous trends are detected. + +No database queries are made in the monitor loop — all reads come +from in-memory deques kept by PersistentLogHandler and psutil. +""" + +import asyncio +import collections +import json +import logging +import threading + +import psutil + +_log = logging.getLogger("meshchatx.health_monitor") + + +class HealthMonitor: + CHECK_INTERVAL = 300 # 5 minutes + ENTROPY_WINDOW = 6 # readings kept (~30 min at default interval) + ENTROPY_WARN_THRESHOLD = 1.5 # out of max ~2.32 for 5 log levels + ERROR_RATE_WARN = 0.3 + MEMORY_WARN_MB = 100 # warn when available < 100 MB + CONSECUTIVE_NEEDED = 2 # consecutive bad readings before alert + + def __init__(self, log_handler, app=None): + self.log_handler = log_handler + self.app = app + self._running = False + self._thread = None + + self._entropy_history = collections.deque(maxlen=self.ENTROPY_WINDOW) + self._error_rate_history = collections.deque(maxlen=self.ENTROPY_WINDOW) + self._mem_available_history = collections.deque(maxlen=self.ENTROPY_WINDOW) + + def start(self): + if self._running: + return + self._running = True + self._thread = threading.Thread(target=self._run_loop, daemon=True) + self._thread.start() + _log.info("HealthMonitor started (interval=%ds)", self.CHECK_INTERVAL) + + def stop(self): + self._running = False + + def _run_loop(self): + asyncio.run(self._async_loop()) + + async def _async_loop(self): + while self._running: + try: + self._check() + except Exception as exc: + _log.debug("HealthMonitor check error: %s", exc) + await asyncio.sleep(self.CHECK_INTERVAL) + + def _check(self): + entropy = 0.0 + error_rate = 0.0 + if self.log_handler: + entropy = self.log_handler.current_log_entropy + error_rate = self.log_handler.current_error_rate + + mem = psutil.virtual_memory() + available_mb = mem.available / (1024 * 1024) + + self._entropy_history.append(entropy) + self._error_rate_history.append(error_rate) + self._mem_available_history.append(available_mb) + + warnings = [] + + if self._detect_entropy_climb(): + warnings.append( + { + "kind": "entropy_climbing", + "message": f"Log entropy rising: {entropy:.2f} bits (threshold {self.ENTROPY_WARN_THRESHOLD})", + "value": round(entropy, 4), + } + ) + + if self._consecutive_above(self._error_rate_history, self.ERROR_RATE_WARN): + warnings.append( + { + "kind": "error_rate_high", + "message": f"Error rate elevated: {error_rate:.0%}", + "value": round(error_rate, 4), + } + ) + + if self._consecutive_below(self._mem_available_history, self.MEMORY_WARN_MB): + warnings.append( + { + "kind": "memory_low", + "message": f"Available memory low: {available_mb:.0f} MB", + "value": round(available_mb, 1), + } + ) + + for w in warnings: + _log.warning("Health warning: %s", w["message"]) + self._broadcast(w) + + def _detect_entropy_climb(self): + """True if 3+ consecutive entropy readings are strictly increasing + AND the latest exceeds the warning threshold.""" + h = self._entropy_history + if len(h) < 3: + return False + vals = list(h) + last_three = vals[-3:] + if last_three[-1] <= self.ENTROPY_WARN_THRESHOLD: + return False + return last_three[0] < last_three[1] < last_three[2] + + @staticmethod + def _consecutive_above(deq, threshold): + if len(deq) < 2: + return False + vals = list(deq) + return vals[-1] > threshold and vals[-2] > threshold + + @staticmethod + def _consecutive_below(deq, threshold): + if len(deq) < 2: + return False + vals = list(deq) + return vals[-1] < threshold and vals[-2] < threshold + + def _broadcast(self, warning_data): + if not self.app: + return + try: + from meshchatx.src.backend.async_utils import AsyncUtils + + AsyncUtils.run_async( + self.app.websocket_broadcast( + json.dumps({"type": "health_warning", "data": warning_data}) + ) + ) + except Exception: + pass + + @property + def latest_snapshot(self): + """Return latest metrics as a dict (useful for diagnostics/tests).""" + return { + "entropy": list(self._entropy_history), + "error_rate": list(self._error_rate_history), + "memory_mb": list(self._mem_available_history), + } diff --git a/tests/backend/test_crash_history.py b/tests/backend/test_crash_history.py new file mode 100644 index 0000000..db59ff7 --- /dev/null +++ b/tests/backend/test_crash_history.py @@ -0,0 +1,247 @@ +import json +import os +import shutil +import tempfile +import time +import unittest + +from meshchatx.src.backend.database import Database +from meshchatx.src.backend.recovery.crash_recovery import CrashRecovery, _DEFAULT_PRIORS + + +class TestCrashHistoryDAO(unittest.TestCase): + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.db_path = os.path.join(self.test_dir, "test_crash.db") + self.db = Database(self.db_path) + self.db.initialize() + + def tearDown(self): + self.db.close_all() + shutil.rmtree(self.test_dir) + + def test_insert_and_retrieve(self): + self.db.crash_history.insert_crash( + timestamp=time.time(), + error_type="ValueError", + error_message="test error", + diagnosed_cause="DB_CORRUPTION", + symptoms={"sqlite_in_msg": True}, + probability=85, + entropy=1.234, + divergence=0.567, + ) + rows = self.db.crash_history.get_recent_crashes(limit=10) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["error_type"], "ValueError") + self.assertEqual(rows[0]["diagnosed_cause"], "DB_CORRUPTION") + self.assertEqual(rows[0]["probability"], 85) + + def test_get_cause_frequencies(self): + now = time.time() + for i in range(5): + self.db.crash_history.insert_crash( + timestamp=now + i, + error_type="E", + error_message="m", + diagnosed_cause="DB_CORRUPTION", + symptoms={}, + probability=50, + entropy=0.0, + divergence=0.0, + ) + for i in range(3): + self.db.crash_history.insert_crash( + timestamp=now + 10 + i, + error_type="E", + error_message="m", + diagnosed_cause="OOM", + symptoms={}, + probability=30, + entropy=0.0, + divergence=0.0, + ) + freqs = self.db.crash_history.get_cause_frequencies(limit=50) + freq_map = {r["diagnosed_cause"]: r["count"] for r in freqs} + self.assertEqual(freq_map["DB_CORRUPTION"], 5) + self.assertEqual(freq_map["OOM"], 3) + + def test_cleanup_old(self): + now = time.time() + for i in range(20): + self.db.crash_history.insert_crash( + timestamp=now + i, + error_type="E", + error_message="m", + diagnosed_cause="X", + symptoms={}, + probability=0, + entropy=0.0, + divergence=0.0, + ) + self.db.crash_history.cleanup_old(max_entries=5) + rows = self.db.crash_history.get_recent_crashes(limit=100) + self.assertEqual(len(rows), 5) + + def test_insert_long_error_message_truncated(self): + long_msg = "x" * 1000 + self.db.crash_history.insert_crash( + timestamp=time.time(), + error_type="E", + error_message=long_msg, + diagnosed_cause="X", + symptoms={}, + probability=0, + entropy=0.0, + divergence=0.0, + ) + rows = self.db.crash_history.get_recent_crashes() + self.assertLessEqual(len(rows[0]["error_message"]), 500) + + def test_symptoms_stored_as_json(self): + syms = {"sqlite_in_msg": True, "low_mem": False} + self.db.crash_history.insert_crash( + timestamp=time.time(), + error_type="E", + error_message="m", + diagnosed_cause="X", + symptoms=syms, + probability=0, + entropy=0.0, + divergence=0.0, + ) + rows = self.db.crash_history.get_recent_crashes() + loaded = json.loads(rows[0]["symptoms"]) + self.assertEqual(loaded, syms) + + +class TestBayesianWeightLearning(unittest.TestCase): + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.db_path = os.path.join(self.test_dir, "test_bayes.db") + self.db = Database(self.db_path) + self.db.initialize() + self.recovery = CrashRecovery( + storage_dir=self.test_dir, + database_path=self.db_path, + database=self.db, + ) + + def tearDown(self): + self.db.close_all() + shutil.rmtree(self.test_dir) + + def test_get_prior_defaults(self): + for key, default_val in _DEFAULT_PRIORS.items(): + self.assertEqual(self.recovery._get_prior(key), default_val) + + def test_get_prior_learned(self): + self.recovery._learned_priors = {"DB_CORRUPTION": 0.42} + self.assertEqual(self.recovery._get_prior("DB_CORRUPTION"), 0.42) + self.assertEqual(self.recovery._get_prior("OOM"), _DEFAULT_PRIORS["OOM"]) + + def test_update_learned_weights_no_data(self): + self.recovery._update_learned_weights() + self.assertIsNone(self.recovery._learned_priors) + + def test_update_learned_weights_with_history(self): + now = time.time() + desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION") + for i in range(10): + self.db.crash_history.insert_crash( + timestamp=now + i, + error_type="E", + error_message="m", + diagnosed_cause=desc, + symptoms={}, + probability=50, + entropy=0.0, + divergence=0.0, + ) + self.recovery._update_learned_weights() + self.assertIsNotNone(self.recovery._learned_priors) + self.assertGreater(self.recovery._learned_priors["DB_CORRUPTION"], 0.01) + self.assertLessEqual(self.recovery._learned_priors["DB_CORRUPTION"], 0.99) + + def test_weights_clamped(self): + now = time.time() + desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION") + for i in range(50): + self.db.crash_history.insert_crash( + timestamp=now + i, + error_type="E", + error_message="m", + diagnosed_cause=desc, + symptoms={}, + probability=99, + entropy=0.0, + divergence=0.0, + ) + self.recovery._update_learned_weights() + for v in self.recovery._learned_priors.values(): + self.assertGreaterEqual(v, 0.01) + self.assertLessEqual(v, 0.99) + + def test_load_learned_priors_from_config(self): + weights = {"DB_CORRUPTION": 0.55, "OOM": 0.12} + self.db.config.set("diagnostic_weights", json.dumps(weights)) + self.recovery._load_learned_priors() + self.assertEqual(self.recovery._learned_priors, weights) + + def test_load_learned_priors_no_config(self): + self.recovery._load_learned_priors() + self.assertIsNone(self.recovery._learned_priors) + + def test_set_database_loads_priors(self): + weights = {"DB_CORRUPTION": 0.33} + self.db.config.set("diagnostic_weights", json.dumps(weights)) + new_recovery = CrashRecovery() + new_recovery.set_database(self.db) + self.assertEqual(new_recovery._learned_priors, weights) + + def test_persist_crash(self): + causes = [ + {"description": "Test Cause", "probability": 75}, + ] + self.recovery._persist_crash("TypeError", "test msg", causes, {}, 1.5, 0.3) + rows = self.db.crash_history.get_recent_crashes() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["diagnosed_cause"], "Test Cause") + self.assertEqual(rows[0]["probability"], 75) + + def test_persist_crash_no_db(self): + recovery_no_db = CrashRecovery() + recovery_no_db._persist_crash("E", "m", [], {}, 0.0, 0.0) + + def test_cause_key_to_description_mapping(self): + for key in _DEFAULT_PRIORS: + desc = CrashRecovery._cause_key_to_description(key) + self.assertNotEqual(desc, key) + self.assertIsInstance(desc, str) + + def test_cause_key_to_description_unknown(self): + self.assertEqual( + CrashRecovery._cause_key_to_description("UNKNOWN_KEY"), + "UNKNOWN_KEY", + ) + + def test_update_weights_needs_min_3_crashes(self): + now = time.time() + desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION") + for i in range(2): + self.db.crash_history.insert_crash( + timestamp=now + i, + error_type="E", + error_message="m", + diagnosed_cause=desc, + symptoms={}, + probability=50, + entropy=0.0, + divergence=0.0, + ) + self.recovery._update_learned_weights() + self.assertIsNone(self.recovery._learned_priors) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/backend/test_health_monitor.py b/tests/backend/test_health_monitor.py new file mode 100644 index 0000000..79925a0 --- /dev/null +++ b/tests/backend/test_health_monitor.py @@ -0,0 +1,167 @@ +import math +import time +import unittest +from unittest.mock import MagicMock, patch + +from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler +from meshchatx.src.backend.recovery.health_monitor import HealthMonitor + + +class TestPersistentLogEntropy(unittest.TestCase): + """Tests for the log entropy and error rate properties.""" + + def setUp(self): + self.handler = PersistentLogHandler(capacity=5000, flush_interval=999) + + def test_entropy_zero_when_empty(self): + self.assertEqual(self.handler.current_log_entropy, 0.0) + + def test_error_rate_zero_when_empty(self): + self.assertEqual(self.handler.current_error_rate, 0.0) + + def test_entropy_zero_single_level(self): + now = time.monotonic() + with self.handler.lock: + for _ in range(100): + self.handler._level_events.append((now, "INFO")) + self.assertAlmostEqual(self.handler.current_log_entropy, 0.0, places=5) + + def test_entropy_max_uniform_distribution(self): + now = time.monotonic() + with self.handler.lock: + for level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): + for _ in range(20): + self.handler._level_events.append((now, level)) + expected = math.log2(5) + self.assertAlmostEqual(self.handler.current_log_entropy, expected, places=3) + + def test_entropy_ignores_old_events(self): + old = time.monotonic() - 120.0 + now = time.monotonic() + with self.handler.lock: + for level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): + for _ in range(20): + self.handler._level_events.append((old, level)) + for _ in range(50): + self.handler._level_events.append((now, "ERROR")) + self.assertAlmostEqual(self.handler.current_log_entropy, 0.0, places=5) + + def test_error_rate_all_errors(self): + now = time.monotonic() + with self.handler.lock: + for _ in range(50): + self.handler._level_events.append((now, "ERROR")) + self.handler._error_events.append(now) + self.assertAlmostEqual(self.handler.current_error_rate, 1.0) + + def test_error_rate_half(self): + now = time.monotonic() + with self.handler.lock: + for _ in range(50): + self.handler._level_events.append((now, "INFO")) + for _ in range(50): + self.handler._level_events.append((now, "ERROR")) + self.handler._error_events.append(now) + self.assertAlmostEqual(self.handler.current_error_rate, 0.5) + + +class TestHealthMonitorDetection(unittest.TestCase): + """Tests for HealthMonitor detection logic (no real threads).""" + + def setUp(self): + self.log_handler = MagicMock() + self.log_handler.current_log_entropy = 0.5 + self.log_handler.current_error_rate = 0.0 + self.monitor = HealthMonitor(log_handler=self.log_handler, app=None) + + def test_no_warnings_normal_state(self): + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + mock_bc.assert_not_called() + + def test_entropy_climb_detected(self): + self.monitor._entropy_history.extend([0.8, 1.2, 1.4]) + self.log_handler.current_log_entropy = 1.6 + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + mock_bc.assert_called_once() + args = mock_bc.call_args[0][0] + self.assertEqual(args["kind"], "entropy_climbing") + + def test_entropy_not_climbing_when_below_threshold(self): + self.monitor._entropy_history.extend([0.2, 0.4, 0.6]) + self.log_handler.current_log_entropy = 0.8 + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + mock_bc.assert_not_called() + + def test_error_rate_high_warning(self): + self.log_handler.current_error_rate = 0.5 + self.monitor._error_rate_history.append(0.4) + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + calls = [c[0][0] for c in mock_bc.call_args_list] + error_warnings = [c for c in calls if c["kind"] == "error_rate_high"] + self.assertEqual(len(error_warnings), 1) + + @patch("meshchatx.src.backend.recovery.health_monitor.psutil") + def test_memory_low_warning(self, mock_psutil): + mem_mock = MagicMock() + mem_mock.available = 50 * 1024 * 1024 # 50 MB + mock_psutil.virtual_memory.return_value = mem_mock + self.monitor._mem_available_history.append(80.0) + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + calls = [c[0][0] for c in mock_bc.call_args_list] + mem_warnings = [c for c in calls if c["kind"] == "memory_low"] + self.assertEqual(len(mem_warnings), 1) + + def test_latest_snapshot_structure(self): + self.monitor._check() + snap = self.monitor.latest_snapshot + self.assertIn("entropy", snap) + self.assertIn("error_rate", snap) + self.assertIn("memory_mb", snap) + self.assertEqual(len(snap["entropy"]), 1) + + def test_entropy_climb_needs_three_readings(self): + self.monitor._entropy_history.extend([1.2, 1.6]) + self.log_handler.current_log_entropy = 1.8 + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + calls = [c[0][0] for c in mock_bc.call_args_list] + entropy_warns = [c for c in calls if c["kind"] == "entropy_climbing"] + self.assertEqual(len(entropy_warns), 1) + + def test_entropy_no_climb_when_decreasing(self): + self.monitor._entropy_history.extend([2.0, 1.8, 1.6]) + self.log_handler.current_log_entropy = 1.4 + with patch.object(self.monitor, "_broadcast") as mock_bc: + self.monitor._check() + calls = [c[0][0] for c in mock_bc.call_args_list] + entropy_warns = [c for c in calls if c["kind"] == "entropy_climbing"] + self.assertEqual(len(entropy_warns), 0) + + def test_stop_prevents_further_checks(self): + self.monitor.stop() + self.assertFalse(self.monitor._running) + + +class TestHealthMonitorEdgeCases(unittest.TestCase): + def test_check_with_no_log_handler(self): + monitor = HealthMonitor(log_handler=None, app=None) + monitor._check() + snap = monitor.latest_snapshot + self.assertEqual(snap["entropy"], [0.0]) + + def test_deque_fixed_size(self): + monitor = HealthMonitor(log_handler=MagicMock(), app=None) + monitor.log_handler.current_log_entropy = 1.0 + monitor.log_handler.current_error_rate = 0.0 + for _ in range(20): + monitor._check() + self.assertEqual(len(monitor._entropy_history), monitor.ENTROPY_WINDOW) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py index 1f472c0..d51e6c7 100644 --- a/tests/backend/test_property_based.py +++ b/tests/backend/test_property_based.py @@ -1045,3 +1045,38 @@ class TestCrashRecoveryMathProperties: for c in causes: assert isinstance(c["probability"], int) assert 0 <= c["probability"] <= 100 + + @given( + prior=st.floats(min_value=0.01, max_value=0.99, allow_nan=False), + count=st.integers(min_value=0, max_value=100), + total=st.integers(min_value=3, max_value=200), + ) + @settings(derandomize=True, deadline=None, max_examples=100) + def test_bayesian_posterior_bounded(self, prior, count, total): + """Beta-Binomial posterior must stay in [0.01, 0.99].""" + import math as m + + count = min(count, total) + alpha = 1.0 + count + beta = 1.0 + (total - count) + posterior = alpha / (alpha + beta) + clamped = max(0.01, min(0.99, round(posterior, 4))) + assert m.isfinite(clamped) + assert 0.01 <= clamped <= 0.99 + + @given( + counts=st.lists( + st.integers(min_value=0, max_value=50), min_size=1, max_size=10 + ), + ) + @settings(derandomize=True, deadline=None, max_examples=50) + def test_bayesian_posteriors_sum_reasonable(self, counts): + """Posteriors from any crash distribution should be valid probabilities.""" + total = sum(counts) + 1 # ensure non-zero + posteriors = [] + for c in counts: + alpha = 1.0 + c + beta = 1.0 + max(0, total - c) + posteriors.append(max(0.01, min(0.99, alpha / (alpha + beta)))) + for p in posteriors: + assert 0.01 <= p <= 0.99