From 9daf16e81e01c39e0753643a661ce5ad463df6cc Mon Sep 17 00:00:00 2001 From: Shree Jejurikar Date: Mon, 29 Dec 2025 15:26:16 +0530 Subject: [PATCH 1/7] feat: Enhance cortex ask with interactive tutor capabilities (Issue #393) - Enhanced system prompt to detect educational vs diagnostic queries - Added LearningTracker class for tracking educational topics - Learning history stored in ~/.cortex/learning_history.json - Increased max_tokens from 500 to 2000 for longer responses - Added terminal-friendly formatting rules - Rich Markdown rendering for proper terminal display - Added 25 new unit tests (50 total) for ask module - Updated COMMANDS.md with cortex ask documentation --- cortex/ask.py | 218 ++++++++++++++++++++++++++++++++++++++++++-- cortex/cli.py | 5 +- docs/COMMANDS.md | 78 ++++++++++++++++ tests/test_ask.py | 228 +++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 518 insertions(+), 11 deletions(-) diff --git a/cortex/ask.py b/cortex/ask.py index 2aa0b932..d1078065 100644 --- a/cortex/ask.py +++ b/cortex/ask.py @@ -1,15 +1,19 @@ """Natural language query interface for Cortex. Handles user questions about installed packages, configurations, -and system state using LLM with semantic caching. +and system state using LLM with semantic caching. Also provides +educational content and tracks learning progress. """ import json import os import platform +import re import shutil import sqlite3 import subprocess +from datetime import datetime +from pathlib import Path from typing import Any @@ -132,6 +136,134 @@ def gather_context(self) -> dict[str, Any]: } +class LearningTracker: + """Tracks educational topics the user has explored.""" + + PROGRESS_FILE = Path.home() / ".cortex" / "learning_history.json" + + # Patterns that indicate educational questions + EDUCATIONAL_PATTERNS = [ + r"^explain\b", + r"^teach\s+me\b", + r"^what\s+is\b", + r"^what\s+are\b", + r"^how\s+does\b", + r"^how\s+do\b", + r"^how\s+to\b", + r"\bbest\s+practices?\b", + r"^tutorial\b", + r"^guide\s+to\b", + r"^learn\s+about\b", + r"^introduction\s+to\b", + r"^basics\s+of\b", + ] + + def __init__(self): + """Initialize the learning tracker.""" + self._compiled_patterns = [ + re.compile(p, re.IGNORECASE) for p in self.EDUCATIONAL_PATTERNS + ] + + def is_educational_query(self, question: str) -> bool: + """Determine if a question is educational in nature.""" + for pattern in self._compiled_patterns: + if pattern.search(question): + return True + return False + + def extract_topic(self, question: str) -> str: + """Extract the main topic from an educational question.""" + # Remove common prefixes + topic = question.lower() + prefixes_to_remove = [ + r"^explain\s+", + r"^teach\s+me\s+about\s+", + r"^teach\s+me\s+", + r"^what\s+is\s+", + r"^what\s+are\s+", + r"^how\s+does\s+", + r"^how\s+do\s+", + r"^how\s+to\s+", + r"^tutorial\s+on\s+", + r"^guide\s+to\s+", + r"^learn\s+about\s+", + r"^introduction\s+to\s+", + r"^basics\s+of\s+", + r"^best\s+practices\s+for\s+", + ] + for prefix in prefixes_to_remove: + topic = re.sub(prefix, "", topic, flags=re.IGNORECASE) + + # Clean up and truncate + topic = topic.strip("? ").strip() + # Take first 50 chars as topic identifier + if len(topic) > 50: + topic = topic[:50].rsplit(" ", 1)[0] + return topic + + def record_topic(self, question: str) -> None: + """Record that the user explored an educational topic.""" + if not self.is_educational_query(question): + return + + topic = self.extract_topic(question) + if not topic: + return + + history = self._load_history() + + # Update or add topic + if topic in history["topics"]: + history["topics"][topic]["count"] += 1 + history["topics"][topic]["last_accessed"] = datetime.now().isoformat() + else: + history["topics"][topic] = { + "count": 1, + "first_accessed": datetime.now().isoformat(), + "last_accessed": datetime.now().isoformat(), + } + + history["total_queries"] = history.get("total_queries", 0) + 1 + self._save_history(history) + + def get_history(self) -> dict[str, Any]: + """Get the learning history.""" + return self._load_history() + + def get_recent_topics(self, limit: int = 5) -> list[str]: + """Get recently explored topics.""" + history = self._load_history() + topics = history.get("topics", {}) + + # Sort by last_accessed + sorted_topics = sorted( + topics.items(), + key=lambda x: x[1].get("last_accessed", ""), + reverse=True, + ) + return [t[0] for t in sorted_topics[:limit]] + + def _load_history(self) -> dict[str, Any]: + """Load learning history from file.""" + if not self.PROGRESS_FILE.exists(): + return {"topics": {}, "total_queries": 0} + + try: + with open(self.PROGRESS_FILE, "r") as f: + return json.load(f) + except (json.JSONDecodeError, OSError): + return {"topics": {}, "total_queries": 0} + + def _save_history(self, history: dict[str, Any]) -> None: + """Save learning history to file.""" + try: + self.PROGRESS_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(self.PROGRESS_FILE, "w") as f: + json.dump(history, f, indent=2) + except OSError: + pass # Silently fail if we can't write + + class AskHandler: """Handles natural language questions about the system.""" @@ -155,6 +287,7 @@ def __init__( self.offline = offline self.model = model or self._default_model() self.info_gatherer = SystemInfoGatherer() + self.learning_tracker = LearningTracker() # Initialize cache try: @@ -201,18 +334,63 @@ def _initialize_client(self): raise ValueError(f"Unsupported provider: {self.provider}") def _get_system_prompt(self, context: dict[str, Any]) -> str: - return f"""You are a helpful Linux system assistant. Answer questions about the user's system clearly and concisely. + return f"""You are a helpful Linux system assistant and tutor. You help users with both system-specific questions AND educational queries about Linux, packages, and best practices. System Context: {json.dumps(context, indent=2)} -Rules: -1. Provide direct, human-readable answers -2. Use the system context to give accurate information +## Query Type Detection + +Automatically detect the type of question and respond appropriately: + +### Educational Questions (tutorials, explanations, learning) +Triggered by questions like: "explain...", "teach me...", "how does X work", "what is...", "best practices for...", "tutorial on...", "learn about...", "guide to..." + +For educational questions: +1. Provide structured, tutorial-style explanations +2. Include practical code examples with proper formatting +3. Highlight best practices and common pitfalls to avoid +4. Break complex topics into digestible sections +5. Use clear section labels and bullet points for readability +6. Mention related topics the user might want to explore next +7. Tailor examples to the user's system when relevant (e.g., use apt for Debian-based systems) + +### Diagnostic Questions (system-specific, troubleshooting) +Triggered by questions about: current system state, "why is my...", "what packages...", "check my...", specific errors, system status + +For diagnostic questions: +1. Analyze the provided system context +2. Give specific, actionable answers 3. Be concise but informative 4. If you don't have enough information, say so clearly -5. For package compatibility questions, consider the system's Python version and OS -6. Return ONLY the answer text, no JSON or markdown formatting""" + +## Output Formatting Rules (CRITICAL - Follow exactly) +1. NEVER use markdown headings (# or ##) - they render poorly in terminals +2. For section titles, use **Bold Text** on its own line instead +3. Use bullet points (-) for lists +4. Use numbered lists (1. 2. 3.) for sequential steps +5. Use triple backticks with language name for code blocks (```bash) +6. Use *italic* sparingly for emphasis +7. Keep lines under 100 characters when possible +8. Add blank lines between sections for readability +9. For tables, use simple text formatting, not markdown tables + +Example of good formatting: +**Installation Steps** + +1. Update your package list: +```bash +sudo apt update +``` + +2. Install the package: +```bash +sudo apt install nginx +``` + +**Key Points** +- Point one here +- Point two here""" def _call_openai(self, question: str, system_prompt: str) -> str: response = self.client.chat.completions.create( @@ -222,7 +400,7 @@ def _call_openai(self, question: str, system_prompt: str) -> str: {"role": "user", "content": question}, ], temperature=0.3, - max_tokens=500, + max_tokens=2000, ) # Defensive: content may be None or choices could be empty in edge cases try: @@ -234,7 +412,7 @@ def _call_openai(self, question: str, system_prompt: str) -> str: def _call_claude(self, question: str, system_prompt: str) -> str: response = self.client.messages.create( model=self.model, - max_tokens=500, + max_tokens=2000, temperature=0.3, system=system_prompt, messages=[{"role": "user", "content": question}], @@ -344,4 +522,26 @@ def ask(self, question: str) -> str: except (OSError, sqlite3.Error): pass # Silently fail cache writes + # Track educational topics for learning history + self.learning_tracker.record_topic(question) + return answer + + def get_learning_history(self) -> dict[str, Any]: + """Get the user's learning history. + + Returns: + Dictionary with topics explored and statistics + """ + return self.learning_tracker.get_history() + + def get_recent_topics(self, limit: int = 5) -> list[str]: + """Get recently explored educational topics. + + Args: + limit: Maximum number of topics to return + + Returns: + List of topic strings + """ + return self.learning_tracker.get_recent_topics(limit) diff --git a/cortex/cli.py b/cortex/cli.py index 274a4f55..8c5f7d33 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -6,6 +6,8 @@ from datetime import datetime from typing import Any +from rich.markdown import Markdown + from cortex.ask import AskHandler from cortex.branding import VERSION, console, cx_header, cx_print, show_banner from cortex.coordinator import InstallationCoordinator, StepStatus @@ -297,7 +299,8 @@ def ask(self, question: str) -> int: offline=self.offline, ) answer = handler.ask(question) - console.print(answer) + # Render as markdown for proper formatting in terminal + console.print(Markdown(answer)) return 0 except ImportError as e: # Provide a helpful message if provider SDK is missing diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 6e4eea4e..7080146f 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -8,6 +8,7 @@ This document provides a comprehensive reference for all commands available in t |---------|-------------| | `cortex` | Show help and available commands | | `cortex install ` | Install software | +| `cortex ask ` | Ask questions about your system or learn about Linux | | `cortex demo` | See Cortex in action | | `cortex wizard` | Configure API key | | `cortex status` | Show comprehensive system status and health checks | @@ -71,6 +72,65 @@ cortex install "python3 with pip and virtualenv" --execute --- +### `cortex ask` + +Ask natural language questions about your system or learn about Linux, packages, and best practices. The AI automatically detects whether you're asking a diagnostic question about your system or an educational question to learn something new. + +**Usage:** +```bash +cortex ask "" +``` + +**Question Types:** + +**Diagnostic Questions** - Questions about your specific system: +```bash +# System status queries +cortex ask "why is my disk full" +cortex ask "what packages need updating" +cortex ask "is my Python version compatible with TensorFlow" +cortex ask "check my GPU drivers" +``` + +**Educational Questions** - Learn about Linux, packages, and best practices: +```bash +# Explanations and tutorials +cortex ask "explain how Docker containers work" +cortex ask "what is systemd and how do I use it" +cortex ask "teach me about nginx configuration" +cortex ask "best practices for securing a Linux server" +cortex ask "how to set up a Python virtual environment" +``` + +**Features:** +- **Automatic Intent Detection**: The AI distinguishes between diagnostic and educational queries +- **System-Aware Responses**: Uses your actual system context (OS, Python version, GPU, etc.) +- **Structured Learning**: Educational responses include examples, best practices, and related topics +- **Learning Progress Tracking**: Educational topics you explore are tracked in `~/.cortex/learning_history.json` +- **Response Caching**: Repeated questions return cached responses for faster performance + +**Examples:** +```bash +# Diagnostic: Get specific info about your system +cortex ask "what version of Python do I have" +cortex ask "can I run PyTorch on this system" + +# Educational: Learn with structured tutorials +cortex ask "explain how apt package management works" +cortex ask "what are best practices for Docker security" +cortex ask "guide to setting up nginx as a reverse proxy" + +# Mix of both +cortex ask "how do I install and configure Redis" +``` + +**Notes:** +- Educational responses are longer and include code examples with syntax highlighting +- The `--offline` flag can be used to only return cached responses +- Learning history helps track what topics you've explored over time + +--- + ### `cortex demo` Run an interactive demonstration of Cortex capabilities. Perfect for first-time users or presentations. @@ -366,6 +426,24 @@ cortex stack webdev --dry-run cortex stack webdev ``` +### Learning with Cortex Ask +```bash +# 1. Ask diagnostic questions about your system +cortex ask "what version of Python do I have" +cortex ask "is Docker installed" + +# 2. Learn about new topics with educational queries +cortex ask "explain how Docker containers work" +cortex ask "best practices for nginx configuration" + +# 3. Get step-by-step tutorials +cortex ask "teach me how to set up a Python virtual environment" +cortex ask "guide to configuring SSH keys" + +# 4. Your learning topics are automatically tracked +# View at ~/.cortex/learning_history.json +``` + --- ## Getting Help diff --git a/tests/test_ask.py b/tests/test_ask.py index aaa9a237..a1329462 100644 --- a/tests/test_ask.py +++ b/tests/test_ask.py @@ -1,14 +1,16 @@ """Unit tests for the ask module.""" +import json import os import sys import tempfile import unittest +from pathlib import Path from unittest.mock import MagicMock, patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from cortex.ask import AskHandler, SystemInfoGatherer +from cortex.ask import AskHandler, LearningTracker, SystemInfoGatherer class TestSystemInfoGatherer(unittest.TestCase): @@ -268,5 +270,229 @@ def test_unsupported_provider(self): AskHandler(api_key="test", provider="unsupported") +class TestLearningTracker(unittest.TestCase): + """Tests for LearningTracker.""" + + def setUp(self): + """Set up test fixtures.""" + self.temp_dir = tempfile.mkdtemp() + self.temp_file = Path(self.temp_dir) / "learning_history.json" + # Patch the PROGRESS_FILE to use temp location + self.patcher = patch.object(LearningTracker, "PROGRESS_FILE", self.temp_file) + self.patcher.start() + self.tracker = LearningTracker() + + def tearDown(self): + """Clean up temporary files.""" + import shutil + + self.patcher.stop() + if os.path.exists(self.temp_dir): + shutil.rmtree(self.temp_dir) + + def test_is_educational_query_explain(self): + """Test detection of 'explain' queries.""" + self.assertTrue(self.tracker.is_educational_query("explain how docker works")) + self.assertTrue(self.tracker.is_educational_query("Explain nginx configuration")) + + def test_is_educational_query_teach_me(self): + """Test detection of 'teach me' queries.""" + self.assertTrue(self.tracker.is_educational_query("teach me about systemd")) + self.assertTrue(self.tracker.is_educational_query("Teach me how to use git")) + + def test_is_educational_query_what_is(self): + """Test detection of 'what is' queries.""" + self.assertTrue(self.tracker.is_educational_query("what is kubernetes")) + self.assertTrue(self.tracker.is_educational_query("What are containers")) + + def test_is_educational_query_how_does(self): + """Test detection of 'how does' queries.""" + self.assertTrue(self.tracker.is_educational_query("how does DNS work")) + self.assertTrue(self.tracker.is_educational_query("How do containers work")) + + def test_is_educational_query_best_practices(self): + """Test detection of 'best practices' queries.""" + self.assertTrue(self.tracker.is_educational_query("best practices for security")) + self.assertTrue( + self.tracker.is_educational_query("what are best practice for nginx") + ) + + def test_is_educational_query_tutorial(self): + """Test detection of 'tutorial' queries.""" + self.assertTrue(self.tracker.is_educational_query("tutorial on docker compose")) + + def test_is_educational_query_non_educational(self): + """Test that non-educational queries return False.""" + self.assertFalse(self.tracker.is_educational_query("why is my disk full")) + self.assertFalse(self.tracker.is_educational_query("what packages need updating")) + self.assertFalse(self.tracker.is_educational_query("check my system status")) + + def test_extract_topic_explain(self): + """Test topic extraction from 'explain' queries.""" + topic = self.tracker.extract_topic("explain how docker containers work") + self.assertEqual(topic, "how docker containers work") + + def test_extract_topic_teach_me(self): + """Test topic extraction from 'teach me' queries.""" + topic = self.tracker.extract_topic("teach me about systemd services") + self.assertEqual(topic, "systemd services") + + def test_extract_topic_what_is(self): + """Test topic extraction from 'what is' queries.""" + topic = self.tracker.extract_topic("what is kubernetes?") + self.assertEqual(topic, "kubernetes") + + def test_extract_topic_truncation(self): + """Test that long topics are truncated.""" + long_question = "explain " + "a" * 100 + topic = self.tracker.extract_topic(long_question) + self.assertLessEqual(len(topic), 50) + + def test_record_topic_creates_file(self): + """Test that recording a topic creates the history file.""" + self.tracker.record_topic("explain docker") + self.assertTrue(self.temp_file.exists()) + + def test_record_topic_stores_data(self): + """Test that recorded topics are stored correctly.""" + self.tracker.record_topic("explain docker containers") + history = self.tracker.get_history() + self.assertIn("docker containers", history["topics"]) + self.assertEqual(history["topics"]["docker containers"]["count"], 1) + + def test_record_topic_increments_count(self): + """Test that repeated topics increment the count.""" + self.tracker.record_topic("explain docker") + self.tracker.record_topic("explain docker") + history = self.tracker.get_history() + self.assertEqual(history["topics"]["docker"]["count"], 2) + + def test_record_topic_ignores_non_educational(self): + """Test that non-educational queries are not recorded.""" + self.tracker.record_topic("why is my disk full") + history = self.tracker.get_history() + self.assertEqual(len(history["topics"]), 0) + + def test_get_recent_topics(self): + """Test getting recent topics.""" + self.tracker.record_topic("explain docker") + self.tracker.record_topic("what is kubernetes") + self.tracker.record_topic("teach me nginx") + + recent = self.tracker.get_recent_topics(limit=2) + self.assertEqual(len(recent), 2) + # Most recent should be first + self.assertEqual(recent[0], "nginx") + + def test_get_recent_topics_empty(self): + """Test getting recent topics when none exist.""" + recent = self.tracker.get_recent_topics() + self.assertEqual(recent, []) + + def test_total_queries_tracked(self): + """Test that total educational queries are tracked.""" + self.tracker.record_topic("explain docker") + self.tracker.record_topic("what is kubernetes") + history = self.tracker.get_history() + self.assertEqual(history["total_queries"], 2) + + +class TestAskHandlerLearning(unittest.TestCase): + """Tests for AskHandler learning features.""" + + def setUp(self): + """Set up test fixtures.""" + self.temp_dir = tempfile.mkdtemp() + self.temp_file = Path(self.temp_dir) / "learning_history.json" + self.patcher = patch.object(LearningTracker, "PROGRESS_FILE", self.temp_file) + self.patcher.start() + + def tearDown(self): + """Clean up temporary files.""" + import shutil + + self.patcher.stop() + if os.path.exists(self.temp_dir): + shutil.rmtree(self.temp_dir) + + def test_ask_records_educational_topic(self): + """Test that educational questions are recorded.""" + os.environ["CORTEX_FAKE_RESPONSE"] = "Docker is a containerization platform..." + handler = AskHandler(api_key="fake-key", provider="fake") + handler.cache = None + + handler.ask("explain how docker works") + + history = handler.get_learning_history() + self.assertIn("how docker works", history["topics"]) + + def test_ask_does_not_record_diagnostic(self): + """Test that diagnostic questions are not recorded.""" + os.environ["CORTEX_FAKE_RESPONSE"] = "Your disk is 80% full." + handler = AskHandler(api_key="fake-key", provider="fake") + handler.cache = None + + handler.ask("why is my disk full") + + history = handler.get_learning_history() + self.assertEqual(len(history["topics"]), 0) + + def test_get_recent_topics_via_handler(self): + """Test getting recent topics through handler.""" + os.environ["CORTEX_FAKE_RESPONSE"] = "Test response" + handler = AskHandler(api_key="fake-key", provider="fake") + handler.cache = None + + handler.ask("explain kubernetes") + handler.ask("what is docker") + + recent = handler.get_recent_topics(limit=5) + self.assertEqual(len(recent), 2) + + def test_system_prompt_contains_educational_instructions(self): + """Test that system prompt includes educational guidance.""" + handler = AskHandler(api_key="fake-key", provider="fake") + context = handler.info_gatherer.gather_context() + prompt = handler._get_system_prompt(context) + + self.assertIn("Educational Questions", prompt) + self.assertIn("Diagnostic Questions", prompt) + self.assertIn("tutorial-style", prompt) + self.assertIn("best practices", prompt) + + +class TestSystemPromptEnhancement(unittest.TestCase): + """Tests for enhanced system prompt.""" + + def test_prompt_includes_query_type_detection(self): + """Test that prompt includes query type detection section.""" + handler = AskHandler(api_key="fake-key", provider="fake") + context = {"python_version": "3.11", "os": {"system": "Linux"}} + prompt = handler._get_system_prompt(context) + + self.assertIn("Query Type Detection", prompt) + self.assertIn("explain", prompt.lower()) + self.assertIn("teach me", prompt.lower()) + + def test_prompt_includes_educational_instructions(self): + """Test that prompt includes educational response instructions.""" + handler = AskHandler(api_key="fake-key", provider="fake") + context = {} + prompt = handler._get_system_prompt(context) + + self.assertIn("code examples", prompt.lower()) + self.assertIn("best practices", prompt.lower()) + self.assertIn("related topics", prompt.lower()) + + def test_prompt_includes_diagnostic_instructions(self): + """Test that prompt includes diagnostic response instructions.""" + handler = AskHandler(api_key="fake-key", provider="fake") + context = {} + prompt = handler._get_system_prompt(context) + + self.assertIn("system context", prompt.lower()) + self.assertIn("actionable", prompt.lower()) + + if __name__ == "__main__": unittest.main() From 1cfb6509df3c0b4e94b6d313ab34a193f3ab8de9 Mon Sep 17 00:00:00 2001 From: Shree Jejurikar Date: Mon, 29 Dec 2025 15:28:14 +0530 Subject: [PATCH 2/7] style: Format code with black and ruff --- cortex/ask.py | 6 ++---- tests/test_ask.py | 4 +--- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/cortex/ask.py b/cortex/ask.py index d1078065..87cf166c 100644 --- a/cortex/ask.py +++ b/cortex/ask.py @@ -160,9 +160,7 @@ class LearningTracker: def __init__(self): """Initialize the learning tracker.""" - self._compiled_patterns = [ - re.compile(p, re.IGNORECASE) for p in self.EDUCATIONAL_PATTERNS - ] + self._compiled_patterns = [re.compile(p, re.IGNORECASE) for p in self.EDUCATIONAL_PATTERNS] def is_educational_query(self, question: str) -> bool: """Determine if a question is educational in nature.""" @@ -249,7 +247,7 @@ def _load_history(self) -> dict[str, Any]: return {"topics": {}, "total_queries": 0} try: - with open(self.PROGRESS_FILE, "r") as f: + with open(self.PROGRESS_FILE) as f: return json.load(f) except (json.JSONDecodeError, OSError): return {"topics": {}, "total_queries": 0} diff --git a/tests/test_ask.py b/tests/test_ask.py index a1329462..a8f53983 100644 --- a/tests/test_ask.py +++ b/tests/test_ask.py @@ -313,9 +313,7 @@ def test_is_educational_query_how_does(self): def test_is_educational_query_best_practices(self): """Test detection of 'best practices' queries.""" self.assertTrue(self.tracker.is_educational_query("best practices for security")) - self.assertTrue( - self.tracker.is_educational_query("what are best practice for nginx") - ) + self.assertTrue(self.tracker.is_educational_query("what are best practice for nginx")) def test_is_educational_query_tutorial(self): """Test detection of 'tutorial' queries.""" From f4abbed3c0f41dab8bcf2e63e737be7f4ae619ab Mon Sep 17 00:00:00 2001 From: Shree Jejurikar Date: Mon, 12 Jan 2026 11:08:45 +0530 Subject: [PATCH 3/7] fix(ask): Track learning topics for cached responses Record topic access even when returning cached responses to ensure last_accessed timestamps and counts are updated correctly. --- cortex/ask.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cortex/ask.py b/cortex/ask.py index 06ca8d34..3d58e588 100644 --- a/cortex/ask.py +++ b/cortex/ask.py @@ -493,6 +493,8 @@ def ask(self, question: str) -> str: system_prompt=system_prompt, ) if cached is not None and len(cached) > 0: + # Track topic access even for cached responses + self.learning_tracker.record_topic(question) return cached[0] # Call LLM From 11e7b35206d4e21784946c9e5bbb13964162e51f Mon Sep 17 00:00:00 2001 From: Shree Date: Mon, 12 Jan 2026 11:38:42 +0530 Subject: [PATCH 4/7] fix(ask): Use lazy initialization for LearningTracker progress file Convert PROGRESS_FILE from class-level constant to lazily-computed property to avoid RuntimeError in restricted environments where Path.home() may fail at import time. Falls back to temp directory. --- cortex/ask.py | 25 ++++++++++++++++++++----- tests/test_ask.py | 13 ++++++------- 2 files changed, 26 insertions(+), 12 deletions(-) diff --git a/cortex/ask.py b/cortex/ask.py index 3d58e588..56c06c60 100644 --- a/cortex/ask.py +++ b/cortex/ask.py @@ -141,7 +141,7 @@ def gather_context(self) -> dict[str, Any]: class LearningTracker: """Tracks educational topics the user has explored.""" - PROGRESS_FILE = Path.home() / ".cortex" / "learning_history.json" + _progress_file: Path | None = None # Patterns that indicate educational questions EDUCATIONAL_PATTERNS = [ @@ -164,6 +164,21 @@ def __init__(self): """Initialize the learning tracker.""" self._compiled_patterns = [re.compile(p, re.IGNORECASE) for p in self.EDUCATIONAL_PATTERNS] + @property + def progress_file(self) -> Path: + """Lazily compute the progress file path to avoid import-time errors.""" + if self._progress_file is None: + try: + self._progress_file = Path.home() / ".cortex" / "learning_history.json" + except RuntimeError: + # Fallback for restricted environments where home is inaccessible + import tempfile + + self._progress_file = ( + Path(tempfile.gettempdir()) / ".cortex" / "learning_history.json" + ) + return self._progress_file + def is_educational_query(self, question: str) -> bool: """Determine if a question is educational in nature.""" for pattern in self._compiled_patterns: @@ -245,11 +260,11 @@ def get_recent_topics(self, limit: int = 5) -> list[str]: def _load_history(self) -> dict[str, Any]: """Load learning history from file.""" - if not self.PROGRESS_FILE.exists(): + if not self.progress_file.exists(): return {"topics": {}, "total_queries": 0} try: - with open(self.PROGRESS_FILE) as f: + with open(self.progress_file) as f: return json.load(f) except (json.JSONDecodeError, OSError): return {"topics": {}, "total_queries": 0} @@ -257,8 +272,8 @@ def _load_history(self) -> dict[str, Any]: def _save_history(self, history: dict[str, Any]) -> None: """Save learning history to file.""" try: - self.PROGRESS_FILE.parent.mkdir(parents=True, exist_ok=True) - with open(self.PROGRESS_FILE, "w") as f: + self.progress_file.parent.mkdir(parents=True, exist_ok=True) + with open(self.progress_file, "w") as f: json.dump(history, f, indent=2) except OSError: pass # Silently fail if we can't write diff --git a/tests/test_ask.py b/tests/test_ask.py index e4bf8394..5f6da44f 100644 --- a/tests/test_ask.py +++ b/tests/test_ask.py @@ -296,16 +296,14 @@ def setUp(self): """Set up test fixtures.""" self.temp_dir = tempfile.mkdtemp() self.temp_file = Path(self.temp_dir) / "learning_history.json" - # Patch the PROGRESS_FILE to use temp location - self.patcher = patch.object(LearningTracker, "PROGRESS_FILE", self.temp_file) - self.patcher.start() self.tracker = LearningTracker() + # Set the progress file to use temp location + self.tracker._progress_file = self.temp_file def tearDown(self): """Clean up temporary files.""" import shutil - self.patcher.stop() if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) @@ -421,14 +419,11 @@ def setUp(self): """Set up test fixtures.""" self.temp_dir = tempfile.mkdtemp() self.temp_file = Path(self.temp_dir) / "learning_history.json" - self.patcher = patch.object(LearningTracker, "PROGRESS_FILE", self.temp_file) - self.patcher.start() def tearDown(self): """Clean up temporary files.""" import shutil - self.patcher.stop() if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) @@ -437,6 +432,8 @@ def test_ask_records_educational_topic(self): os.environ["CORTEX_FAKE_RESPONSE"] = "Docker is a containerization platform..." handler = AskHandler(api_key="fake-key", provider="fake") handler.cache = None + # Set the progress file to use temp location + handler.learning_tracker._progress_file = self.temp_file handler.ask("explain how docker works") @@ -448,6 +445,7 @@ def test_ask_does_not_record_diagnostic(self): os.environ["CORTEX_FAKE_RESPONSE"] = "Your disk is 80% full." handler = AskHandler(api_key="fake-key", provider="fake") handler.cache = None + handler.learning_tracker._progress_file = self.temp_file handler.ask("why is my disk full") @@ -459,6 +457,7 @@ def test_get_recent_topics_via_handler(self): os.environ["CORTEX_FAKE_RESPONSE"] = "Test response" handler = AskHandler(api_key="fake-key", provider="fake") handler.cache = None + handler.learning_tracker._progress_file = self.temp_file handler.ask("explain kubernetes") handler.ask("what is docker") From 0440f8db517356a5cc2979233f4c2e732d7f104c Mon Sep 17 00:00:00 2001 From: Shree Date: Tue, 13 Jan 2026 13:17:44 +0530 Subject: [PATCH 5/7] fix: Address code quality and robustness issues in cortex ask module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit resolves 18 outstanding issues identified in code review: CODE QUALITY IMPROVEMENTS: - Removed unused 'import json' from test_ask.py - Added return type hints (-> None) to LearningTracker.__init__ - Added type annotation for _compiled_patterns: list[re.Pattern[str]] - Enhanced __init__ docstring with pattern compilation details DESIGN & LOGIC ENHANCEMENTS: - Refactored _compiled_patterns as class variable to avoid regex recompilation - Improved topic truncation logic with explicit boundary checking (len > 50) - Switched to UTC timestamps (datetime.now(timezone.utc).isoformat()) - Simplified is_educational_query() using any() for idiomatic Python ERROR HANDLING & ROBUSTNESS: - Added explicit UTF-8 encoding to json.load() and json.dump() - Added debug-level logging for save failures without breaking CLI - Fixed test environment pollution by cleaning up CORTEX_FAKE_RESPONSE - Verified lazy Path.home() initialization with RuntimeError fallback CONSISTENCY & DOCUMENTATION: - Removed obsolete --offline flag documentation reference - Replaced markdown headings with bold text in system prompt for consistency - Added num_predict: 2000 to Ollama options matching OpenAI/Claude tokens - Added edge case tests for malformed JSON and Path.home() failures - Added concurrency documentation notes in record_topic() method - Formatted all changes with Black and Ruff TESTING: - All 51 tests passing (+ 2 new edge case tests) - Black formatter: ✓ All 129 files properly formatted - Ruff linter: ✓ All checks passed Closes: Related to enhanced cortex ask with tutor capabilities --- cortex/ask.py | 88 ++++++++++++++++++++++++++++++++++------------- docs/COMMANDS.md | 1 - tests/test_ask.py | 36 +++++++++++++++++-- 3 files changed, 98 insertions(+), 27 deletions(-) diff --git a/cortex/ask.py b/cortex/ask.py index 56c06c60..f14a3cdc 100644 --- a/cortex/ask.py +++ b/cortex/ask.py @@ -6,18 +6,22 @@ """ import json +import logging import os import platform import re import shutil import sqlite3 import subprocess -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from typing import Any from cortex.config_utils import get_ollama_model +# Module logger for debug diagnostics +logger = logging.getLogger(__name__) + class SystemInfoGatherer: """Gathers local system information for context-aware responses.""" @@ -160,9 +164,18 @@ class LearningTracker: r"^basics\s+of\b", ] - def __init__(self): - """Initialize the learning tracker.""" - self._compiled_patterns = [re.compile(p, re.IGNORECASE) for p in self.EDUCATIONAL_PATTERNS] + # Compiled patterns shared across all instances for efficiency + _compiled_patterns: list[re.Pattern[str]] = [ + re.compile(p, re.IGNORECASE) for p in EDUCATIONAL_PATTERNS + ] + + def __init__(self) -> None: + """Initialize the learning tracker. + + Uses pre-compiled educational patterns for efficient matching + across multiple queries. Patterns are shared as class variables + to avoid recompilation overhead. + """ @property def progress_file(self) -> Path: @@ -181,10 +194,7 @@ def progress_file(self) -> Path: def is_educational_query(self, question: str) -> bool: """Determine if a question is educational in nature.""" - for pattern in self._compiled_patterns: - if pattern.search(question): - return True - return False + return any(pattern.search(question) for pattern in self._compiled_patterns) def extract_topic(self, question: str) -> str: """Extract the main topic from an educational question.""" @@ -211,13 +221,28 @@ def extract_topic(self, question: str) -> str: # Clean up and truncate topic = topic.strip("? ").strip() - # Take first 50 chars as topic identifier + + # Truncate at word boundaries to keep topic identifier meaningful + # If topic exceeds 50 chars, truncate at the last space within those 50 chars + # to preserve whole words. If the first 50 chars contain no spaces, + # keep the full 50-char prefix. if len(topic) > 50: - topic = topic[:50].rsplit(" ", 1)[0] + truncated = topic[:50] + # Try to split at word boundary; keep full 50 chars if no spaces found + words = truncated.rsplit(" ", 1) + topic = words[0] + return topic def record_topic(self, question: str) -> None: - """Record that the user explored an educational topic.""" + """Record that the user explored an educational topic. + + Note: This method performs a read-modify-write cycle on the history file + without file locking. If multiple cortex ask processes run concurrently, + concurrent updates could theoretically be lost. This is acceptable for a + single-user CLI tool where concurrent invocations are rare and learning + history is non-critical, but worth noting for future enhancements. + """ if not self.is_educational_query(question): return @@ -227,15 +252,18 @@ def record_topic(self, question: str) -> None: history = self._load_history() + # Use UTC timestamps for consistency and accurate sorting + utc_now = datetime.now(timezone.utc).isoformat() + # Update or add topic if topic in history["topics"]: history["topics"][topic]["count"] += 1 - history["topics"][topic]["last_accessed"] = datetime.now().isoformat() + history["topics"][topic]["last_accessed"] = utc_now else: history["topics"][topic] = { "count": 1, - "first_accessed": datetime.now().isoformat(), - "last_accessed": datetime.now().isoformat(), + "first_accessed": utc_now, + "last_accessed": utc_now, } history["total_queries"] = history.get("total_queries", 0) + 1 @@ -264,19 +292,28 @@ def _load_history(self) -> dict[str, Any]: return {"topics": {}, "total_queries": 0} try: - with open(self.progress_file) as f: + with open(self.progress_file, encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError): return {"topics": {}, "total_queries": 0} def _save_history(self, history: dict[str, Any]) -> None: - """Save learning history to file.""" + """Save learning history to file. + + Silently handles save failures to keep CLI clean, but logs at debug level + for diagnostics. Failures may occur due to permission issues or disk space. + """ try: self.progress_file.parent.mkdir(parents=True, exist_ok=True) - with open(self.progress_file, "w") as f: + with open(self.progress_file, "w", encoding="utf-8") as f: json.dump(history, f, indent=2) - except OSError: - pass # Silently fail if we can't write + except OSError as e: + # Log at debug level to help diagnose permission/disk issues + # without breaking CLI output or crashing the application + logger.debug( + f"Failed to save learning history to {self.progress_file}: {e}", + exc_info=False, + ) class AskHandler: @@ -358,11 +395,12 @@ def _get_system_prompt(self, context: dict[str, Any]) -> str: System Context: {json.dumps(context, indent=2)} -## Query Type Detection +**Query Type Detection** Automatically detect the type of question and respond appropriately: -### Educational Questions (tutorials, explanations, learning) +**Educational Questions (tutorials, explanations, learning)** + Triggered by questions like: "explain...", "teach me...", "how does X work", "what is...", "best practices for...", "tutorial on...", "learn about...", "guide to..." For educational questions: @@ -374,7 +412,8 @@ def _get_system_prompt(self, context: dict[str, Any]) -> str: 6. Mention related topics the user might want to explore next 7. Tailor examples to the user's system when relevant (e.g., use apt for Debian-based systems) -### Diagnostic Questions (system-specific, troubleshooting) +**Diagnostic Questions (system-specific, troubleshooting)** + Triggered by questions about: current system state, "why is my...", "what packages...", "check my...", specific errors, system status For diagnostic questions: @@ -383,7 +422,8 @@ def _get_system_prompt(self, context: dict[str, Any]) -> str: 3. Be concise but informative 4. If you don't have enough information, say so clearly -## Output Formatting Rules (CRITICAL - Follow exactly) +**Output Formatting Rules (CRITICAL - Follow exactly)** + 1. NEVER use markdown headings (# or ##) - they render poorly in terminals 2. For section titles, use **Bold Text** on its own line instead 3. Use bullet points (-) for lists @@ -455,7 +495,7 @@ def _call_ollama(self, question: str, system_prompt: str) -> str: "model": self.model, "prompt": prompt, "stream": False, - "options": {"temperature": 0.3}, + "options": {"temperature": 0.3, "num_predict": 2000}, } ).encode("utf-8") diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 05e7ec60..a2cab0d1 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -127,7 +127,6 @@ cortex ask "how do I install and configure Redis" **Notes:** - Educational responses are longer and include code examples with syntax highlighting -- The `--offline` flag can be used to only return cached responses - Learning history helps track what topics you've explored over time --- diff --git a/tests/test_ask.py b/tests/test_ask.py index 5f6da44f..0b7d97a4 100644 --- a/tests/test_ask.py +++ b/tests/test_ask.py @@ -1,6 +1,5 @@ """Unit tests for the ask module.""" -import json import os import sys import tempfile @@ -411,6 +410,34 @@ def test_total_queries_tracked(self): history = self.tracker.get_history() self.assertEqual(history["total_queries"], 2) + def test_load_history_with_malformed_json(self): + """Test that _load_history gracefully handles malformed JSON.""" + # Write invalid JSON to the file + self.tracker._progress_file = self.temp_file + self.temp_file.parent.mkdir(parents=True, exist_ok=True) + with open(self.temp_file, "w") as f: + f.write("{invalid json}") + + # Should return empty history instead of crashing + history = self.tracker._load_history() + self.assertEqual(history, {"topics": {}, "total_queries": 0}) + + def test_progress_file_fallback_on_home_error(self): + """Test that progress_file property falls back to tempdir when Path.home() fails.""" + tracker = LearningTracker() + + # Mock Path.home() to raise RuntimeError + with patch("pathlib.Path.home", side_effect=RuntimeError("HOME not set")): + # Reset the cached progress file so it re-evaluates + tracker._progress_file = None + + # Should fall back to tempfile location without crashing + progress_file = tracker.progress_file + self.assertIsNotNone(progress_file) + self.assertIn("cortex", str(progress_file)) + # Verify it's in the temp directory, not trying to use home + self.assertIn(tempfile.gettempdir(), str(progress_file)) + class TestAskHandlerLearning(unittest.TestCase): """Tests for AskHandler learning features.""" @@ -421,12 +448,17 @@ def setUp(self): self.temp_file = Path(self.temp_dir) / "learning_history.json" def tearDown(self): - """Clean up temporary files.""" + """Clean up temporary files and environment variables.""" import shutil if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) + # Clean up any environment variables set during tests to prevent pollution + # when test order changes or new tests are added + if "CORTEX_FAKE_RESPONSE" in os.environ: + del os.environ["CORTEX_FAKE_RESPONSE"] + def test_ask_records_educational_topic(self): """Test that educational questions are recorded.""" os.environ["CORTEX_FAKE_RESPONSE"] = "Docker is a containerization platform..." From 3990c62eab13e1c37819c2669d321f5c3f9b9d12 Mon Sep 17 00:00:00 2001 From: Shree Date: Tue, 13 Jan 2026 13:29:00 +0530 Subject: [PATCH 6/7] fix(demo): use gpu_info consistently instead of gpu variable - Renamed 'gpu' variable to 'gpu_info' for better clarity - Now consistently uses 'gpu_info' throughout instead of 'gpu' - Improves code readability and consistency --- cortex/demo.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/cortex/demo.py b/cortex/demo.py index 0f01108b..d9b0610e 100644 --- a/cortex/demo.py +++ b/cortex/demo.py @@ -17,15 +17,15 @@ def run_demo() -> int: print(f"✔ CPU: {hw.cpu.model}") print(f"✔ RAM: {hw.memory.total_gb} GB") - gpu = hw.gpu - if gpu and len(gpu) > 0: - print(f"✔ GPU: {gpu[0].model}") + gpu_info = hw.gpu + if gpu_info and len(gpu_info) > 0: + print(f"✔ GPU: {gpu_info[0].model}") else: print("⚠️ GPU: Not detected (CPU mode enabled)") # 2️⃣ Model Recommendations print("\n🤖 Model Recommendations:") - if gpu and len(gpu) > 0: + if gpu_info and len(gpu_info) > 0: print("• LLaMA-3-8B → Optimized for your GPU") print("• Mistral-7B → High performance inference") else: From 4532c9dccd671234c28e46a03f6a5f71859381ed Mon Sep 17 00:00:00 2001 From: Shree Date: Tue, 13 Jan 2026 13:32:44 +0530 Subject: [PATCH 7/7] Revert "fix(demo): use gpu_info consistently instead of gpu variable" This reverts commit 3990c62eab13e1c37819c2669d321f5c3f9b9d12. --- cortex/demo.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/cortex/demo.py b/cortex/demo.py index d9b0610e..0f01108b 100644 --- a/cortex/demo.py +++ b/cortex/demo.py @@ -17,15 +17,15 @@ def run_demo() -> int: print(f"✔ CPU: {hw.cpu.model}") print(f"✔ RAM: {hw.memory.total_gb} GB") - gpu_info = hw.gpu - if gpu_info and len(gpu_info) > 0: - print(f"✔ GPU: {gpu_info[0].model}") + gpu = hw.gpu + if gpu and len(gpu) > 0: + print(f"✔ GPU: {gpu[0].model}") else: print("⚠️ GPU: Not detected (CPU mode enabled)") # 2️⃣ Model Recommendations print("\n🤖 Model Recommendations:") - if gpu_info and len(gpu_info) > 0: + if gpu and len(gpu) > 0: print("• LLaMA-3-8B → Optimized for your GPU") print("• Mistral-7B → High performance inference") else: