mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-10-23 11:02:26 +00:00
Refactor crash recovery diagnostics and remove unused curvature calculations
- Simplified diagnostic output by renaming "Probabilistic Root Cause Analysis" to "Root Cause Analysis". - Removed unused methods and calculations related to manifold curvature. - Updated tests to reflect changes in diagnostic output and removed obsolete assertions.
This commit is contained in:
3 files changed
+8
-120
No files matched your search
@@ -1,11 +1,10 @@
|
|||||||
"""CRASH RECOVERY & ADAPTIVE DIAGNOSTIC ENGINE
|
"""CRASH RECOVERY & ADAPTIVE DIAGNOSTIC ENGINE
|
||||||
--------------------------------------------------
|
--------------------------------------------------
|
||||||
Mathematically grounded diagnostic system for MeshChatX.
|
Diagnostic system for MeshChatX.
|
||||||
|
|
||||||
Uses Shannon Entropy, KL-Divergence, and Bayesian weight learning
|
Uses Shannon Entropy, KL-Divergence, and Bayesian weight learning
|
||||||
to map application failures onto deterministic manifold constraints.
|
to diagnose application failures. Crash history is persisted and
|
||||||
Crash history is persisted and priors are refined over time using
|
priors are refined over time using a conjugate Beta-Binomial model.
|
||||||
a conjugate Beta-Binomial model.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import contextlib
|
import contextlib
|
||||||
@@ -213,18 +212,13 @@ class CrashRecovery:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
out.write(f" [ERROR] Failed to complete diagnosis: {e}\n")
|
out.write(f" [ERROR] Failed to complete diagnosis: {e}\n")
|
||||||
|
|
||||||
# Enhanced Explanation Engine (Analytic logic)
|
out.write("\nRoot Cause Analysis:\n")
|
||||||
out.write("\nProbabilistic Root Cause Analysis:\n")
|
|
||||||
causes = self._analyze_cause(exc_type, exc_value, diagnosis_results)
|
causes = self._analyze_cause(exc_type, exc_value, diagnosis_results)
|
||||||
|
|
||||||
# Calculate advanced system state metrics
|
|
||||||
entropy, divergence = self._calculate_system_entropy(diagnosis_results)
|
entropy, divergence = self._calculate_system_entropy(diagnosis_results)
|
||||||
curvature = self._calculate_manifold_curvature(causes)
|
|
||||||
|
|
||||||
out.write(f" [System Entropy: {entropy:.4f} bits]\n")
|
out.write(f" [System Entropy: {entropy:.4f} bits]\n")
|
||||||
out.write(f" [Systemic Divergence (KL): {divergence:.4f} bits]\n")
|
out.write(f" [KL-Divergence: {divergence:.4f} bits]\n")
|
||||||
out.write(f" [Manifold Curvature: {curvature:.2f}κ]\n")
|
|
||||||
out.write(" [Deterministic Manifold Constraints: V1,V4 Active]\n")
|
|
||||||
|
|
||||||
if self.log_handler:
|
if self.log_handler:
|
||||||
log_ent = self.log_handler.current_log_entropy
|
log_ent = self.log_handler.current_log_entropy
|
||||||
@@ -277,7 +271,7 @@ class CrashRecovery:
|
|||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
def _analyze_cause(self, exc_type, exc_value, diagnosis):
|
def _analyze_cause(self, exc_type, exc_value, diagnosis):
|
||||||
"""Uses probabilistic active inference and heuristic pattern matching
|
"""Uses heuristic pattern matching and Bayesian priors
|
||||||
to determine the likely root cause of the application crash.
|
to determine the likely root cause of the application crash.
|
||||||
"""
|
"""
|
||||||
causes = []
|
causes = []
|
||||||
@@ -470,21 +464,6 @@ class CrashRecovery:
|
|||||||
|
|
||||||
causes.sort(key=lambda x: x["probability"], reverse=True)
|
causes.sort(key=lambda x: x["probability"], reverse=True)
|
||||||
|
|
||||||
# Apply Mathematical Grounding via Active Inference Directives if possible
|
|
||||||
if causes:
|
|
||||||
# We "ground" the top cause
|
|
||||||
top_cause = causes[0]
|
|
||||||
if top_cause["probability"] > 90:
|
|
||||||
top_cause["reasoning"] += (
|
|
||||||
" This diagnosis has reached a high-confidence threshold grounded in "
|
|
||||||
"deterministic manifold constraints (V1,V4) and active inference."
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
top_cause["reasoning"] += (
|
|
||||||
" This diagnosis is based on probabilistic heuristic matching of "
|
|
||||||
"current system entropy against known failure manifolds."
|
|
||||||
)
|
|
||||||
|
|
||||||
return causes
|
return causes
|
||||||
|
|
||||||
def _calculate_system_entropy(self, diagnosis):
|
def _calculate_system_entropy(self, diagnosis):
|
||||||
@@ -547,23 +526,6 @@ class CrashRecovery:
|
|||||||
|
|
||||||
return entropy, divergence
|
return entropy, divergence
|
||||||
|
|
||||||
def _calculate_manifold_curvature(self, causes):
|
|
||||||
"""Calculates 'Manifold Curvature' (κ) based on the gradient of probabilities.
|
|
||||||
High curvature indicates a 'sharp' failure where one cause is dominant.
|
|
||||||
Low curvature indicates an 'ambiguous' failure landscape.
|
|
||||||
"""
|
|
||||||
if not causes:
|
|
||||||
return 0.0
|
|
||||||
|
|
||||||
probs = [c["probability"] / 100.0 for c in causes]
|
|
||||||
if len(probs) < 2:
|
|
||||||
# If there's only one cause and it's high probability, curvature is high
|
|
||||||
return probs[0] * 10.0
|
|
||||||
|
|
||||||
# Curvature is the 'steepness' between the top two causes
|
|
||||||
gradient = probs[0] - probs[1]
|
|
||||||
return gradient * 10.0
|
|
||||||
|
|
||||||
def run_diagnosis(self, file=sys.stderr):
|
def run_diagnosis(self, file=sys.stderr):
|
||||||
"""Performs a series of OS-agnostic checks on the application's environment."""
|
"""Performs a series of OS-agnostic checks on the application's environment."""
|
||||||
results = {
|
results = {
|
||||||
|
|||||||
@@ -120,7 +120,7 @@ class TestCrashRecovery(unittest.TestCase):
|
|||||||
self.assertIn("!!! APPLICATION CRASH DETECTED !!!", report)
|
self.assertIn("!!! APPLICATION CRASH DETECTED !!!", report)
|
||||||
self.assertIn("Type: ValueError", report)
|
self.assertIn("Type: ValueError", report)
|
||||||
self.assertIn("Message: Simulated error for testing", report)
|
self.assertIn("Message: Simulated error for testing", report)
|
||||||
self.assertIn("Probabilistic Root Cause Analysis:", report)
|
self.assertIn("Root Cause Analysis:", report)
|
||||||
self.assertIn("Recovery Suggestions:", report)
|
self.assertIn("Recovery Suggestions:", report)
|
||||||
|
|
||||||
def test_heuristic_analysis_sqlite(self):
|
def test_heuristic_analysis_sqlite(self):
|
||||||
@@ -164,10 +164,6 @@ class TestCrashRecovery(unittest.TestCase):
|
|||||||
causes = self.recovery._analyze_cause(exc_type, exc_value, diagnosis)
|
causes = self.recovery._analyze_cause(exc_type, exc_value, diagnosis)
|
||||||
self.assertEqual(causes[0]["description"], "Missing Reticulum Configuration")
|
self.assertEqual(causes[0]["description"], "Missing Reticulum Configuration")
|
||||||
self.assertEqual(causes[0]["probability"], 99)
|
self.assertEqual(causes[0]["probability"], 99)
|
||||||
self.assertIn(
|
|
||||||
"deterministic manifold constraints",
|
|
||||||
causes[0]["reasoning"].lower(),
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_entropy_calculation_levels(self):
|
def test_entropy_calculation_levels(self):
|
||||||
"""Test how entropy reflects system disorder."""
|
"""Test how entropy reflects system disorder."""
|
||||||
@@ -201,27 +197,6 @@ class TestCrashRecovery(unittest.TestCase):
|
|||||||
# Actually for a "disorder" metric, we might want it to peak when things are most uncertain.
|
# Actually for a "disorder" metric, we might want it to peak when things are most uncertain.
|
||||||
# But in our context, we are showing entropy of the "State Predictability".
|
# But in our context, we are showing entropy of the "State Predictability".
|
||||||
|
|
||||||
def test_confidence_grounding_text(self):
|
|
||||||
"""Verify that reasoning text reflects grounding logic."""
|
|
||||||
# High confidence scenario
|
|
||||||
exc_type = RuntimeError
|
|
||||||
exc_value = RuntimeError("no current event loop")
|
|
||||||
diagnosis = {} # probability 88% -> heuristic matching
|
|
||||||
causes_low = self.recovery._analyze_cause(exc_type, exc_value, diagnosis)
|
|
||||||
self.assertIn(
|
|
||||||
"probabilistic heuristic matching",
|
|
||||||
causes_low[0]["reasoning"].lower(),
|
|
||||||
)
|
|
||||||
|
|
||||||
# Near-certainty scenario
|
|
||||||
diagnosis_certain = {"config_missing": True}
|
|
||||||
causes_high = self.recovery._analyze_cause(
|
|
||||||
exc_type,
|
|
||||||
exc_value,
|
|
||||||
diagnosis_certain,
|
|
||||||
)
|
|
||||||
self.assertIn("high-confidence threshold", causes_high[0]["reasoning"].lower())
|
|
||||||
|
|
||||||
def test_heuristic_analysis_lxmf_storage(self):
|
def test_heuristic_analysis_lxmf_storage(self):
|
||||||
"""Test LXMF storage failure detection."""
|
"""Test LXMF storage failure detection."""
|
||||||
exc_type = RuntimeError
|
exc_type = RuntimeError
|
||||||
@@ -280,8 +255,7 @@ class TestCrashRecovery(unittest.TestCase):
|
|||||||
|
|
||||||
report = output.getvalue()
|
report = output.getvalue()
|
||||||
self.assertIn("[System Entropy:", report)
|
self.assertIn("[System Entropy:", report)
|
||||||
self.assertIn("[Deterministic Manifold Constraints:", report)
|
self.assertIn("[KL-Divergence:", report)
|
||||||
self.assertIn("deterministic manifold constraints", report.lower())
|
|
||||||
|
|
||||||
def test_heuristic_analysis_unsupported_python(self):
|
def test_heuristic_analysis_unsupported_python(self):
|
||||||
"""Test detection of unsupported Python versions."""
|
"""Test detection of unsupported Python versions."""
|
||||||
@@ -344,38 +318,6 @@ class TestCrashRecovery(unittest.TestCase):
|
|||||||
finally:
|
finally:
|
||||||
sys.excepthook = original
|
sys.excepthook = original
|
||||||
|
|
||||||
# ==================================================================
|
|
||||||
# _calculate_manifold_curvature
|
|
||||||
# ==================================================================
|
|
||||||
|
|
||||||
def test_curvature_empty_causes(self):
|
|
||||||
"""Empty causes list should return 0.0 curvature."""
|
|
||||||
self.assertEqual(self.recovery._calculate_manifold_curvature([]), 0.0)
|
|
||||||
|
|
||||||
def test_curvature_single_cause(self):
|
|
||||||
"""Single cause should return probability * 10."""
|
|
||||||
causes = [{"probability": 95}]
|
|
||||||
self.assertAlmostEqual(self.recovery._calculate_manifold_curvature(causes), 9.5)
|
|
||||||
|
|
||||||
def test_curvature_two_causes_gradient(self):
|
|
||||||
"""Curvature should be 10 * (top - second)."""
|
|
||||||
causes = [{"probability": 90}, {"probability": 40}]
|
|
||||||
self.assertAlmostEqual(self.recovery._calculate_manifold_curvature(causes), 5.0)
|
|
||||||
|
|
||||||
def test_curvature_equal_causes(self):
|
|
||||||
"""Equal probabilities should give curvature = 0."""
|
|
||||||
causes = [{"probability": 50}, {"probability": 50}]
|
|
||||||
self.assertAlmostEqual(self.recovery._calculate_manifold_curvature(causes), 0.0)
|
|
||||||
|
|
||||||
def test_curvature_many_causes(self):
|
|
||||||
"""Only top 2 probabilities matter for curvature."""
|
|
||||||
causes = [
|
|
||||||
{"probability": 80},
|
|
||||||
{"probability": 30},
|
|
||||||
{"probability": 10},
|
|
||||||
]
|
|
||||||
self.assertAlmostEqual(self.recovery._calculate_manifold_curvature(causes), 5.0)
|
|
||||||
|
|
||||||
# ==================================================================
|
# ==================================================================
|
||||||
# _calculate_system_entropy edge cases
|
# _calculate_system_entropy edge cases
|
||||||
# ==================================================================
|
# ==================================================================
|
||||||
|
|||||||
@@ -998,22 +998,6 @@ class TestCrashRecoveryMathProperties:
|
|||||||
assert entropy >= 0, f"Negative entropy: {entropy}"
|
assert entropy >= 0, f"Negative entropy: {entropy}"
|
||||||
assert divergence >= 0, f"Negative divergence: {divergence}"
|
assert divergence >= 0, f"Negative divergence: {divergence}"
|
||||||
|
|
||||||
@given(
|
|
||||||
probs=st.lists(
|
|
||||||
st.integers(min_value=0, max_value=100), min_size=0, max_size=10
|
|
||||||
),
|
|
||||||
)
|
|
||||||
@settings(derandomize=True, deadline=None)
|
|
||||||
def test_manifold_curvature_bounded(self, probs):
|
|
||||||
"""Manifold curvature must always be a non-negative finite float."""
|
|
||||||
import math as m
|
|
||||||
|
|
||||||
recovery = self._make_recovery()
|
|
||||||
causes = [{"probability": p} for p in probs]
|
|
||||||
curvature = recovery._calculate_manifold_curvature(causes)
|
|
||||||
assert m.isfinite(curvature), f"Non-finite curvature: {curvature}"
|
|
||||||
assert curvature >= 0 or len(probs) >= 2, f"Unexpected negative: {curvature}"
|
|
||||||
|
|
||||||
@given(
|
@given(
|
||||||
exc_msg=st.text(min_size=0, max_size=200),
|
exc_msg=st.text(min_size=0, max_size=200),
|
||||||
exc_type_name=st.sampled_from(
|
exc_type_name=st.sampled_from(
|
||||||
|
|||||||
Reference in new issue
Block a user