Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions cortex/ask.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}"
Expand Down
23 changes: 23 additions & 0 deletions cortex/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2825,6 +2825,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) -> int:
"""Run interactive troubleshooter."""
from cortex.troubleshoot import Troubleshooter

troubleshooter = Troubleshooter()
return troubleshooter.start()

# --------------------------


Expand Down Expand Up @@ -3422,6 +3436,11 @@ def main():
)
# --------------------------

# Doctor command
subparsers.add_parser("doctor", help="System health check")

# Troubleshoot command
subparsers.add_parser("troubleshoot", help="Interactive system troubleshooter")
Comment on lines +3440 to +3443
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No arguments/options added to these subparsers. Consider adding --verbose or --no-execute flags for troubleshoot.

# License and upgrade commands
subparsers.add_parser("upgrade", help="Upgrade to Cortex Pro")
subparsers.add_parser("license", help="Show license status")
Expand Down Expand Up @@ -3623,6 +3642,10 @@ 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()
elif args.command == "upgrade":
from cortex.licensing import open_upgrade_page

Expand Down
87 changes: 87 additions & 0 deletions cortex/resolutions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""
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 json
import os
from pathlib import Path
from typing import TypedDict


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."""
import time

resolution: Resolution = {
"issue": issue,
"fix": fix,
"timestamp": time.time(),
}

try:
with open(self.storage_path) as f:
resolutions = json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
resolutions = []

resolutions.append(resolution)

# Keep only the last 50 resolutions to prevent unlimited growth
if len(resolutions) > 50:
resolutions = resolutions[-50:]

with open(self.storage_path, "w") as f:
json.dump(resolutions, f, indent=2)

def search(self, query: str, limit: int = 3) -> list[Resolution]:
Comment on lines +51 to +57
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Never use magic numbers, instead use constants.

"""
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
Comment on lines +76 to +78
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Silent skip of malformed entries. Should log a warning for debugging corrupted resolution files.

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]]
Loading
Loading