feat: add interactive troubleshooting assistant

cortex/ask.py

@@ -285,11 +285,12 @@ def _call_fake(self, question: str, system_prompt: str) -> str:
             return f"You have Python {platform.python_version()} installed."
         return "I cannot answer that question in test mode."
 
-    def ask(self, question: str) -> str:
+    def ask(self, question: str, system_prompt: str | None = None) -> str:
         """Ask a natural language question about the system.
 
         Args:
             question: Natural language question
+            system_prompt: Optional override for the system prompt
 
         Returns:
             Human-readable answer string
@@ -302,8 +303,11 @@ def ask(self, question: str) -> str:
             raise ValueError("Question cannot be empty")
 
         question = question.strip()
-        context = self.info_gatherer.gather_context()
-        system_prompt = self._get_system_prompt(context)
+
+        # Use provided system prompt or generate default
+        if system_prompt is None:
+            context = self.info_gatherer.gather_context()
+            system_prompt = self._get_system_prompt(context)
 
         # Cache lookup uses both question and system context (via system_prompt) for system-specific answers
         cache_key = f"ask:{question}"

cortex/cli.py

@@ -2978,6 +2978,20 @@ def progress_callback(current: int, total: int, step: InstallationStep) -> None:
                 console.print(f"Error: {result.error_message}", style="red")
             return 1
 
+    def doctor(self) -> int:
+        """Run system health checks."""
+        from cortex.doctor import SystemDoctor
+
+        doc = SystemDoctor()
+        return doc.run_checks()
+
+    def troubleshoot(self, no_execute: bool = False) -> int:
+        """Run interactive troubleshooter."""
+        from cortex.troubleshoot import Troubleshooter
+
+        troubleshooter = Troubleshooter(no_execute=no_execute)
+        return troubleshooter.start()
+
     # --------------------------
 
 
@@ -3162,6 +3176,8 @@ def show_rich_help():
     table.add_row("docker permissions", "Fix Docker bind-mount permissions")
     table.add_row("sandbox <cmd>", "Test packages in Docker sandbox")
     table.add_row("update", "Check for and install updates")
+    table.add_row("doctor", "System health check")
+    table.add_row("troubleshoot", "Interactive system troubleshooter")
 
     console.print(table)
     console.print()
@@ -3745,6 +3761,18 @@ def main():
     )
     # --------------------------
 
+    # Doctor command
+    doctor_parser = subparsers.add_parser("doctor", help="System health check")
+
+    # Troubleshoot command
+    troubleshoot_parser = subparsers.add_parser(
+        "troubleshoot", help="Interactive system troubleshooter"
+    )
+    troubleshoot_parser.add_argument(
+        "--no-execute",
+        action="store_true",
+        help="Disable automatic command execution (read-only mode)",
+    )
     # License and upgrade commands
     subparsers.add_parser("upgrade", help="Upgrade to Cortex Pro")
     subparsers.add_parser("license", help="Show license status")
@@ -3958,6 +3986,12 @@ def main():
             return 1
         elif args.command == "env":
             return cli.env(args)
+        elif args.command == "doctor":
+            return cli.doctor()
+        elif args.command == "troubleshoot":
+            return cli.troubleshoot(
+                no_execute=getattr(args, "no_execute", False),
+            )
         elif args.command == "config":
             return cli.config(args)
         elif args.command == "upgrade":

cortex/resolutions.py

@@ -0,0 +1,100 @@
+"""
+Resolution Manager for Cortex Troubleshooter.
+
+This module handles the storage and retrieval of successful troubleshooting resolutions.
+It uses a simple JSON file for storage and keyword matching for retrieval.
+"""
+
+import fcntl
+import json
+import os
+import time
+from pathlib import Path
+from typing import TypedDict
+
+MAX_RESOLUTIONS = 50
+DEFAULT_SEARCH_LIMIT = 3
+
+
+class Resolution(TypedDict):
+    issue: str
+    fix: str
+    timestamp: float
+
+
+class ResolutionManager:
+    def __init__(self, storage_path: str = "~/.cortex/resolutions.json"):
+        self.storage_path = Path(os.path.expanduser(storage_path))
+        self._ensure_storage()
+
+    def _ensure_storage(self) -> None:
+        """Ensure the storage file exists."""
+        if not self.storage_path.exists():
+            self.storage_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(self.storage_path, "w") as f:
+                json.dump([], f)
+
+    def save(self, issue: str, fix: str) -> None:
+        """Save a new resolution."""
+
+        resolution: Resolution = {
+            "issue": issue,
+            "fix": fix,
+            "timestamp": time.time(),
+        }
+
+        # Use r+ to allow reading and writing with a lock
+        with open(self.storage_path, "r+") as f:
+            # Acquire an exclusive lock to prevent race conditions
+            fcntl.flock(f, fcntl.LOCK_EX)
+            try:
+                try:
+                    resolutions = json.load(f)
+                except (json.JSONDecodeError, FileNotFoundError):
+                    resolutions = []
+
+                resolutions.append(resolution)
+
+                # Keep only the last N resolutions to prevent unlimited growth
+                if len(resolutions) > MAX_RESOLUTIONS:
+                    resolutions = resolutions[-MAX_RESOLUTIONS:]
+
+                # Rewind to the beginning of the file to overwrite
+                f.seek(0)
+                f.truncate()
+                json.dump(resolutions, f, indent=2)
+            finally:
+                # Always release the lock
+                fcntl.flock(f, fcntl.LOCK_UN)
+
+    def search(self, query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> list[Resolution]:
+        """
+        Search for resolutions relevant to the query.
+
+        Uses simple keyword matching: finds resolutions where the issue description
+        shares words with the query.
+        """
+        try:
+            with open(self.storage_path) as f:
+                resolutions: list[Resolution] = json.load(f)
+        except (json.JSONDecodeError, FileNotFoundError):
+            return []
+
+        if not resolutions:
+            return []
+
+        query_words = set(query.lower().split())
+        scored_resolutions = []
+
+        for res in resolutions:
+            if "issue" not in res or "fix" not in res:
+                continue
+            issue_words = set(res["issue"].lower().split())
+            # Calculate overlap score
+            score = len(query_words.intersection(issue_words))
+            if score > 0:
+                scored_resolutions.append((score, res))
+
+        # Sort by score (descending) and take top N
+        scored_resolutions.sort(key=lambda x: x[0], reverse=True)
+        return [res for _, res in scored_resolutions[:limit]]