feat(phase_4): implement core engine orchestrator, prediction logic, stability analysis, and system state model

Author: AsieduDevelopmentHub

dashboard/lib/models/system_state.dart

@@ -3,13 +3,17 @@ class SystemState {
   final EngineInfo engine;
   final StressResult? stress;
   final NormalizedData? normalized;
+  final PredictionResult? prediction;
+  final StabilityIndex? stability;
   final AlertInfo alert;
   final BufferInfo buffers;
 
   SystemState({
     required this.engine,
     this.stress,
     this.normalized,
+    this.prediction,
+    this.stability,
     required this.alert,
     required this.buffers,
   });
@@ -23,6 +27,12 @@ class SystemState {
       normalized: json['normalized'] != null
           ? NormalizedData.fromJson(json['normalized'])
           : null,
+      prediction: json['prediction'] != null
+          ? PredictionResult.fromJson(json['prediction'])
+          : null,
+      stability: json['stability'] != null
+          ? StabilityIndex.fromJson(json['stability'])
+          : null,
       alert: AlertInfo.fromJson(json['alert'] ?? {}),
       buffers: BufferInfo.fromJson(json['buffers'] ?? {}),
     );
@@ -309,3 +319,43 @@ class SystemEvent {
     );
   }
 }
+
+class PredictionResult {
+  final double? memoryExhaustionEtaSec;
+  final double? cpuCriticalEtaSec;
+  final double riskScore;
+
+  PredictionResult({
+    this.memoryExhaustionEtaSec,
+    this.cpuCriticalEtaSec,
+    required this.riskScore,
+  });
+
+  factory PredictionResult.fromJson(Map<String, dynamic> json) {
+    return PredictionResult(
+      memoryExhaustionEtaSec: json['memory_exhaustion_eta_sec']?.toDouble(),
+      cpuCriticalEtaSec: json['cpu_critical_eta_sec']?.toDouble(),
+      riskScore: (json['risk_score'] ?? 0).toDouble(),
+    );
+  }
+}
+
+class StabilityIndex {
+  final double score;
+  final String state;
+  final Map<String, dynamic> components;
+
+  StabilityIndex({
+    required this.score,
+    required this.state,
+    required this.components,
+  });
+
+  factory StabilityIndex.fromJson(Map<String, dynamic> json) {
+    return StabilityIndex(
+      score: (json['score'] ?? 100).toDouble(),
+      state: json['state'] ?? 'unknown',
+      components: Map<String, dynamic>.from(json['components'] ?? {}),
+    );
+  }
+}

engine/intelligence/prediction_engine.py

@@ -0,0 +1,127 @@
+"""
+SentraCore — Prediction & Risk Engine.
+
+Forecasts resource exhaustion and calculates degradation probability based
+on smoothed historical trends.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from engine.intelligence.trend_analyzer import TrendResult
+    from engine.normalization.normalizer import NormalizedSnapshot
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class PredictionResult:
+    """Forecast of system degradation and resource exhaustion."""
+
+    memory_exhaustion_eta_sec: float | None
+    cpu_critical_eta_sec: float | None
+    risk_score: float  # 0 to 100%
+
+    def to_dict(self) -> dict:
+        return {
+            "memory_exhaustion_eta_sec": (
+                round(self.memory_exhaustion_eta_sec, 1) if self.memory_exhaustion_eta_sec else None
+            ),
+            "cpu_critical_eta_sec": (
+                round(self.cpu_critical_eta_sec, 1) if self.cpu_critical_eta_sec else None
+            ),
+            "risk_score": round(self.risk_score, 2),
+        }
+
+
+class PredictionEngine:
+    """
+    Computes Time-To-Exhaustion (TTE) for critical resources using
+    smoothed trend slopes, and produces an overarching risk score.
+    """
+
+    def __init__(self, smoothing_factor: float = 0.2) -> None:
+        """
+        Args:
+            smoothing_factor: Weight given to the new slope (EMA) to prevent chaotic ETAs.
+        """
+        self._smoothing_factor = smoothing_factor
+        self._smoothed_cpu_slope = 0.0
+        self._smoothed_memory_slope = 0.0
+
+    def predict(self, trend: 'TrendResult', snapshot: 'NormalizedSnapshot') -> PredictionResult:
+        """
+        Forecast resource exhaustion.
+
+        Args:
+            trend: The current trend analysis (containing slopes).
+            snapshot: Current normalized telemetry (for starting points).
+
+        Returns:
+            A PredictionResult indicating ETAs and overall risk.
+        """
+        # Apply Exponential Moving Average (EMA) to the slopes to smooth them
+        if self._smoothed_cpu_slope == 0.0:
+            self._smoothed_cpu_slope = trend.cpu_slope
+        else:
+            self._smoothed_cpu_slope = (
+                self._smoothing_factor * trend.cpu_slope
+                + (1 - self._smoothing_factor) * self._smoothed_cpu_slope
+            )
+
+        if self._smoothed_memory_slope == 0.0:
+            self._smoothed_memory_slope = trend.memory_slope
+        else:
+            self._smoothed_memory_slope = (
+                self._smoothing_factor * trend.memory_slope
+                + (1 - self._smoothing_factor) * self._smoothed_memory_slope
+            )
+
+        # Calculate Memory ETA to 98%
+        mem_eta = None
+        if self._smoothed_memory_slope > 0.05:  # Require at least 0.05% growth per second to forecast
+            remaining_mem = 98.0 - snapshot.memory_percent_smoothed
+            if remaining_mem > 0:
+                mem_eta = remaining_mem / self._smoothed_memory_slope
+
+        # Calculate CPU ETA to 95%
+        cpu_eta = None
+        if self._smoothed_cpu_slope > 0.1:  # Require at least 0.1% growth per second to forecast
+            remaining_cpu = 95.0 - snapshot.cpu_percent_smoothed
+            if remaining_cpu > 0:
+                cpu_eta = remaining_cpu / self._smoothed_cpu_slope
+
+        # Calculate Probabilistic Risk Score (0-100)
+        risk_score = 0.0
+        
+        # Risk from Memory exhaustion proximity
+        if mem_eta is not None:
+            if mem_eta < 60:
+                risk_score += 80.0  # Under 1 minute to OOM is critical
+            elif mem_eta < 300:
+                risk_score += 40.0  # Under 5 minutes is high risk
+            else:
+                risk_score += 10.0  # Leaking, but slow
+                
+        # Risk from CPU exhaustion proximity
+        if cpu_eta is not None:
+            if cpu_eta < 30:
+                risk_score += 60.0  # Fast spike to saturation
+            elif cpu_eta < 120:
+                risk_score += 30.0
+                
+        # Base risk based on sheer resource usage, regardless of slope
+        if snapshot.memory_percent_smoothed > 90.0:
+            risk_score += 40.0
+        if snapshot.cpu_percent_smoothed > 90.0:
+            risk_score += 30.0
+
+        return PredictionResult(
+            memory_exhaustion_eta_sec=mem_eta,
+            cpu_critical_eta_sec=cpu_eta,
+            risk_score=min(100.0, risk_score),
+        )

engine/intelligence/stability_index.py

@@ -0,0 +1,97 @@
+"""
+SentraCore — System Stability Index Calculator.
+
+Computes a global, unified metric of system health (1-100) by blending
+instantaneous stress, statistical anomalies, and predictive risk.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from engine.intelligence.anomaly_detector import AnomalyResult
+    from engine.intelligence.prediction_engine import PredictionResult
+    from engine.stress.stress_engine import StressResult
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class StabilityIndex:
+    """Unified system health score."""
+
+    score: float       # 1-100 (100 = perfect, 1 = critical)
+    state: str         # "stable", "degraded", "critical"
+    components: dict   # Breakdown of the penalty points
+
+    def to_dict(self) -> dict:
+        return {
+            "score": round(self.score, 1),
+            "state": self.state,
+            "components": self.components,
+        }
+
+
+class StabilityCalculator:
+    """
+    Computes System Stability Index.
+    
+    100 means the system is perfectly stable.
+    Lower scores indicate increasing levels of degradation.
+    """
+
+    def __init__(
+        self,
+        stress_weight: float = 0.5,
+        risk_weight: float = 0.3,
+        anomaly_weight: float = 0.2,
+    ) -> None:
+        self._w_stress = stress_weight
+        self._w_risk = risk_weight
+        self._w_anomaly = anomaly_weight
+
+        # Ensure weights sum to 1.0
+        total = self._w_stress + self._w_risk + self._w_anomaly
+        self._w_stress /= total
+        self._w_risk /= total
+        self._w_anomaly /= total
+
+    def calculate(
+        self,
+        stress: 'StressResult',
+        prediction: 'PredictionResult',
+        anomaly: 'AnomalyResult',
+    ) -> StabilityIndex:
+        """
+        Calculate global stability score.
+        """
+        # Calculate penalty points from each component (0-100)
+        stress_penalty = stress.score * self._w_stress
+        risk_penalty = prediction.risk_score * self._w_risk
+        anomaly_penalty = anomaly.score * self._w_anomaly
+
+        total_penalty = stress_penalty + risk_penalty + anomaly_penalty
+
+        # Stability is the inverse of the penalty
+        score = max(1.0, 100.0 - total_penalty)
+
+        # Determine state
+        if score >= 80.0:
+            state = "stable"
+        elif score >= 40.0:
+            state = "degraded"
+        else:
+            state = "critical"
+
+        return StabilityIndex(
+            score=score,
+            state=state,
+            components={
+                "stress_penalty": round(stress_penalty, 1),
+                "risk_penalty": round(risk_penalty, 1),
+                "anomaly_penalty": round(anomaly_penalty, 1),
+            },
+        )

engine/main.py

@@ -39,6 +39,8 @@
 from engine.stress.stress_engine import StressEngine, StressResult
 from engine.intelligence.trend_analyzer import TrendAnalyzer, TrendResult
 from engine.intelligence.anomaly_detector import AnomalyDetector, AnomalyResult
+from engine.intelligence.prediction_engine import PredictionEngine, PredictionResult
+from engine.intelligence.stability_index import StabilityCalculator, StabilityIndex
 
 logger = logging.getLogger(__name__)
 
@@ -60,14 +62,18 @@ def __init__(self) -> None:
         self.event_logger = EventLogger()
         self.trend_analyzer = TrendAnalyzer()
         self.anomaly_detector = AnomalyDetector()
+        self.prediction_engine = PredictionEngine()
         self.stress_engine = StressEngine()
+        self.stability_calculator = StabilityCalculator()
         self.alert_manager = AlertManager()
 
         # Current state (updated each cycle)
         self._current_stress: StressResult | None = None
         self._current_normalized: NormalizedSnapshot | None = None
         self._current_trend: TrendResult | None = None
         self._current_anomaly: AnomalyResult | None = None
+        self._current_prediction: PredictionResult | None = None
+        self._current_stability: StabilityIndex | None = None
         self._last_alert: Alert | None = None
 
         # Control
@@ -89,7 +95,9 @@ def get_current_state(self) -> dict:
             "normalized": self._current_normalized.to_dict() if self._current_normalized else None,
             "trend": self._current_trend.to_dict() if self._current_trend else None,
             "anomaly": self._current_anomaly.to_dict() if self._current_anomaly else None,
+            "prediction": self._current_prediction.to_dict() if self._current_prediction else None,
             "stress": self._current_stress.to_dict() if self._current_stress else None,
+            "stability": self._current_stability.to_dict() if self._current_stability else None,
             "alert": {
                 "total_fired": self.alert_manager.total_alerts,
                 "in_cooldown": self.alert_manager.is_in_cooldown,
@@ -201,7 +209,14 @@ async def run(self) -> None:
                 stress = self.stress_engine.compute(normalized, trend=trend, anomaly=anomaly)
                 self._current_stress = stress
 
-                # 9. Evaluate alerts
+                # 9. Phase 4: Prediction & Risk
+                prediction = self.prediction_engine.predict(trend, normalized)
+                self._current_prediction = prediction
+                
+                stability = self.stability_calculator.calculate(stress, prediction, anomaly)
+                self._current_stability = stability
+
+                # 10. Evaluate alerts
                 top_procs = self.process_tracker.get_top_consumers(5)
                 recent_events = self.event_logger.get_recent_events(20)
                 self.alert_manager.evaluate(stress, top_procs, recent_events)
@@ -212,17 +227,15 @@ async def run(self) -> None:
                 # Periodic log
                 if cycle % 15 == 0:  # Every ~30 seconds
                     logger.info(
-                        "Cycle %d | Stress: %.0f (%s) | CPU: %.1f%% | Mem: %.1f%% | "
-                        "Disk: %.0f ops/s | Baseline: %s (%d samples) | Events: %d",
+                        "Cycle %d | Stability: %.0f/100 (%s) | Risk: %.0f%% | CPU: %.1f%% | Mem: %.1f%% | "
+                        "Baseline: %s",
                         cycle,
-                        stress.score,
-                        stress.level,
+                        stability.score,
+                        stability.state,
+                        prediction.risk_score,
                         normalized.cpu_percent_smoothed,
                         normalized.memory_percent_smoothed,
-                        normalized.disk_total_ops_per_sec,
                         "ready" if self.baseline.is_ready else "learning",
-                        self.baseline.sample_count,
-                        self.event_logger.event_count,
                     )
 
             except Exception as exc: