Improve crash recovery and monitoring capabilities.

2026-08-29 04:48:21 +00:00 · 2026-03-06 12:03:50 -06:00
parent 83716134d8
commit f10a29d21a
11 changed files with 896 additions and 19 deletions
@@ -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,
@@ -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()
@@ -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,),
+        )
@@ -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(
            """
@@ -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
@@ -1,3 +1,4 @@
 from .crash_recovery import CrashRecovery
+from .health_monitor import HealthMonitor

-__all__ = ["CrashRecovery"]
+__all__ = ["CrashRecovery", "HealthMonitor"]
@@ -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": [
@@ -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),
+        }
@@ -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()
@@ -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()
@@ -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