feat: Add system health monitoring module (#128)

cortex/cli.py

@@ -38,7 +38,7 @@
     validate_installation_id,
     ValidationError
 )
-# Import the new Notification Manager
+# Import Notification Manager
 from cortex.notification_manager import NotificationManager
 
 
@@ -112,10 +112,9 @@
         sys.stdout.write('\r\033[K')
         sys.stdout.flush()
 
-    # --- New Notification Method ---
+    # --- Notification Method ---
     def notify(self, args):
         """Handle notification commands"""
-        # Addressing CodeRabbit feedback: Handle missing subcommand gracefully
         if not args.notify_action:
             self._print_error("Please specify a subcommand (config/enable/disable/dnd/send)")
             return 1
@@ -132,24 +131,21 @@
 
         elif args.notify_action == 'enable':
             mgr.config["enabled"] = True
-            # Addressing CodeRabbit feedback: Ideally should use a public method instead of private _save_config,
-            # but keeping as is for a simple fix (or adding a save method to NotificationManager would be best).
-            mgr._save_config() 
+            mgr.save_config() 
             self._print_success("Notifications enabled")
             return 0
 
         elif args.notify_action == 'disable':
             mgr.config["enabled"] = False
-            mgr._save_config()
-            cx_print("Notifications disabled (Critical alerts will still show)", "warning")
+            mgr.save_config()
+            self._print_success("Notifications disabled (Critical alerts will still show)")
             return 0
 
         elif args.notify_action == 'dnd':
             if not args.start or not args.end:
                 self._print_error("Please provide start and end times (HH:MM)")
                 return 1
 
-            # Addressing CodeRabbit feedback: Add time format validation
             try:
                 datetime.strptime(args.start, "%H:%M")
                 datetime.strptime(args.end, "%H:%M")
@@ -159,7 +155,7 @@
 
             mgr.config["dnd_start"] = args.start
             mgr.config["dnd_end"] = args.end
-            mgr._save_config()
+            mgr.save_config()
             self._print_success(f"DND Window updated: {args.start} - {args.end}")
             return 0
 
@@ -174,7 +170,56 @@
         else:
             self._print_error("Unknown notify command")
             return 1
-    # -------------------------------
+
+    # --- New Health Command ---
+    def health(self, args):
+        """Run system health checks and show recommendations"""
+        from cortex.health.monitor import HealthMonitor
+
+        self._print_status("🔍", "Running system health checks...")
+        monitor = HealthMonitor()
+        report = monitor.run_all()
+
+        # --- Display Results ---
+        score = report['total_score']
+
+        # Color code the score
+        score_color = "green"
+        if score < 60: score_color = "red"
+        elif score < 80: score_color = "yellow"
+
+        console.print()
+        console.print(f"📊 [bold]System Health Score:[/bold] [{score_color}]{score}/100[/{score_color}]")
+        console.print()
+
+        console.print("[bold]Factors:[/bold]")
+        recommendations = []
+
+        for res in report['results']:
+            status_icon = "✅"
+            if res['status'] == 'WARNING': status_icon = "⚠️ "
+            elif res['status'] == 'CRITICAL': status_icon = "❌"
+
+            console.print(f"   {status_icon}  {res['name']:<15}: {res['score']}/100 ({res['details']})")
+
+            if res['recommendation']:
+                recommendations.append(res['recommendation'])
+
+        console.print()
+
+        if recommendations:
+            console.print("[bold]Recommendations:[/bold]")
+            for i, rec in enumerate(recommendations, 1):
+                console.print(f"   {i}. {rec}")
+
+            console.print()
+            # Note: Auto-fix logic would go here, prompting user to apply specific commands.
+            # For this iteration, we display actionable advice.
+            console.print("[dim]Run suggested commands manually to improve your score.[/dim]")
+        else:
+            self._print_success("System is in excellent health! No actions needed.")
+
+        return 0
 
     def install(self, software: str, execute: bool = False, dry_run: bool = False):
         # Validate input first
@@ -543,7 +588,8 @@
     table.add_row("install <pkg>", "Install software")
     table.add_row("history", "View history")
     table.add_row("rollback <id>", "Undo installation")
-    table.add_row("notify", "Manage desktop notifications")  # Added this line
+    table.add_row("notify", "Manage desktop notifications")
+    table.add_row("health", "Check system health score") # Added this line
 
     console.print(table)
     console.print()
@@ -598,7 +644,7 @@
     edit_pref_parser.add_argument('key', nargs='?')
     edit_pref_parser.add_argument('value', nargs='?')
 
-    # --- New Notify Command ---
+    # --- Notify Command ---
     notify_parser = subparsers.add_parser('notify', help='Manage desktop notifications')
     notify_subs = notify_parser.add_subparsers(dest='notify_action', help='Notify actions')
 
@@ -615,6 +661,9 @@
     send_parser.add_argument('--title', default='Cortex Notification')
     send_parser.add_argument('--level', choices=['low', 'normal', 'critical'], default='normal')
     send_parser.add_argument('--actions', nargs='*', help='Action buttons')
+
+    # --- New Health Command ---
+    health_parser = subparsers.add_parser('health', help='Check system health score')
     # --------------------------
 
     args = parser.parse_args()
@@ -642,9 +691,11 @@
             return cli.check_pref(key=args.key)
         elif args.command == 'edit-pref':
             return cli.edit_pref(action=args.action, key=args.key, value=args.value)
-        # Handle the new notify command
         elif args.command == 'notify':
             return cli.notify(args)
+        # Handle new command
+        elif args.command == 'health':
+            return cli.health(args)
         else:
             parser.print_help()
             return 1

cortex/health/checks/disk.py

@@ -0,0 +1,33 @@
+import shutil
+from ..monitor import HealthCheck, CheckResult
+
+class DiskCheck(HealthCheck):
+    def run(self) -> CheckResult:
+        total, used, free = shutil.disk_usage("/")
+        # Calculate usage percentage
+        usage_percent = (used / total) * 100
+
+        score = 100
+        status = "OK"
+        details = f"{usage_percent:.1f}% Used"
+        rec = None
+
+        # Scoring logic (Spec compliant)
+        if usage_percent > 90:
+            score = 0
+            status = "CRITICAL"
+            rec = "Clean package cache (+50 pts)"
+        elif usage_percent > 80:
+            score = 50
+            status = "WARNING"
+            rec = "Clean package cache (+10 pts)"
+
+        return CheckResult(
+            name="Disk Space",
+            category="disk",
+            score=score,
+            status=status,
+            details=details,
+            recommendation=rec,
+            weight=0.15  # 15%
+        )

cortex/health/checks/performance.py

@@ -0,0 +1,63 @@
+import os
+import multiprocessing
+from ..monitor import HealthCheck, CheckResult
+
+class PerformanceCheck(HealthCheck):
+    def run(self) -> CheckResult:
+        score = 100
+        issues = []
+        rec = None
+
+        # 1. Load Average (1min)
+        try:
+            load1, _, _ = os.getloadavg()
+            cores = multiprocessing.cpu_count()
+            # Load ratio against core count
+            load_ratio = load1 / cores
+
+            if load_ratio > 1.0:
+                score -= 50
+                issues.append(f"High Load ({load1:.2f})")
+                rec = "Check top processes"
+        except Exception:
+            pass # Skip on Windows etc.
+
+        # 2. Memory Usage (Linux /proc/meminfo)
+        try:
+            with open('/proc/meminfo', 'r') as f:
+                meminfo = {}
+                for line in f:
+                    parts = line.split(':')
+                    if len(parts) == 2:
+                        meminfo[parts[0].strip()] = int(parts[1].strip().split()[0])
+
+                if 'MemTotal' in meminfo and 'MemAvailable' in meminfo:
+                    total = meminfo['MemTotal']
+                    avail = meminfo['MemAvailable']
+                    used_percent = ((total - avail) / total) * 100
+
+                    if used_percent > 80:
+                        penalty = int(used_percent - 80)
+                        score -= penalty
+                        issues.append(f"High Memory ({used_percent:.0f}%)")
+        except FileNotFoundError:
+            pass # Non-Linux systems
+
+        # Summary of results
+        status = "OK"
+        if score < 50:
+            status = "CRITICAL"
+        elif score < 90:
+            status = "WARNING"
+
+        details = ", ".join(issues) if issues else "Optimal"
+
+        return CheckResult(
+            name="System Load",
+            category="performance",
+            score=max(0, score),
+            status=status,
+            details=details,
+            recommendation=rec,
+            weight=0.20 # 20%
+        )

cortex/health/checks/security.py

@@ -0,0 +1,57 @@
+import subprocess
+import os
+from ..monitor import HealthCheck, CheckResult
+
+class SecurityCheck(HealthCheck):
+    def run(self) -> CheckResult:
+        score = 100
+        issues = []
+        recommendations = []
+
+        # 1. Firewall (UFW) Check
+        ufw_active = False
+        try:
+            res = subprocess.run(
+                ["systemctl", "is-active", "ufw"], 
+                capture_output=True, text=True
+            )
+            # Fix: Use exact match to avoid matching "inactive" which contains "active"
+            if res.returncode == 0 and res.stdout.strip() == "active":
+                ufw_active = True
+        except FileNotFoundError:
+            pass # Environment without systemctl (e.g., Docker or non-systemd)
+
+        if not ufw_active:
+            score = 0 # Spec: 0 points if Firewall is inactive
+            issues.append("Firewall Inactive")
+            recommendations.append("Enable UFW Firewall")
+
+        # 2. SSH Root Login Check
+        try:
+            ssh_config = "/etc/ssh/sshd_config"
+            if os.path.exists(ssh_config):
+                with open(ssh_config, 'r') as f:
+                    for line in f:
+                        line = line.strip()
+                        # Check for uncommented PermitRootLogin yes
+                        if line.startswith("PermitRootLogin") and "yes" in line.split():
+                            score -= 50
+                            issues.append("Root SSH Allowed")
+                            recommendations.append("Disable SSH Root Login in sshd_config")
+                            break
+        except PermissionError:
+            pass # Cannot read config, skip check
+
+        status = "OK"
+        if score < 50: status = "CRITICAL"
+        elif score < 100: status = "WARNING"
+
+        return CheckResult(
+            name="Security Posture",
+            category="security",
+            score=max(0, score),
+            status=status,
+            details=", ".join(issues) if issues else "Secure",
+            recommendation=", ".join(recommendations) if recommendations else None,
+            weight=0.35
+        )

cortex/health/checks/updates.py

@@ -0,0 +1,56 @@
+import subprocess
+from ..monitor import HealthCheck, CheckResult
+
+class UpdateCheck(HealthCheck):
+    def run(self) -> CheckResult:
+        score = 100
+        pkg_count = 0
+        sec_count = 0
+        rec = None
+
+        # Parse apt list --upgradable
+        try:
+            # Execute safely without pipeline
+            res = subprocess.run(
+                ["apt", "list", "--upgradable"], 
+                capture_output=True, text=True
+            )
+
+            lines = res.stdout.splitlines()
+            # Skip first line "Listing..."
+            for line in lines[1:]:
+                if line.strip():
+                    pkg_count += 1
+                    if "security" in line.lower():
+                        sec_count += 1
+
+            # Scoring
+            score -= (pkg_count * 2) # -2 pts per normal package
+            score -= (sec_count * 10) # -10 pts per security package
+
+            if pkg_count > 0:
+                rec = f"Install {pkg_count} updates (+{100-score} pts)"
+
+        except FileNotFoundError:
+            # Skip on non-apt environments (100 pts)
+            return CheckResult("Updates", "updates", 100, "SKIP", "apt not found", weight=0.30)
+        except Exception:
+            pass # Ignore errors
+
+        status = "OK"
+        if score < 60: status = "CRITICAL"
+        elif score < 100: status = "WARNING"
+
+        details = f"{pkg_count} pending"
+        if sec_count > 0:
+            details += f" ({sec_count} security)"
+
+        return CheckResult(
+            name="System Updates",
+            category="updates",
+            score=max(0, score),
+            status=status,
+            details=details,
+            recommendation=rec,
+            weight=0.30 # 30%
+        )