[Feature] cortex ask --do - AI-Powered Command Execution with Auto-Repair
#663
Conversation
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughAdds a full "ask --do" feature: CLI flag and interactive execution mode, a Do Runner package (models, executor, diagnosis_v2, verification, terminal monitor, managers, DB), watch service and monitoring, setup scripts/docs, demo simplification, README updates, and re-export wrappers to expose the new public API surface. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant CLI as CLI Layer
participant LLM as LLM (Claude/Ollama)
participant Executor as TaskExecutor
participant Diagnosis as DiagnosisEngine
participant DB as DoRunDatabase
User->>CLI: cortex ask --do "install nginx"
CLI->>LLM: generate command plan (intent → commands)
LLM-->>CLI: command list
CLI->>Executor: build TaskTree
Executor->>Executor: run pre-flight ConflictDetector checks
CLI->>User: show commands for approval
User->>CLI: approve
loop execute tasks
Executor->>Executor: execute command
alt success
Executor->>DB: log command success
else failure
Executor->>Diagnosis: diagnose error (stderr/stdout)
Diagnosis->>LLM: request fix-plan (if needed)
LLM-->>Diagnosis: fix plan
Diagnosis-->>Executor: repair commands
Executor->>Executor: attempt auto-fix and retry
end
end
Executor->>DB: save run/session
CLI->>LLM: request summary
LLM-->>CLI: summary
CLI->>User: display results
sequenceDiagram
participant Terminal as User Terminal
participant Hook as Watch Hook
participant WatchLog as Watch Log File
participant Monitor as TerminalMonitor
participant LLM2 as Optional LLM
participant Cortex as Cortex Session
Cortex->>Monitor: start monitoring (expected commands)
Monitor->>Monitor: discover terminal sources & watch file
Terminal->>Hook: user runs command (sudo in other terminal)
Hook->>WatchLog: append command + metadata
Monitor->>WatchLog: poll and parse new entries
Monitor->>Monitor: match against expected commands
alt match success
Monitor->>Cortex: mark command observed
else error or unmatched
Monitor->>LLM2: analyze error (if configured)
LLM2-->>Monitor: suggestion/fix
Monitor->>Terminal: notify user
end
Monitor->>Cortex: provide observed commands and context
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120+ minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Summary of ChangesHello @Dhruv-89, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances Cortex by integrating an AI-powered command execution and auto-repair system. It transforms the Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
There was a problem hiding this comment.
Pull request overview
This PR introduces cortex ask --do, an AI-powered command execution feature with auto-repair capabilities, intelligent error handling, resource conflict detection, and real-time terminal monitoring.
Changes:
- Adds interactive AI assistant mode for executing Linux commands based on natural language requests
- Implements auto-repair system for common command failures
- Integrates terminal monitoring service for real-time command tracking across sessions
Reviewed changes
Copilot reviewed 19 out of 25 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| scripts/setup_ask_do.sh | Bash setup script for installing Ollama, watch service, and shell hooks |
| scripts/setup_ask_do.py | Python setup script providing programmatic installation of all components |
| requirements.txt | Adds pydantic dependency for data validation |
| docs/ASK_DO_ARCHITECTURE.md | Comprehensive architecture documentation for the new feature |
| cortex/watch_service.py | Background daemon for monitoring terminal activity across sessions |
| cortex/system_info_generator.py | LLM-based system information retrieval with read-only command validation |
| cortex/semantic_cache.py | Minor improvement to error message clarity |
| cortex/do_runner/verification.py | Resource conflict detection and verification testing system |
| cortex/do_runner/terminal.py | Real-time terminal monitoring with AI-powered error analysis |
| cortex/do_runner/models.py | Data models for task trees, command logs, and execution tracking |
| cortex/do_runner/managers.py | User privilege and protected path management |
| cortex/do_runner/executor.py | Task tree executor with auto-repair capabilities |
| cortex/do_runner/Untitled | Empty file to be removed |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
Feature Video Screencast.from.21-01-26.05.53.02.PM.IST.webm |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the impressive cortex ask --do feature, which enables AI-powered command execution with auto-repair and terminal monitoring. The architecture is well-thought-out, with a clear separation of concerns into different modules like diagnosis, verification, and execution. The interactive session and signal handling in the CLI are also great additions. I've identified a few critical security vulnerabilities related to command injection that need to be addressed, as well as a suggestion for improving code maintainability in the main CLI file. Overall, this is a fantastic and significant feature addition.
I am having trouble creating individual review comments. Click here to see my feedback.
cortex/do_runner/executor.py (243-249)
Executing commands received from an LLM using shell=True without proper validation is a critical security risk. A compromised or malicious LLM response could lead to arbitrary command execution on the user's system. While terminal.py includes a _is_safe_fix_command check, that protection is missing here.
Before executing the command, you should validate it against a security policy. You could adapt the CommandValidator from cortex.ask or, even better, reuse the _is_safe_fix_command logic from cortex/do_runner/terminal.py to ensure only safe commands are executed.
cortex/do_runner/diagnosis_v2.py (1474-1476)
The execute_fix_commands function constructs a command by substituting variables directly into a command template using string.replace(). The variable values can originate from various sources, including the user's query or LLM output, and are not sanitized. This creates a command injection vulnerability. For example, if a variable contained a value like "; rm -rf /, it would be executed as part of the command when shell=True is used.
To prevent this, you must sanitize all variable values before substituting them into the command string. The standard and safest way to do this in Python is with shlex.quote().
import shlex
for var_name, value in resolved_variables.items():
command = command.replace(f"{{{var_name}}}", shlex.quote(value))
cortex/do_runner/terminal.py (1547-1550)
The DANGEROUS_PATTERNS list is a great security measure. However, the patterns for blocking command piping to a shell are a bit too specific and can be bypassed. For example, curl ... | /bin/bash or curl ... | zsh would not be caught.
To make this more robust, consider using a regex pattern that blocks piping to common shells, regardless of their path, as recommended by the repository's general rules.
r'|\s*(?:.*/)?(?:bash|sh|zsh)\b', # Block piping to common shells
References
- When blacklisting dangerous shell commands, include a regex pattern to block piping content to common shells like bash, sh, and zsh, even when specified with a path (e.g.,
r'|\s*(?:.*/)?(?:bash|sh|zsh)\b')
cortex/cli.py (851-1129)
This function, _run_interactive_do_session, is quite long (nearly 300 lines) and handles many different responsibilities, including session setup, signal handling, UI rendering, and command processing logic for history, help, clear, etc. This reduces readability and makes the function difficult to maintain and test.
Consider refactoring this into several smaller, more focused helper methods. For example:
_setup_interactive_session(): To handle the initial setup and welcome panel._handle_session_command(query, ...): To process meta-commands likehistory,clear, andhelp._process_user_query(query, handler, ...): For the main logic of passing the query to theAskHandler._render_session_history(history): To encapsulate therich.tablelogic for displaying history.
Breaking down this function will improve modularity and make the interactive session flow easier to follow.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
🤖 Fix all issues with AI agents
In `@cortex/do_runner/executor.py`:
- Around line 248-274: The _execute_command method currently auto-prepends sudo
and uses shell=True which causes silent privilege escalation and injection
risks; remove the automatic sudo insertion in _execute_command and instead have
it detect sudo need via _needs_sudo and return a clear signal (e.g., raise a
specific exception or include a flag in the return tuple) so the caller can
prompt the user for explicit approval before re-invoking the command with sudo;
when executing, avoid shell=True by tokenizing the command with shlex.split()
and calling subprocess.run(..., shell=False, args=cmd_list), validate/sanitize
input tokens when applicable, and preserve existing timeout/exception handling
while returning descriptive error messages for caller-driven sudo escalation.
In `@cortex/do_runner/managers.py`:
- Around line 138-186: The create_user method currently runs multiple sudo
commands without asking for consent; before executing any privilege-escalation
subprocess calls in create_user, call the handler's request_user_confirmation()
(the same pattern used elsewhere), and if it returns False, immediately return
(False, "User declined privilege escalation") or similar; leave the existing
user_exists check intact and only prompt when user creation is needed, then
proceed with the subprocess.run calls when confirmation is granted.
In `@cortex/do_runner/terminal.py`:
- Around line 1489-1512: The code currently attempts to auto-run a
non‑interactive sudo probe via subprocess.run when needs_sudo is true; remove
that probe and never attempt to execute sudo automatically. In the block where
needs_sudo = fix_cmd.strip().startswith("sudo "), immediately append fix_cmd to
sudo_commands_pending, append (fix_cmd, "sudo", None) to results, print the
console message (using the existing console.print call that shows "(needs
sudo)"), and continue; delete the subprocess.run(...) check and its try/except
so no automatic sudo check/execution occurs. Ensure the changes are applied to
the logic surrounding needs_sudo, fix_cmd, sudo_commands_pending, results, and
the console.print call.
In `@cortex/system_info_generator.py`:
- Around line 518-534: The codebase uses httpx in cortex/licensing.py
(module-level import) and dynamically in cortex/system_info_generator.py when
provider == "ollama" (the block referencing self.ollama_url and self.model), but
httpx is not declared as a project dependency; add httpx to the project's
dependency manifest (e.g., pyproject.toml [tool.poetry.dependencies] or
requirements.txt) so importing cortex.licensing or using provider="ollama" won't
raise ImportError, and update lock files/install accordingly.
In `@requirements.txt`:
- Around line 19-21: requirements.txt removed PyYAML but multiple modules import
it (cortex/config_manager.py, cortex/i18n/translator.py, cortex/i18n/config.py,
cortex/permission_manager.py, cortex/dashboard.py), causing ImportError at
runtime; restore PyYAML to requirements.txt (add a dependency like PyYAML with
an appropriate version specifier, e.g., PyYAML>=5.4) so the YAML imports succeed
and CI/runtime installs the package. Ensure the package name matches PyPI casing
(PyYAML) and commit the updated requirements.txt.
🟠 Major comments (18)
cortex/system_info_generator.py-149-152 (1)
149-152: Template commands conflict withCommandValidatorallowlist.
ufw statusis blocked andaa-statusis not in the allowed list, so these templates will always fail. Either update the validator to allow these read‑only commands or replace them with approved alternatives.cortex/watch_service.py-68-73 (1)
68-73: Harden log file permissions to avoid leaking command history.These logs can contain sensitive commands; with a typical umask they may be world‑readable. Ensure
.cortexand its log files are restricted (e.g., dir0700, files0600).🔒 Example fix (apply to all log files)
def log(self, message: str): """Log a message to the service log.""" log_file = self.cortex_dir / "watch_service.log" timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(log_file, "a") as f: f.write(f"[{timestamp}] {message}\n") + os.chmod(log_file, 0o600)Also applies to: 392-394, 434-435
cortex/do_runner/terminal.py-717-739 (1)
717-739: Don’t overwrite existingPROMPT_COMMAND.The hook currently replaces any existing
PROMPT_COMMAND, which can break user shells. Preserve and append instead, and avoid re‑adding if already present.✅ Proposed fix
- echo "${{tty_name:-unknown}}|$cmd" >> {watch_file} -}} -export PROMPT_COMMAND='history -a; __cortex_log_cmd' + echo "${{tty_name:-unknown}}|$cmd" >> {watch_file} +}} +if [[ -z "$PROMPT_COMMAND" ]]; then + export PROMPT_COMMAND='history -a; __cortex_log_cmd' +elif [[ "$PROMPT_COMMAND" != *"__cortex_log_cmd"* ]]; then + export PROMPT_COMMAND="${{PROMPT_COMMAND}}; __cortex_log_cmd" +ficortex/watch_service.py-524-536 (1)
524-536: Ensurestop()only returns success if the daemon actually exited.After waiting, the code returns success even if the PID still exists.
✅ Proposed fix
for _ in range(10): try: os.kill(pid, 0) time.sleep(0.5) except ProcessLookupError: break - - return True, f"Service stopped (PID: {pid})" + try: + os.kill(pid, 0) + return False, f"Service still running (PID: {pid})" + except ProcessLookupError: + return True, f"Service stopped (PID: {pid})"cortex/do_runner/terminal.py-1027-1056 (1)
1027-1056: Parse the newTTY|COMMANDlog format.When the watch service or hook writes
tty|command, the monitor currently treats the entire line as the command, which pollutes command detection. Split and use the command part.✅ Proposed fix
- # Handle format with timestamp: "HH:MM:SS command" + # Handle format with terminal prefix: "TTY|COMMAND" + if "|" in line and not re.match(r"^\d{2}:\d{2}:\d{2}\s+", line): + parts = line.split("|", 1) + if len(parts) == 2 and parts[1].strip(): + line = parts[1].strip() + + # Handle format with timestamp: "HH:MM:SS command" if re.match(r"^\d{2}:\d{2}:\d{2}\s+", line): parts = line.split(" ", 1) if len(parts) == 2 and parts[1].strip(): self._process_observed_command(parts[1].strip(), "live_terminal") else: # Plain command self._process_observed_command(line, "live_terminal")cortex/do_runner/terminal.py-2230-2284 (1)
2230-2284: Avoidshell=Truewith user‑derived tokens in verification commands.
service_name/container_nameare derived from observed commands; building shell strings can execute unintended tokens (e.g., embedded;). Use argv lists and trim output in Python.✅ Safer pattern (example for systemctl)
- check_cmd = f"systemctl status {service_name} 2>&1 | head -5" + safe_service = re.sub(r"[^a-zA-Z0-9_.@-]", "", service_name) + check_cmd = ["systemctl", "status", safe_service] @@ - try: - result = subprocess.run( - check_cmd, shell=True, capture_output=True, text=True, timeout=5 - ) - - output = result.stdout + result.stderr + try: + result = subprocess.run( + check_cmd, capture_output=True, text=True, timeout=5 + ) + output = "\n".join((result.stdout + result.stderr).splitlines()[:5])cortex/system_info_generator.py-327-362 (1)
327-362: Add missing return type hints for__init__and_initialize_client.Both methods lack return type annotations. Per coding guidelines requiring type hints on all Python code, add
-> None:to these methods.Proposed fix
def __init__( self, api_key: str | None = None, provider: str = "claude", model: str | None = None, debug: bool = False, - ): + ) -> None: """ Initialize the system info generator.- def _initialize_client(self): + def _initialize_client(self) -> None: """Initialize the LLM client."""cortex/watch_service.py-30-98 (1)
30-98: Add explicit return type hints to all public and internal methods.The module contains 15 methods lacking return type annotations, violating the project's typing requirement. Add return types to:
__init__(→None),_handle_signal(→None),_handle_reload(→None),log(→None),_load_state(→None),_save_state(→None),_monitor_bash_history(→None),_monitor_watch_hook(→None),_log_to_json(→None),_log_command(→None),_cleanup_stale_terminals(→None),start(→bool),_shutdown(→None),stop(→tuple[bool, str]), andmain(→None). Additionally, the signal handler methods are missing parameter type hints forsignumandframe.cortex/do_runner/models.py-90-97 (1)
90-97: Dead code:get_depth()contains an infinite loop and is never used.The method assigns
node = node(line 96), which will cause an infinite loop ifparent_idis set. Additionally, codebase search shows this method is never called anywhere, indicating it is dead code.The fundamental issue is that
TaskNodehas no reference to its parent node or to the containingTaskTree, making it impossible for this method to traverse up the tree correctly. SinceTaskTreemaintains a complete_all_tasksdictionary, the most practical solution is to remove this broken method fromTaskNodeentirely, or move the depth-calculation logic toTaskTreeas a utility method if needed in the future.Recommendation: Remove the unused method unless there is a planned feature that requires it. If depth calculation becomes necessary, implement it on
TaskTreeusing the_all_taskslookup mechanism.cortex/do_runner/managers.py-220-240 (1)
220-240: Security: chmod fallback grants overly permissive access.The
_grant_privilege_chmodfallback useso+rw(others read/write), which grants access to ALL users on the system, not just the cortex user. This is significantly more permissive than the intendedsetfaclbehavior and creates a security risk.Consider using group-based permissions as a safer fallback:
Suggested safer fallback
`@classmethod` def _grant_privilege_chmod(cls, file_path: str, mode: str) -> tuple[bool, str]: """Fallback privilege granting using chmod.""" try: - chmod_mode = "" - if "r" in mode: - chmod_mode = "o+r" - if "w" in mode: - chmod_mode = "o+rw" if chmod_mode else "o+w" - if "x" in mode: - chmod_mode = chmod_mode + "x" if chmod_mode else "o+x" + # Use group permissions instead of world-readable/writable + chmod_mode = "g+" + if "r" in mode: + chmod_mode += "r" + if "w" in mode: + chmod_mode += "w" + if "x" in mode: + chmod_mode += "x" + if chmod_mode == "g+": + chmod_mode = "g+r" # Default to read subprocess.run( - ["sudo", "chmod", chmod_mode, file_path], + ["sudo", "chgrp", cls.CORTEX_GROUP, file_path], + check=True, + capture_output=True, + ) + subprocess.run( + ["sudo", "chmod", chmod_mode, file_path], check=True, capture_output=True, ) - return True, f"Granted {mode} access to {file_path} (chmod fallback)" + return True, f"Granted {mode} access to {file_path} (group chmod fallback)"scripts/setup_ask_do.py-289-312 (1)
289-312: Preserve existingPROMPT_COMMANDinstead of overwriting.Both the hook file and
.bashrcblock replacePROMPT_COMMAND, which can break existing shell integrations (direnv, pyenv, starship, etc.). Compose with the existing value instead.🛠️ Safer PROMPT_COMMAND composition
@@ -export PROMPT_COMMAND='history -a; __cortex_log_cmd' +export PROMPT_COMMAND="history -a; __cortex_log_cmd${PROMPT_COMMAND:+; $PROMPT_COMMAND}"@@ -export PROMPT_COMMAND='history -a; __cortex_log_cmd' +export PROMPT_COMMAND="history -a; __cortex_log_cmd${{PROMPT_COMMAND:+; $PROMPT_COMMAND}}"Also applies to: 328-347
scripts/setup_ask_do.py-43-68 (1)
43-68: Add explicit return type hints for public helpers andmain().The new helpers are public APIs in this module, but they omit return type hints (e.g.,
print_*,main). Please add-> None/-> intconsistently. As per coding guidelines, type hints are required in Python code.🧷 Example adjustments
-def print_header(text: str): +def print_header(text: str) -> None: @@ -def print_step(text: str): +def print_step(text: str) -> None: @@ -def print_success(text: str): +def print_success(text: str) -> None: @@ -def print_warning(text: str): +def print_warning(text: str) -> None: @@ -def print_error(text: str): +def print_error(text: str) -> None: @@ -def main(): +def main() -> int:Also applies to: 544-562
scripts/setup_ask_do.py-70-87 (1)
70-87: Handle missing command binaries to avoid crashing the setup wizard.
run_cmddoesn’t catchFileNotFoundError, so calls withcheck=False(e.g.,systemctl,cortex) can still raise and abort the script. Suggest handling this explicitly so the wizard degrades gracefully.🧯 Suggested fix
@@ - except subprocess.TimeoutExpired: + except FileNotFoundError: + print_error(f"Command not found: {' '.join(cmd)}") + if check: + raise + stdout = "" if capture else None + stderr = "command not found" if capture else None + return subprocess.CompletedProcess(cmd, 127, stdout=stdout, stderr=stderr) + except subprocess.TimeoutExpired: print_error(f"Command timed out: {' '.join(cmd)}") raisecortex/do_runner/verification.py-193-208 (1)
193-208: Potential command injection via unsanitizedcontainer_name.The
container_nameextracted from regex (line 190) is directly interpolated into shell commands (lines 194, 200, 206). If the inputcmdcontains crafted values like--name "foo; malicious_cmd", this could lead to command injection.Consider validating/sanitizing extracted values or using parameterized execution where possible.
🛡️ Suggested mitigation
import shlex # Validate container name format (alphanumeric, dashes, underscores) if not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_.-]*$', container_name): return result # Invalid container name, skip check # Use shlex.quote for shell safety safe_name = shlex.quote(container_name) success, container_id, _ = self._execute_command( f"docker ps -aq --filter name=^{safe_name}$", needs_sudo=False )cortex/do_runner/verification.py-23-34 (1)
23-34: Logic error: Double sudo execution whenneeds_sudo=True.When
needs_sudo=True, line 24-25 prepends"sudo "tocmd, then line 27-28 runs["sudo", "bash", "-c", cmd]. This results in executingsudo bash -c "sudo <original_cmd>"— double sudo invocation.🐛 Proposed fix
def _execute_command( self, cmd: str, needs_sudo: bool = False, timeout: int = 120 ) -> tuple[bool, str, str]: """Execute a single command.""" try: - if needs_sudo and not cmd.strip().startswith("sudo"): - cmd = f"sudo {cmd}" - - result = subprocess.run( - ["sudo", "bash", "-c", cmd] if needs_sudo else cmd, - shell=not needs_sudo, - capture_output=True, - text=True, - timeout=timeout, - ) + if needs_sudo: + result = subprocess.run( + ["sudo", "bash", "-c", cmd], + capture_output=True, + text=True, + timeout=timeout, + ) + else: + result = subprocess.run( + cmd, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + ) return result.returncode == 0, result.stdout.strip(), result.stderr.strip()cortex/do_runner/verification.py-602-607 (1)
602-607: Destructive default action:fuser -kkills processes without confirmation.The
stop_existingalternative action usessudo fuser -k {port}/tcpwhich forcefully terminates processes. This could disrupt critical services. Consider using a less aggressive approach or adding a warning.💡 Suggested safer alternative
{ "action": "stop_existing", - "description": f"Stop process using port {port}", - "commands": [f"sudo fuser -k {port}/tcp"], + "description": f"Stop process using port {port} (WARNING: may disrupt services)", + "commands": [ + f"sudo lsof -i :{port} -t", # First identify the process + f"# Review above PID, then: sudo kill <PID>", + ], },cortex/do_runner/diagnosis_v2.py-1462-1471 (1)
1462-1471: Security concern: Password exposed in pip config and process arguments.Line 1463 embeds credentials in a URL that gets written to pip config (visible in plaintext). Line 1470-1471 substitutes password into a command string executed via shell, which could expose it in process listings.
🔒 Suggested safer approach for pip
elif "pip" in command.lower() or "pypi" in host.lower(): - login_cmd = f"pip config set global.index-url https://{username}:{{password}}@pypi.org/simple/" + # Use environment variable or keyring instead of embedding in config + console.print("[yellow]For PyPI, consider using keyring or environment variables:[/yellow]") + console.print(" export TWINE_USERNAME=<username>") + console.print(" export TWINE_PASSWORD=<password>") + login_cmd = None # Don't store password in configcortex/do_runner/verification.py-1044-1053 (1)
1044-1053: Bug:operation_typeis overwritten in loop, only last value used.The
operation_typevariable is reassigned on each iteration (line 1052), so after the loop, only the last matched operation type is retained. This affects the logic at line 1102 whereoperation_typeis checked for alltarget_files.🐛 Proposed fix: Track operation type per file
- target_files = [] - operation_type = None + target_files: list[tuple[str, str]] = [] # (path, operation_type) for pattern, op_type in file_creation_patterns: matches = re.findall(pattern, cmd) for match in matches: if match.startswith("/") or match.startswith("~"): - target_files.append(match) - operation_type = op_type + target_files.append((match, op_type)) - result["files_checked"] = target_files + result["files_checked"] = [f[0] for f in target_files] - for file_path in target_files: + for file_path, operation_type in target_files:
🟡 Minor comments (11)
cortex/do_runner/Untitled-1-1 (1)
1-1: Remove accidental placeholder file.This file contains only a stray character and should be deleted to avoid shipping meaningless artifacts.
🧹 Proposed fix
-icortex/demo.py-49-55 (1)
49-55: Remove trailing whitespace (Ruff W291) on Line 54.This currently fails linting.
🧹 Proposed fix
-• **Install** - Install packages with AI interpretation +• **Install** - Install packages with AI interpretationcortex/do_runner/terminal.py-185-305 (1)
185-305: Add explicit return type hints for public methods.The public methods
__init__andstartin the specified range (lines 185-305) are missing return type annotations. Additionally,start_monitoring(line 319) lacks a return type hint. All three should be annotated with-> Nonefor consistency with coding guidelines requiring type hints on all public APIs.scripts/setup_ask_do.sh-257-260 (1)
257-260: Remove unused variableCORTEX_PATH.The static analysis correctly identified that
CORTEX_PATHis assigned but never used. This should be removed to avoid confusion.Suggested fix
# Get Python path PYTHON_PATH=$(which python3) - CORTEX_PATH=$(which cortex 2>/dev/null || echo "$HOME/.local/bin/cortex")cortex/do_runner/models.py-267-269 (1)
267-269: Minor: Command truncation always appends "..." even for short commands.The print output unconditionally appends
...after truncating to 50 characters, which will display asshort_cmd...even when the command is shorter than 50 characters.Suggested fix
- console.print( - f"{indent}{prefix}{icon} [{color}][{node.task_type.value}][/{color}] {node.command[:50]}..." - ) + cmd_display = node.command[:50] + "..." if len(node.command) > 50 else node.command + console.print( + f"{indent}{prefix}{icon} [{color}][{node.task_type.value}][/{color}] {cmd_display}" + )cortex/do_runner/managers.py-259-270 (1)
259-270: chmod revoke fallback may remove unrelated permissions.Similar to the grant fallback,
chmod o-rwxremoves permissions for ALL "others", not just the cortex user. This could unintentionally break access for other legitimate users or services.cortex/do_runner/database.py-129-157 (1)
129-157: Fix trailing whitespace flagged by linter.Static analysis reports trailing whitespace on lines 130, 148, 149, and 150. Remove the trailing spaces to pass the lint check.
Fix trailing whitespace
cursor = conn.execute(""" - SELECT run_id, full_data FROM do_runs + SELECT run_id, full_data FROM do_runs WHERE total_commands IS NULL OR total_commands = 0 OR commands_list IS NULL """) # ... lines 146-156 ... conn.execute( """ - UPDATE do_runs SET - total_commands = ?, - successful_commands = ?, + UPDATE do_runs SET + total_commands = ?, + successful_commands = ?, failed_commands = ?,cortex/do_runner/database.py-276-298 (1)
276-298: Missing null check forfull_databefore JSON parsing.If
full_dataisNULLin the database,json.loads(row["full_data"])at line 284 will raiseTypeError. Add defensive handling.Suggested fix
if row: + if not row["full_data"]: + return None full_data = json.loads(row["full_data"])cortex/do_runner/managers.py-56-56 (1)
56-56: Class attributeUSER_PROTECTED_PATHSis shared across all instances.
USER_PROTECTED_PATHSis defined as a class attribute (set[str] = set()), meaning it's shared across all instances ofProtectedPathsManager. The_load_user_pathsmethod assigns toself.USER_PROTECTED_PATHS, which shadows the class attribute. This could lead to unexpected behavior if multiple instances are created.Suggested fix
class ProtectedPathsManager: """Manages the list of protected files and folders requiring user authentication.""" SYSTEM_PROTECTED_PATHS: set[str] = { # ... system paths ... } - USER_PROTECTED_PATHS: set[str] = set() - def __init__(self): + self.user_protected_paths: set[str] = set() self.config_file = Path.home() / ".cortex" / "protected_paths.json" self._ensure_config_dir() self._load_user_paths()Then update all references from
self.USER_PROTECTED_PATHStoself.user_protected_paths.cortex/cli.py-3972-3975 (1)
3972-3975: Truncate run output to avoid terminal flooding.
cmd["output"]/cmd["error"]can be large; the fallback path already truncates. Consider a small cap here too.✂️ Suggested truncation
@@ - if cmd["output"]: - console.print(f" [dim]Output:[/dim] {cmd['output']}") - if cmd["error"]: - console.print(f" [red]Error:[/red] {cmd['error']}") + if cmd["output"]: + output = cmd["output"] + output = output[:250] + "..." if len(output) > 250 else output + console.print(f" [dim]Output:[/dim] {output}") + if cmd["error"]: + err = cmd["error"] + err = err[:250] + "..." if len(err) > 250 else err + console.print(f" [red]Error:[/red] {err}")cortex/do_runner/verification.py-302-305 (1)
302-305: Incomplete filtering for system paths.The check
if path in ["/dev/null", "/etc/os-release", "/proc/", "/sys/"]uses exact match, but paths like/proc/cpuinfoor/sys/class/netwon't match. Usestartswith()for directory prefixes.🐛 Proposed fix
for path in paths_in_cmd: # Skip common read paths - if path in ["/dev/null", "/etc/os-release", "/proc/", "/sys/"]: + if path == "/dev/null" or path == "/etc/os-release" or path.startswith("/proc/") or path.startswith("/sys/"): continue
🧹 Nitpick comments (20)
cortex/demo.py (1)
16-24: Add a timeout to subprocess calls to avoid hanging the demo.If
cortexblocks, the demo becomes unresponsive; a reasonable timeout makes it safer.⏱️ Example adjustment
- result = subprocess.run(cmd, capture_output=True, text=True) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) @@ - result = subprocess.run(cmd) + result = subprocess.run(cmd, timeout=60)docs/ASK_DO_ARCHITECTURE.md (1)
451-463: Consider adding language specifiers to log format examples.The static analysis tool flagged missing language specifiers on fenced code blocks. While ASCII diagrams don't benefit from syntax highlighting, the log file format examples could use
textorlogspecifiers for consistency.Suggested fix
-``` +```text pts_1|docker ps pts_1|sudo systemctl restart nginx-```json +```json {"timestamp": "2026-01-16T14:15:00.123", ...}scripts/setup_ask_do.sh (1)
303-356: Code duplication: Watch hook logic defined twice.The watch hook function
__cortex_log_cmdis duplicated—once inwatch_hook.sh(lines 303-325) and again inline in.bashrc(lines 334-356). This duplication could lead to maintenance issues if the logic needs updating.Consider having
.bashrcsource the hook file instead of duplicating the code:Suggested approach
# Add to .bashrc MARKER="# Cortex Terminal Watch Hook" if [ -f ~/.bashrc ]; then if ! grep -q "$MARKER" ~/.bashrc; then print_step "Adding hook to .bashrc..." cat >> ~/.bashrc << 'EOF' # Cortex Terminal Watch Hook -__cortex_last_histnum="" -__cortex_log_cmd() { - # ... duplicated code ... -} -export PROMPT_COMMAND='history -a; __cortex_log_cmd' +if [ -f "$HOME/.cortex/watch_hook.sh" ]; then + source "$HOME/.cortex/watch_hook.sh" +fi alias cw="source ~/.cortex/watch_hook.sh" EOFcortex/do_runner/executor.py (4)
37-49: Type hint foruser_managerparameter is incorrect.The
user_managerparameter is typed astype(a metaclass), but based on usage it should be typed as the actual manager class or instance. Looking at the code, this appears to be passed but never used in this file.Suggested fix
def __init__( self, - user_manager: type, + user_manager: Any, paths_manager: Any, llm_callback: Callable[[str], dict] | None = None, ):Alternatively, if
CortexUserManageris the expected type, import and use it directly for better type safety.
330-419: Move repeatedimport reto module level.The
remodule is imported three times inside conditional branches (lines 341, 367, 394). This is inefficient and unconventional. Move the import to the top of the file with other imports.Suggested fix
Add at top of file:
import reThen remove the inline imports at lines 341, 367, and 394.
54-64: Consider validating command entries before adding to tree.Empty or malformed command entries (missing "command" key) will result in empty strings being added to the task tree. Consider adding validation.
Suggested improvement
def build_tree_from_commands( self, commands: list[dict[str, str]], ) -> TaskTree: """Build a task tree from a list of commands.""" for cmd in commands: - self.tree.add_root_task( - command=cmd.get("command", ""), - purpose=cmd.get("purpose", ""), - ) + command = cmd.get("command", "").strip() + if command: # Skip empty commands + self.tree.add_root_task( + command=command, + purpose=cmd.get("purpose", ""), + ) return self.tree
114-246: Dead code:was_repairedvariable is unused.The
was_repairedvariable is initialized at line 121 but is never set toTrue. When repairs succeed, the method returns(True, True)directly (lines 192, 221, 241), and the final return at line 246 always returns(False, was_repaired)which is(False, False).Additionally, lines 169-192 and 194-221 contain duplicated repair execution logic that could be refactored.
Suggested fix for dead variable
Either remove
was_repairedentirely and simplify the returns, or use it properly:- was_repaired = False # ... existing code ... - return False, was_repaired + return False, Falsecortex/do_runner/managers.py (1)
104-116: Minor redundancy inis_protectedlogic.Line 113 checks
path == protected, but this case is already handled by the exact match check at line 109. The redundant check can be removed.Simplified version
def is_protected(self, path: str) -> bool: """Check if a path requires authentication for access.""" path = os.path.abspath(path) all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS if path in all_protected: return True for protected in all_protected: - if path.startswith(protected + "/") or path == protected: + if path.startswith(protected + "/"): return True return Falsecortex/do_runner/database.py (3)
122-127: Silent exception handling may hide migration errors.The
except sqlite3.OperationalError: passat lines 126-127 silently ignores all operational errors during schema migration. While this handles the case where a column already exists, it also hides legitimate errors (e.g., disk full, database corruption).Consider more specific handling
if col_name not in existing_columns: try: conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}") - except sqlite3.OperationalError: - pass + except sqlite3.OperationalError as e: + # Column likely already exists from a previous migration + if "duplicate column" not in str(e).lower(): + console.print(f"[yellow]Warning: Schema migration issue: {e}[/yellow]")
404-416: Session ID generation has lower entropy than run ID generation.
create_sessionusesmd5with a predictable timestamp, while_generate_run_idusessha256withos.urandom(16). For consistency and better uniqueness guarantees, consider using the same approach.Suggested improvement
def create_session(self) -> str: """Create a new session and return the session ID.""" - session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{hashlib.md5(str(datetime.datetime.now().timestamp()).encode()).hexdigest()[:8]}" + timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S%f") + random_part = hashlib.sha256(os.urandom(16)).hexdigest()[:8] + session_id = f"session_{timestamp}_{random_part}"
378-402: Add defensive handling for corruptedfull_datain bulk retrieval.
get_recent_runsiterates rows and parsesfull_datawithout exception handling. A single corrupted record would cause the entire method to fail. Consider wrapping in try-except to skip malformed entries.Suggested improvement
runs = [] for row in cursor: - full_data = json.loads(row["full_data"]) - run = DoRun( - run_id=full_data["run_id"], - # ... - ) - run.session_id = row["session_id"] - runs.append(run) + try: + if not row["full_data"]: + continue + full_data = json.loads(row["full_data"]) + run = DoRun( + run_id=full_data["run_id"], + summary=full_data["summary"], + mode=RunMode(full_data["mode"]), + commands=[CommandLog.from_dict(c) for c in full_data["commands"]], + started_at=full_data.get("started_at", ""), + completed_at=full_data.get("completed_at", ""), + user_query=full_data.get("user_query", ""), + files_accessed=full_data.get("files_accessed", []), + privileges_granted=full_data.get("privileges_granted", []), + ) + run.session_id = row["session_id"] + runs.append(run) + except (json.JSONDecodeError, KeyError, TypeError): + continue # Skip malformed records return runscortex/do_runner/verification.py (3)
19-38: Duplicate_execute_commandmethod across three classes.The
_execute_commandmethod is identically implemented inConflictDetector,VerificationRunner, andFileUsefulnessAnalyzer. This violates DRY and creates maintenance burden.♻️ Suggested refactor: Extract to a base class or helper module
+class CommandExecutor: + """Base class providing command execution capabilities.""" + + def _execute_command( + self, cmd: str, needs_sudo: bool = False, timeout: int = 120 + ) -> tuple[bool, str, str]: + """Execute a single command.""" + try: + if needs_sudo and not cmd.strip().startswith("sudo"): + cmd = f"sudo {cmd}" + + result = subprocess.run( + ["sudo", "bash", "-c", cmd] if needs_sudo else cmd, + shell=not needs_sudo, + capture_output=True, + text=True, + timeout=timeout, + ) + return result.returncode == 0, result.stdout.strip(), result.stderr.strip() + except subprocess.TimeoutExpired: + return False, "", f"Command timed out after {timeout} seconds" + except Exception as e: + return False, "", str(e) + + -class ConflictDetector: +class ConflictDetector(CommandExecutor): """Detects conflicts with existing configurations.""" - - def _execute_command( - self, cmd: str, needs_sudo: bool = False, timeout: int = 120 - ) -> tuple[bool, str, str]: - ...Also applies to: 817-836, 997-1016
932-950: Arbitrary limit of 5 files may miss important verification targets.The
[:5]slice on line 932 limits file existence checks to 5 files. This arbitrary limit could skip verification of important configuration files. Consider making this configurable or documenting the rationale.+ MAX_FILE_CHECKS = 5 # Class constant, configurable + # File existence tests - for file_path in list(files_to_check)[:5]: + for file_path in list(files_to_check)[:self.MAX_FILE_CHECKS]:
1222-1231: Magic number for keyword overlap threshold.The threshold
> 2on line 1227 is a magic number. Consider extracting it as a class constant for clarity and configurability.+ MIN_KEYWORD_OVERLAP = 2 # Minimum keyword matches to consider content related + # Generic - check for keyword overlap else: purpose_words = set(purpose_lower.split()) content_words = set(content_lower.split()) overlap = purpose_words & content_words - if len(overlap) > 2: + if len(overlap) > self.MIN_KEYWORD_OVERLAP:cortex/do_runner/diagnosis_v2.py (6)
20-20: Unused import:Callable.The
Callabletype fromcollections.abcis imported but not used in this file.-from collections.abc import Callable
488-490: API key may not match selected provider.The API key is loaded from
ANTHROPIC_API_KEYorOPENAI_API_KEYwithout checking if it matches theproviderparameter. If user setsprovider="openai"but onlyANTHROPIC_API_KEYexists, the wrong key will be used.💡 Suggested improvement
- self.api_key = ( - api_key or os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY") - ) + if api_key: + self.api_key = api_key + elif self.provider == "claude": + self.api_key = os.environ.get("ANTHROPIC_API_KEY") + elif self.provider == "openai": + self.api_key = os.environ.get("OPENAI_API_KEY") + else: + self.api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")
1705-1711: Hardcoded timeout values should be configurable.The 120-second timeout is hardcoded in multiple places (lines 1711, 1777, and also in
execute_fix_commandsat line 1555). Consider making this a class constant or constructor parameter for configurability.class DiagnosisEngine: MAX_FIX_ATTEMPTS = 5 MAX_STACK_DEPTH = 10 + DEFAULT_TIMEOUT = 120 # Then use self.DEFAULT_TIMEOUT instead of hardcoded 120Also applies to: 1771-1777
1802-1823: Missing error handling for LLM API calls.The
_call_llmmethod doesn't handle API-specific errors like rate limits, context length exceeded, or network timeouts. While the caller has a broadtry/except, specific handling would improve reliability and user feedback.💡 Suggested error handling
def _call_llm(self, system_prompt: str, user_prompt: str) -> str: """Call the LLM and return response text.""" + if not self.client: + raise ValueError("LLM client not initialized") + if self.provider == "claude": - response = self.client.messages.create( - model=self.model, - max_tokens=2048, - system=system_prompt, - messages=[{"role": "user", "content": user_prompt}], - ) - return response.content[0].text + try: + response = self.client.messages.create( + model=self.model, + max_tokens=2048, + system=system_prompt, + messages=[{"role": "user", "content": user_prompt}], + ) + return response.content[0].text + except Exception as e: + if "rate" in str(e).lower(): + raise RuntimeError(f"Rate limit exceeded: {e}") from e + raise
1893-1899: Redundant API key loading in factory function.The
get_diagnosis_enginefunction loads API keys from environment variables, butDiagnosisEngine.__init__already does this whenapi_key=None. The factory function can be simplified.def get_diagnosis_engine( provider: str = "claude", debug: bool = False, ) -> DiagnosisEngine: """Factory function to create a DiagnosisEngine.""" - api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY") - return DiagnosisEngine(api_key=api_key, provider=provider, debug=debug) + return DiagnosisEngine(provider=provider, debug=debug)
1906-1908: Unused import in test block.
import syson line 1907 is not used in the test block.if __name__ == "__main__": - import sys - console.print("[bold]Diagnosis Engine Test[/bold]\n")
Dhruv-89
force-pushed
the
do-ask
branch
2 times, most recently
from
366ae66 to
7f1dec5
Compare
There was a problem hiding this comment.
Actionable comments posted: 18
🤖 Fix all issues with AI agents
In `@cortex/demo.py`:
- Around line 49-56: The multi-line string assigned to intro contains trailing
whitespace on the line with "• **Install** - Install packages with AI
interpretation" which triggers Ruff W291; open the intro triple-quoted string
(variable intro) and remove any trailing spaces at the end of that line so the
line ends immediately after the text, then save to clear the lint warning.
- Around line 27-37: The helper functions _wait_for_enter and _section are
missing explicit return type annotations; add "-> None" to both function
definitions (i.e., change def _wait_for_enter(): and def _section(title: str,
problem: str): to include "-> None") so they conform to the project's typing
guidelines, leaving the implementations unchanged.
In `@cortex/do_runner/database.py`:
- Around line 129-131: The SQL strings passed to conn.execute (the SELECT:
"SELECT run_id, full_data FROM do_runs WHERE total_commands IS NULL OR
total_commands = 0 OR commands_list IS NULL" and the subsequent UPDATE
statement) contain trailing whitespace characters flagged by static analysis;
remove any trailing spaces/newlines at the ends of those multi-line SQL lines so
the SQL literals (used by cursor and conn.execute) have no trailing whitespace,
and re-run linters to confirm lines inside the SELECT/UPDATE blocks (the
multi-line strings around the SELECT and UPDATE statements) are cleaned.
In `@cortex/do_runner/diagnosis_v2.py`:
- Line 1458: The subprocess.run call invoking "git config --global
credential.helper store" enables plaintext credential storage; modify the code
around that subprocess.run invocation to first warn the user via the existing
logger (or raise a confirmation prompt) about the security risk of storing
credentials in ~/.git-credentials, and instead default to a safer alternative
(e.g., use "git config --global credential.helper cache" or prefer a platform
credential helper like "manager" or "osxkeychain"), or only set "store" when an
explicit opt-in flag is provided; ensure the warning message is clear and
include which helper will be configured and how to revert it.
- Line 1463: The code constructs login_cmd embedding plaintext credentials into
a pip index URL via the login_cmd variable (currently f"pip config set
global.index-url https://{username}:{{password}}@pypi.org/simple/"), which must
be removed; instead set the index URL without credentials and supply auth via a
secure mechanism (e.g., set PIP_INDEX_URL as an environment variable at runtime
or configure authentication through a keyring/netrc or a pypirc token) so
secrets are never written into pip config; update the code that builds login_cmd
to use "https://pypi.org/simple/" (no credentials) and ensure subsequent install
or environment setup reads credentials from an env var (PIP_INDEX_URL) or a
keyring helper rather than embedding username/password into the URL.
- Around line 1287-1327: The apt-file subprocess invocation in diagnosis_v2.py
(inside the block handling var_name == "package") builds a shell string using
cmd_name derived from command, which can be manipulated; replace this with a
safe approach by validating/sanitizing cmd_name (allow only [A-Za-z0-9._+-] and
reject/return if it fails) and call subprocess.run with a list argument (e.g.,
["apt-file","search","--regexp", f"bin/{cmd_name}$"]) or better yet use
shutil.which(cmd_name) and the existing package_map before invoking apt-file;
update the code paths that call subprocess.run here to avoid shell=True and to
handle errors/timeouts robustly.
- Around line 1549-1556: The execute_fix_commands function currently calls
subprocess.run(command, shell=True) which risks shell injection when variable
values are substituted; change it to build a safe argument list and call
subprocess.run with shell=False (or pass a list) by tokenizing the command using
shlex.split (or by assembling arguments explicitly) and ensure any injected
variable values are shlex.quote()-escaped before insertion; update the call
sites in execute_fix_commands that construct the command string so they produce
a list of args (or safely quoted values) and use subprocess.run(...,
shell=False, capture_output=True, text=True, timeout=120).
In `@cortex/do_runner/executor.py`:
- Around line 142-208: The auto-repair flow currently only gates on
_identify_paths_needing_privileges() causing commands that require sudo (e.g.,
apt, systemctl, useradd) to run without explicit approval; update the logic
around fix_commands handling to also check each command with
_needs_sudo(command) or detect explicit "sudo" and treat those as privileged,
and require confirm_callback approval before executing any repair_task that
needs sudo (i.e., do not auto-run privileged repairs when confirm_callback is
None); apply this check where repair_tasks are created and before calling
_execute_task_with_repair and _execute_command so privileged fixes are skipped
or deferred unless confirm_callback approves.
In `@cortex/do_runner/managers.py`:
- Around line 220-240: _grant_privilege_chmod currently uses "o+..." which makes
files world-accessible; change the fallback to avoid granting permissions to
"others" by first attempting to set ownership to the "cortex" user/group (use
subprocess.run to call chown on file_path to cortex:cortex) and then use chmod
with user/group qualifiers (u+ or g+ derived from the requested mode string)
instead of o+ so only the cortex user/group gets the access; if chown fails,
fall back to the least-privileged chmod (u+...) and add a warning log that a
broad fallback was required. Ensure you update the subprocess.run calls that
reference file_path and mode in _grant_privilege_chmod and include clear logging
for failures.
In `@cortex/do_runner/models.py`:
- Around line 349-360: In get_commands_log_string, the output line always
appends "..." even when cmd.output is shorter than the 500-char slice; change
the logic to check the length of cmd.output and log either the full output or
the first 500 characters followed by "..." when truncated (refer to
get_commands_log_string and cmd.output), and ensure you handle None safely
before slicing.
- Around line 90-97: get_depth() currently loops forever because inside the
while it does node = node instead of moving to the parent; update get_depth to
reassign node to its parent each iteration (e.g., node = node.parent) or, if
this model only stores parent_id, look up the parent node by id (via the
TaskTree or the appropriate repository method) and assign that to node until
parent_id is falsy; ensure you reference the get_depth method and the parent_id
field when making the change so the traversal actually ascends the tree.
In `@cortex/do_runner/terminal.py`:
- Around line 714-739: The setup_auto_watch() hook currently overwrites
PROMPT_COMMAND; change it to preserve and append to any existing PROMPT_COMMAND
instead of replacing it: detect the current PROMPT_COMMAND value (may be empty),
build a composed command that runs the original PROMPT_COMMAND then runs
"history -a; __cortex_log_cmd" (ensuring proper quoting/escaping), and export
that composed value; update the install string that defines __cortex_log_cmd and
the export line so it references and composes with the preexisting
PROMPT_COMMAND safely (keep references to __cortex_log_cmd, PROMPT_COMMAND, and
setup_auto_watch() to locate the change).
In `@cortex/do_runner/verification.py`:
- Around line 19-33: In _execute_command: remove the code that prepends "sudo"
to the cmd string when needs_sudo is True (the if needs_sudo and not
cmd.strip().startswith("sudo"): cmd = f"sudo {cmd}" block) so you do not produce
"sudo sudo ..."; instead pass the original cmd into subprocess.run and rely on
the existing wrapper invocation (["sudo", "bash", "-c", cmd] when needs_sudo) to
provide a single sudo layer; apply this same change to the other helper
functions in this file that use the same pattern (the other needs_sudo command
invocations around lines similar to the current diff) so no function ever
prepends sudo before using the ["sudo","bash","-c", cmd] wrapper.
In `@cortex/semantic_cache.py`:
- Around line 83-89: The fallback path assignment currently sets self.db_path to
~/.cortex/cache.db without verifying that the fallback directory (user_dir) is
writable; modify the except block handling PermissionError/OSError so that after
creating user_dir (Path.home()/".cortex") you also check writability
(os.access(user_dir, os.W_OK)) and if it is not writable raise a clear
PermissionError (e.g., "Fallback directory {user_dir} is not writable"); ensure
mkdir remains inside the try/except that can surface OSError if creation fails,
and only assign self.db_path when the writability check passes so failures are
fast and explicit (references: variable db_dir, user_dir, and attribute
self.db_path).
In `@cortex/watch_service.py`:
- Line 624: The subprocess.run call that executes loginctl enable-linger
currently uses os.getenv("USER", "") which can pass an empty username; change
this to obtain a valid username (e.g. use getpass.getuser() or
pwd.getpwuid(os.getuid()).pw_name) and only call
subprocess.run(["loginctl","enable-linger", username], capture_output=True) when
username is non-empty; update the code around the subprocess.run invocation (the
loginctl enable-linger call in watch_service.py) to compute and validate
username first and skip or log an error if none can be determined.
- Around line 397-407: The _is_duplicate method mutates shared list
_recent_commands and can be called from multiple threads (_monitor_bash_history,
_monitor_watch_hook); protect these mutations with a mutex by adding a
threading.Lock (e.g., self._recent_commands_lock) during class initialization
and wrap the entire body of _is_duplicate in a with self._recent_commands_lock:
block so the membership check, append, length check and pop are atomic; ensure
any other direct accesses to _recent_commands elsewhere in the class also use
this lock.
In `@README.md`:
- Around line 217-224: The README's inventory of files in ~/.cortex is missing
the new ask-do session DB; add an entry for ~/.cortex/do_runs.db alongside
history.db in the listed tree so the configuration section reflects the
do-runner persistence. Update the list that currently shows history.db to
include a line like "├── do_runs.db # Ask/do-runner sessions
(SQLite)" referencing the filename do_runs.db so readers can find the persisted
runs.
In `@scripts/setup_ask_do.py`:
- Around line 289-347: The hook currently overwrites PROMPT_COMMAND; update both
hook_content and bashrc_addition to compose with existing PROMPT_COMMAND instead
of clobbering it: detect if PROMPT_COMMAND already contains "__cortex_log_cmd"
and if not set PROMPT_COMMAND="history -a;
__cortex_log_cmd${PROMPT_COMMAND:+;}$PROMPT_COMMAND" (or similar safe
composition) rather than export PROMPT_COMMAND='history -a; __cortex_log_cmd',
and ensure the same logic is used when writing the standalone hook and when
appending to .bashrc; keep the function name __cortex_log_cmd as the idempotent
check to avoid duplicate additions.
♻️ Duplicate comments (2)
cortex/do_runner/Untitled (1)
1-1: Remove accidental placeholder file.This single-character file looks like an accidental commit and can confuse tooling. Please delete it rather than keeping an empty placeholder.
🧹 Proposed fix
-icortex/system_info_generator.py (1)
519-533: Ensurehttpxis declared as a dependency for Ollama support.Line 519 imports
httpxat runtime; if it isn’t declared in the project dependencies,provider="ollama"will fail with ImportError. Please confirm it’s included in the dependency manifest(s).#!/bin/bash # Check whether httpx is declared in dependency manifests. rg -n "httpx" -g 'pyproject.toml' -g 'requirements*.txt' -g 'setup.cfg' -g 'setup.py'
🧹 Nitpick comments (11)
cortex/watch_service.py (2)
68-73: Missing type hints for thelogmethod.Per coding guidelines, type hints are required for Python code.
Proposed fix
- def log(self, message: str): + def log(self, message: str) -> None: """Log a message to the service log."""
184-317: Consider adding timeout to subprocess call in_monitor_bash_history.The
subprocess.runcall at line 152-154 has a 2-second timeout, which is good. However, the inotify file descriptor is not closed in theexceptblock if an exception occurs during the main loop.Proposed fix for inotify cleanup
while self.running: # Wait for inotify event with timeout r, _, _ = select.select([fd], [], [], 1.0) if not r: continue data = os.read(fd, 4096) # ... processing - os.close(fd) - return + try: + os.close(fd) + except OSError: + pass + return except Exception as e: self.log(f"Inotify not available, using polling: {e}") + try: + os.close(fd) + except (OSError, NameError): + passscripts/setup_ask_do.sh (4)
20-20: Consider addingset -ufor stricter variable handling.Using
set -eu(orset -euo pipefail) would catch undefined variable usage, which is a shell scripting best practice.Proposed fix
-set -e +set -euo pipefail
121-127:sed -iwith pattern range may leave orphaned content if markers are malformed.The sed command
sed -i '/# Cortex Terminal Watch Hook/,/^$/d'deletes from the marker to the next empty line. If there's no empty line after the block, it could delete more than intended. Consider using a more precise end marker.
172-178: Sourcingvenv/bin/activatemay fail in subshell context.The
source venv/bin/activateinside the script activates the venv for the script's process, but this won't persist after the script ends. The warning message is helpful, but consider clarifying this to users.
334-356: Duplicate code between watch_hook.sh and .bashrc injection.The same hook code is written to
~/.cortex/watch_hook.sh(lines 303-325) and also appended directly to.bashrc(lines 337-355). This creates maintenance burden. Consider sourcing the hook file from.bashrcinstead.Proposed simplification
# Add to .bashrc MARKER="# Cortex Terminal Watch Hook" if [ -f ~/.bashrc ]; then if ! grep -q "$MARKER" ~/.bashrc; then print_step "Adding hook to .bashrc..." cat >> ~/.bashrc << 'EOF' # Cortex Terminal Watch Hook -__cortex_last_histnum="" -__cortex_log_cmd() { - # ... duplicate code ... -} -export PROMPT_COMMAND='history -a; __cortex_log_cmd' +if [ -f ~/.cortex/watch_hook.sh ]; then + source ~/.cortex/watch_hook.sh +fi alias cw="source ~/.cortex/watch_hook.sh" EOFcortex/do_runner/models.py (1)
267-269: Command truncation may cut mid-word or mid-escape sequence.
node.command[:50]truncates without checking if it cuts mid-word or mid-escape. While minor, it could display oddly.cortex/do_runner/database.py (3)
125-127: SQL injection risk in schema migration via string interpolation.The column name and type are interpolated directly into the SQL string. While currently controlled by internal constants, this pattern is fragile.
Proposed fix using parameterized approach
for col_name, col_type in new_columns: if col_name not in existing_columns: try: - conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}") + # Column names can't be parameterized, but validate against allowlist + if col_name.isidentifier() and col_type.replace(" ", "").replace("_", "").isalnum(): + conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}") except sqlite3.OperationalError: passSince the values come from hardcoded constants in this file, this is low risk but worth noting for future maintenance.
404-416: Weak session ID generation using MD5.The session ID uses MD5 for hashing, which is cryptographically weak. While this is for session identification (not security), using
hashlib.sha256orsecrets.token_hexwould be more appropriate.Proposed fix
+import secrets + def create_session(self) -> str: """Create a new session and return the session ID.""" - session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{hashlib.md5(str(datetime.datetime.now().timestamp()).encode()).hexdigest()[:8]}" + session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{secrets.token_hex(4)}"
276-298: Missing error handling for malformedfull_dataJSON.If
full_datais corrupted or missing required keys, this will raise an exception. Consider defensive handling.Proposed fix
def get_run(self, run_id: str) -> DoRun | None: """Get a specific run by ID.""" with sqlite3.connect(str(self.db_path)) as conn: conn.row_factory = sqlite3.Row cursor = conn.execute("SELECT * FROM do_runs WHERE run_id = ?", (run_id,)) row = cursor.fetchone() if row: - full_data = json.loads(row["full_data"]) + try: + full_data = json.loads(row["full_data"]) + except (json.JSONDecodeError, TypeError): + return None run = DoRun(cortex/do_runner/managers.py (1)
104-116: Prefix matching could allow bypass via path manipulation.The
is_protectedcheck at line 113 usespath.startswith(protected + "/")which could be bypassed with symlinks or path traversal. Consider resolving symlinks.Proposed enhancement
def is_protected(self, path: str) -> bool: """Check if a path requires authentication for access.""" - path = os.path.abspath(path) + try: + path = os.path.realpath(path) # Resolve symlinks + except OSError: + path = os.path.abspath(path) all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@cortex/do_runner.py`:
- Around line 10-32: cortex/do_runner.py is a redundant wrapper that attempts to
re-export names (AutoFixer, DoHandler, DoRun, TaskTree, TaskTreeExecutor,
VerificationRunner, get_do_handler, setup_cortex_user, etc.) but Python will
load the cortex.do_runner package instead; delete the cortex/do_runner.py file
entirely and ensure the package's __init__.py (cortex.do_runner package) already
defines and re-exports the same symbols so existing imports continue to work
without change.
In `@cortex/system_info_generator.py`:
- Around line 388-400: The prompt string assigned to prompt includes "except for
fallback patterns" and encourages use of "||" fallbacks which contradicts the
CommandValidator and the file's own note; update the prompt (the prompt variable
in system_info_generator.py) to either remove the allowance for "||" chaining
entirely or explicitly state the exact safe fallback patterns permitted by the
validator, e.g., forbid all command chaining and "||" in the bullet list, or
list approved fallbacks and reference the validator rules so generated commands
won't be rejected; edit the prompt variable text accordingly to keep it
consistent with CommandValidator and the surrounding code.
♻️ Duplicate comments (15)
cortex/demo.py (1)
27-34: Add explicit-> Nonereturn types for helper functions.
_wait_for_enterand_sectionare missing return annotations. The project requires type hints for all Python code, so both should declare-> None. As per coding guidelines, please add explicit return type annotations.✏️ Suggested fix
-def _wait_for_enter(): +def _wait_for_enter() -> None: """Wait for user to press enter.""" console.print("\n[dim]Press Enter to continue...[/dim]") input() -def _section(title: str, problem: str): +def _section(title: str, problem: str) -> None: """Display a compact section header.""" console.print(f"\n[bold cyan]{'─' * 50}[/bold cyan]") console.print(f"[bold white]{title}[/bold white]") console.print(f"[dim]{problem}[/dim]\n")cortex/system_info_generator.py (1)
519-534: Declarehttpxas a runtime dependency for Ollama.The Ollama branch imports
httpx, which will raiseImportErrorif it isn't listed in the runtime dependencies. Please add it to the dependency manifest (or provide a clear install-time error).#!/bin/bash # Verify whether httpx is declared in dependency manifests rg -n "httpx" -g "pyproject.toml" -g "requirements*.txt" -g "setup.cfg" -g "setup.py"cortex/watch_service.py (2)
397-405: Guard_recent_commandsupdates with a lock.
_is_duplicatemutates_recent_commandsfrom multiple threads; list operations are not atomic. Wrap the membership check/append/pop in a lock to prevent races.✏️ Suggested fix
class CortexWatchDaemon: def __init__(self): # ... self._recent_commands: list[str] = [] # Last 100 commands for dedup + self._recent_commands_lock = threading.Lock() def _is_duplicate(self, command: str) -> bool: """Check if command was recently logged to avoid duplicates.""" - if command in self._recent_commands: - return True - - # Keep last 100 commands - self._recent_commands.append(command) - if len(self._recent_commands) > 100: - self._recent_commands.pop(0) - - return False + with self._recent_commands_lock: + if command in self._recent_commands: + return True + self._recent_commands.append(command) + if len(self._recent_commands) > 100: + self._recent_commands.pop(0) + return False
620-624: Avoid empty username inloginctl enable-linger.If
USERis unset, an empty string is passed tologinctl, which can fail silently. Resolve a real username and only call when non-empty.✏️ Suggested fix
+import getpass @@ - subprocess.run(["loginctl", "enable-linger", os.getenv("USER", "")], capture_output=True) + user = os.getenv("USER") or getpass.getuser() + if user: + subprocess.run(["loginctl", "enable-linger", user], capture_output=True)cortex/do_runner/diagnosis_v2.py (4)
1318-1325: Sanitizecmd_nameinapt-file searchor avoidshell=True.
cmd_namecomes from the original command and is interpolated into a shell string, creating injection risk. Validate allowed characters and run withshell=False, or useshutil.whichbefore falling back to apt‑file.✏️ Suggested fix
- result = subprocess.run( - f"apt-file search --regexp 'bin/{cmd_name}$' 2>/dev/null | head -1 | cut -d: -f1", - shell=True, - capture_output=True, - text=True, - timeout=10, - ) + if not re.fullmatch(r"[A-Za-z0-9._+-]+", cmd_name): + return None + result = subprocess.run( + ["apt-file", "search", "--regexp", f"bin/{cmd_name}$"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0 and result.stdout.strip(): + return result.stdout.splitlines()[0].split(":", 1)[0].strip()
1453-1459: Avoid enabling plaintext git credential storage without opt‑in.
git config --global credential.helper storewrites credentials in plain text to~/.git-credentials. Use a safer helper (cache/manager) or prompt the user to opt in with a clear warning.✏️ Suggested fix
- subprocess.run("git config --global credential.helper store", shell=True) + console.print( + "[yellow]⚠ Git credential.helper 'store' saves passwords in plain text.[/yellow]" + ) + subprocess.run("git config --global credential.helper cache", shell=True)
1463-1471: Do not embed credentials in pip index URL.
pip config set global.index-url https://{username}:{password}@...persists secrets to config and shell history. Set the index URL without credentials and supply auth via env (PIP_INDEX_URL) or keyring/netrc.✏️ Suggested fix
- login_cmd = f"pip config set global.index-url https://{username}:{{password}}@pypi.org/simple/" + login_cmd = "pip config set global.index-url https://pypi.org/simple/" + # Provide credentials via env (PIP_INDEX_URL) or a secure helper instead
1549-1556: Avoidshell=Truewhen executing fix commands.Resolved variables can be user/LLM-derived; executing with
shell=Trueinvites injection. Prefershlex.split+shell=Falseand only fall back toshell=Truefor truly complex commands.✏️ Suggested fix
+import shlex @@ - result = subprocess.run( - command, - shell=True, - capture_output=True, - text=True, - timeout=120, - ) + try: + args = shlex.split(command) + result = subprocess.run( + args, + capture_output=True, + text=True, + timeout=120, + ) + except ValueError: + result = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + timeout=120, + )cortex/do_runner/executor.py (2)
151-205: Auto-repair can execute sudo‑required fixes without explicit approval.Line 151 only checks protected paths, so repairs like
apt install/systemctlcan run without confirmation whenconfirm_callbackis missing ornew_pathsis empty;_execute_commandwill still prepend sudo. Please gate on_needs_sudo()or explicitsudotoo, and skip auto‑fix when confirmation isn’t available. As per coding guidelines, avoid silent sudo execution.🔧 Suggested guard
- new_paths = self._identify_paths_needing_privileges(fix_commands) - if new_paths and confirm_callback: + new_paths = self._identify_paths_needing_privileges(fix_commands) + needs_privilege = new_paths or any( + self._needs_sudo(cmd) or cmd.strip().startswith("sudo") + for cmd in fix_commands + ) + if needs_privilege and confirm_callback:
66-112: Add test coverage and fix silent sudo execution in TaskTreeExecutor.This execution engine requires >80% test coverage per coding guidelines (currently no tests exist). Additionally, the file matches the
**/*exec*.pypattern requiring explicit user approval for elevated permissions. However,_execute_command()silently addssudoto qualifying commands without checkingconfirm_callbackfirst—only repair tasks with privilege-needing paths verify approval. Refactor to require confirm_callback approval before any sudo elevation, then add comprehensive unit tests for success, failure, auto-repair, permission gating, and sudo approval flows.cortex/do_runner/models.py (2)
90-96:get_depth()never advances—this can infinite loop.Line 94 assigns
node = node, so the loop never progresses whenparent_idis set. This can hang any caller. Consider passing aTaskTreeto resolve parents or store a direct parent reference.🧭 Example fix
- def get_depth(self) -> int: + def get_depth(self, tree: "TaskTree") -> int: """Get the depth of this node in the tree.""" depth = 0 - node = self - while node.parent_id: - depth += 1 - node = node + current_id = self.parent_id + while current_id: + depth += 1 + parent = tree.get_task(current_id) + current_id = parent.parent_id if parent else None return depth
349-356: Output preview always shows “...” even when not truncated.Line 356 appends ellipsis unconditionally. This makes short outputs look truncated.
✅ Fix
- if cmd.output: - lines.append(f" Output: {cmd.output[:500]}...") + if cmd.output: + preview = cmd.output[:500] + suffix = "..." if len(cmd.output) > 500 else "" + lines.append(f" Output: {preview}{suffix}")scripts/setup_ask_do.py (1)
289-347: Don’t clobber existingPROMPT_COMMANDin hooks.Lines 310 and 346 overwrite
PROMPT_COMMAND, which can break users’ prompt tooling. Please append instead of replace.💡 Safer composition
-export PROMPT_COMMAND='history -a; __cortex_log_cmd' +export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"cortex/do_runner/terminal.py (2)
714-739: Preserve existingPROMPT_COMMANDwhen installing hooks.Line 738 overwrites
PROMPT_COMMAND, which can break user prompt tooling. Compose with the existing value.💡 Safer composition
-export PROMPT_COMMAND='history -a; __cortex_log_cmd' +export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"
1489-1512: Never auto‑execute sudo, even with NOPASSWD.Lines 1492–1506 attempt a
sudo -n trueprobe and then proceed to run sudo commands without interactive approval. This violates the “no automatic sudo execution” guardrail. As per coding guidelines, always require explicit user approval for sudo.✅ Remove auto‑sudo probe
- if needs_sudo: - try: - check_sudo = subprocess.run( - ["sudo", "-n", "true"], capture_output=True, timeout=5 - ) - - if check_sudo.returncode != 0: - sudo_commands_pending.append(fix_cmd) - results.append((fix_cmd, "sudo", None)) - console.print( - f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" - ) - continue - except Exception: - sudo_commands_pending.append(fix_cmd) - results.append((fix_cmd, "sudo", None)) - console.print( - f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" - ) - continue + if needs_sudo: + sudo_commands_pending.append(fix_cmd) + results.append((fix_cmd, "sudo", None)) + console.print( + f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" + ) + continue
🧹 Nitpick comments (3)
cortex/watch_service.py (1)
467-545: Add return type annotations for daemon lifecycle APIs.
start,stop, andstatus(plusmain/log) lack explicit return types. Please add return annotations to satisfy the typing requirement. As per coding guidelines, add explicit return type annotations.✏️ Suggested fix
- def start(self): + def start(self) -> bool: @@ - def stop(self): + def stop(self) -> tuple[bool, str]: @@ - def status(self) -> dict: + def status(self) -> dict[str, Any]: @@ -def main(): +def main() -> None:cortex/do_runner/database.py (1)
26-41: Add return type annotations for internal DB helpers.
_ensure_directory,_init_db,_migrate_schema,update_session, andend_sessionlack return annotations. Please add-> Nonewhere appropriate. As per coding guidelines, add explicit return type annotations.✏️ Suggested fix
- def _ensure_directory(self): + def _ensure_directory(self) -> None: @@ - def _init_db(self): + def _init_db(self) -> None: @@ - def _migrate_schema(self, conn: sqlite3.Connection): + def _migrate_schema(self, conn: sqlite3.Connection) -> None: @@ - def update_session( - self, session_id: str, query: str | None = None, increment_runs: bool = False - ): + def update_session( + self, session_id: str, query: str | None = None, increment_runs: bool = False + ) -> None: @@ - def end_session(self, session_id: str): + def end_session(self, session_id: str) -> None:scripts/setup_ask_do.py (1)
43-67: Add return type hints + docstring for public APIs to meet guidelines.Public helpers (e.g.,
print_header,print_step,print_success,main) are missing return type annotations, andmain()lacks a docstring. As per coding guidelines, public APIs should be fully typed and documented.✍️ Example updates
-def print_header(text: str): +def print_header(text: str) -> None: """Print a section header.""" @@ -def print_step(text: str): +def print_step(text: str) -> None: @@ -def print_success(text: str): +def print_success(text: str) -> None: @@ -def print_warning(text: str): +def print_warning(text: str) -> None: @@ -def print_error(text: str): +def print_error(text: str) -> None: @@ -def main(): +def main() -> int: + """Run the ask --do setup wizard."""Also applies to: 544-605
|
There was a problem hiding this comment.
Actionable comments posted: 12
🤖 Fix all issues with AI agents
In `@cortex/ask.py`:
- Around line 386-394: The code silently swallows ImportError when initializing
self._do_handler in the constructor (do_mode path), leaving do_mode effectively
enabled but non-functional; update the block around self._do_handler and
DoHandler so that if importing cortex.do_runner.handler fails you either raise a
clear RuntimeError (e.g., "DoRunner requested but not installed: install package
X") or set do_mode=False and emit a visible warning via the existing logger;
ensure the change affects the same symbols (_do_handler, do_mode, DoHandler) so
callers see a clear error or the mode is explicitly disabled with a warning
instead of silently continuing.
In `@cortex/cli.py`:
- Around line 925-928: The current flow calls handler.ask(question) and prints
Markdown directly, bypassing the execution pipeline; change this to route the
question through the DoRunner/DoHandler entrypoint instead: detect if a
DoRunner/DoHandler instance is available and pass the question to its run/handle
method (instead of handler.ask) so approval/execution/auto-repair executes, and
if no DoRunner/DoHandler is present fail fast with an explicit error message;
update references around handler.ask, DoRunner, and DoHandler to ensure the CLI
uses the execution entrypoint for `--do` mode.
In `@cortex/do_runner/database.py`:
- Around line 283-289: The code calling json.loads(row["full_data"]) when
building a DoRun can raise on malformed JSON and crash retrieval; wrap the
json.loads and subsequent mapping to DoRun/CommandLog.from_dict in a try/except
catching json.JSONDecodeError (and ValueError as a fallback) and on failure
return None (or a partial DoRun) instead of propagating the exception so callers
handle missing runs gracefully; update the block that constructs DoRun (run_id,
summary, mode=RunMode(...), commands=[CommandLog.from_dict(...)]) to live inside
the try and ensure any decode/mapping error is logged and handled.
In `@cortex/do_runner/managers.py`:
- Around line 104-115: The is_protected function currently uses os.path.abspath
which doesn’t resolve symlinks and can be bypassed; update is_protected to
resolve real paths for the input and for each entry in SYSTEM_PROTECTED_PATHS
and USER_PROTECTED_PATHS (e.g., via os.path.realpath or pathlib.Path.resolve),
then compare the resolved path against the set and use startswith checks on
resolved protected entries (ensuring you append a separator when checking
prefixes) to prevent symlink bypasses.
In `@cortex/do_runner/terminal.py`:
- Around line 231-239: The LLM callback _llm_for_autofix returns {"response":
...} which AutoFixer._get_llm_fix ignores; change _llm_for_autofix to return the
keys AutoFixer expects (e.g., return {"answer": result} or parse and return
{"fix_commands": [...]} / {"do_commands": [...]}), so AutoFixer._get_llm_fix can
consume the LLM output; update the return shape in _llm_for_autofix (and ensure
max_tokens/timeout behavior remains) to supply "answer" or parsed
"fix_commands"/"do_commands" instead of "response".
- Around line 1666-1704: The _is_safe_fix_command() currently whitelists "sudo"
patterns which allows privileged commands to be auto-executed; remove any
special-casing of sudo so sudo-prefixed commands are not considered auto-safe.
Specifically, delete "sudo ..." entries from safe_patterns and remove the entire
sudo-specific block that checks cmd_lower.startswith("sudo ") (including the
safe_sudo list and rest handling), ensuring the function continues to allow the
non-sudo equivalents (e.g., "systemctl", "apt", "nginx", "chmod", "chown",
"mkdir") but never treats sudo-prefixed commands (cmd_lower, rest) as
automatically safe.
In `@cortex/do_runner/verification.py`:
- Around line 1246-1258: apply_file_recommendations() currently runs sudo
mutations directly in the "backup_and_replace" and "create_parent" branches
(constructing backup_cmd and mkdir_cmd and calling _execute_command), which
bypasses the no-automatic-sudo guardrail; change this so the function does not
execute sudo commands automatically: either return the prepared commands
(backup_cmd, mkdir_cmd) for the caller to run, or add an explicit allow_sudo
boolean parameter that must be true to call _execute_command with
needs_sudo=True, and otherwise append the commands to commands_executed (or a
separate pending_sudo list) without executing them; ensure references to
backup_and_replace, create_parent, backup_cmd, mkdir_cmd and _execute_command
are updated accordingly.
In `@cortex/system_info_generator.py`:
- Around line 810-827: The API key check in get_system_info_generator currently
always raises if neither ANTHROPIC_API_KEY nor OPENAI_API_KEY is set, preventing
use of the local "ollama" provider; change the logic so api_key is only required
for non-ollama providers (e.g., check provider.lower() != "ollama" before
raising). Update the existing api_key assignment and the subsequent raise
ValueError to conditionally raise only when provider is not "ollama" (refer to
the api_key variable and the ValueError line inside get_system_info_generator).
In `@cortex/watch_service.py`:
- Around line 381-394: Concurrent writes in _log_to_json can interleave and
corrupt the JSON lines; fix by serializing file access—either add a dedicated
threading.Lock (e.g., self._json_write_lock initialized in the class __init__)
and wrap the open/write block in a with self._json_write_lock: section, or
implement a single writer thread with a Queue (e.g., enqueue entry objects in
_log_to_json and have a background _json_writer thread consume the queue and
perform open/append writes). Update _log_to_json to enqueue or to acquire the
lock before opening/writing and ensure any new background thread is
started/stopped in the service lifecycle methods.
- Around line 45-46: The _watch_hook_commands set is mutated from multiple
threads and can grow unbounded; replace it with a bounded structure (e.g.,
collections.deque(maxlen=...) or a TTL map) and protect all accesses with a lock
— preferably reuse the existing recent-command dedupe lock used elsewhere (the
same lock guarding recent-command dedupe logic referenced in the review) or add
a dedicated threading.Lock (e.g., self._watch_hook_lock) if that lock is not
accessible; update all code paths that read/write _watch_hook_commands
(including the code around lines 369-422) to acquire the lock before mutating or
checking membership to ensure thread-safety and bounded memory usage.
- Around line 58-572: Add explicit type annotations to the daemon methods and
helpers mentioned: annotate _handle_signal(self, signum: int, frame:
Optional[types.FrameType]) -> None and _handle_reload(self, signum: int, frame:
Optional[types.FrameType]) -> None; annotate log(self, message: str) -> None,
_load_state(self) -> None, _save_state(self) -> None,
_monitor_bash_history(self) -> None, _monitor_watch_hook(self) -> None,
_log_to_json(self, command: str, source: str, terminal_id: str) -> None (ensure
existing signature matches), _log_command(self, command: str, source: str =
"unknown", terminal_id: Optional[str] = None) -> None,
_cleanup_stale_terminals(self) -> None, _shutdown(self) -> None, start(self) ->
bool, and stop(self) -> tuple[bool, str]; import any needed typing names
(Optional, types.FrameType) at top and update signatures for these functions
accordingly to satisfy typing requirements.
- Around line 30-52: The daemon has thread-safety, typing, and an edge-case
issue: protect concurrent access to _watch_hook_commands and _recent_commands by
adding a threading.Lock() (e.g., self._cmd_lock) and using it around all
reads/writes in methods like _monitor_watch_hook, _monitor_bash_history, and any
other thread functions; add explicit return type annotations per PEP 484/8 for
the listed methods (__init__, _handle_signal, _handle_reload, _load_state,
_save_state, _monitor_bash_history, _monitor_watch_hook,
_cleanup_stale_terminals, start, stop) (use None for void methods and proper
types for loaders), and handle the loginctl edge case by checking
os.getenv("USER") is non-empty before calling loginctl enable-linger (or
catch/handle subprocess errors where loginctl is invoked) in the code path
around the loginctl call so failures are graceful. Ensure updates reference the
class CortexWatchDaemon and its attributes (_watch_hook_commands,
_recent_commands, _cmd_lock, terminals_dir) and that unit tests are added to
cover threading, type annotations, and the USER/loginctl edge case to keep
coverage >80%.
♻️ Duplicate comments (16)
cortex/watch_service.py (2)
397-405: Protect_recent_commandswith a lock.
Concurrent membership check + append + pop across threads is non-atomic and can race.
623-624: Avoid passing an empty username tologinctl enable-linger.
IfUSERis unset, Line 624 passes an empty string. Resolve a real username (getpass/pwd) and guard the call.🔧 Suggested fix
- subprocess.run(["loginctl", "enable-linger", os.getenv("USER", "")], capture_output=True) + import getpass + + user = os.getenv("USER") or getpass.getuser() + if user: + subprocess.run(["loginctl", "enable-linger", user], capture_output=True)cortex/demo.py (1)
27-34: Add-> Nonereturn types for helper functions.As per coding guidelines, these helpers should have explicit return type annotations.
✏️ Proposed fix
-def _wait_for_enter(): +def _wait_for_enter() -> None: """Wait for user to press enter.""" console.print("\n[dim]Press Enter to continue...[/dim]") input() -def _section(title: str, problem: str): +def _section(title: str, problem: str) -> None: """Display a compact section header.""" console.print(f"\n[bold cyan]{'─' * 50}[/bold cyan]") console.print(f"[bold white]{title}[/bold white]") console.print(f"[dim]{problem}[/dim]\n")cortex/do_runner/managers.py (1)
220-237: chmod fallback grants world-level access.The fallback uses
o+rw, which makes the file world-readable/writable and exceeds the intended privilege scope.cortex/do_runner/models.py (2)
90-97: Infinite loop inget_depth().
node = nodenever advances, so this loops forever whenparent_idis set. It needs a parent reference or a tree lookup.
349-357: Output always marked as truncated.
cmd.output[:500]}...appends...even when the output is shorter than 500 chars.cortex/do_runner/diagnosis_v2.py (4)
1318-1326: Avoidshell=Truewith user-influencedcmd_name.
apt-file searchis built via a shell string usingcmd_name, which can be manipulated.
1458-1459:credential.helper storesaves passwords in plaintext.This writes credentials to
~/.git-credentialsin plain text; please warn and/or require explicit opt-in or use a safer helper.
1463-1471: Credentials embedded in pip index URL.This stores secrets in config and may expose them in logs/history.
1549-1556: Shell execution with substituted variables risks injection.
execute_fix_commandsrunsshell=Trueafter variable substitution; useshlex.splitandshell=False, or validate/escape inputs.cortex/do_runner/executor.py (1)
151-208: Privileged auto-repairs can run without explicit approval.The gate only checks protected paths; sudo-required repairs (e.g.,
apt,systemctl) can execute automatically, and whenconfirm_callbackisNoneeven protected-path repairs run. This violates the explicit-approval requirement for privileged commands. As per coding guidelines, require confirmation for any sudo-needed repair, and skip/defer if confirmation isn't available.🔧 Suggested guard
- new_paths = self._identify_paths_needing_privileges(fix_commands) - if new_paths and confirm_callback: + new_paths = self._identify_paths_needing_privileges(fix_commands) + needs_privilege = new_paths or any( + self._needs_sudo(cmd) or cmd.strip().startswith("sudo") + for cmd in fix_commands + ) + if needs_privilege: + if not confirm_callback: + console.print("[yellow]⚠ Privileged repairs require approval; skipping auto-repair.[/yellow]") + task.status = CommandStatus.FAILED + task.reasoning = "Privileged repair requires explicit approval" + return False, was_repaired + + if needs_privilege and confirm_callback:cortex/system_info_generator.py (2)
388-400: Align prompt with CommandValidator (no chaining).The prompt still allows
||fallbacks while validation blocks chaining, so generated commands will be rejected. Keep the prompt consistent with validator behavior.🔧 Suggested prompt adjustment
-- NEVER use dangerous command chaining (;, &&, ||) except for fallback patterns -- Commands should handle missing files/tools gracefully using || echo fallbacks +- NEVER use command chaining (;, &&, ||) +- Commands should handle missing files/tools gracefully without chaining
518-534: Declarehttpxas a dependency for the Ollama path.
httpxis imported for the Ollama client, but it isn’t declared in dependency manifests. This will raiseImportErrorwhenprovider="ollama"is used.cortex/do_runner/verification.py (1)
19-33: Avoid double‑sudo in_execute_command.When
needs_sudo=True, the code prependssudoand then wraps with["sudo","bash","-c", ...], yieldingsudo sudo ...and breaking execution.✅ Safer execution pattern
- if needs_sudo and not cmd.strip().startswith("sudo"): - cmd = f"sudo {cmd}" - - result = subprocess.run( - ["sudo", "bash", "-c", cmd] if needs_sudo else cmd, - shell=not needs_sudo, + exec_cmd = ["sudo", "bash", "-c", cmd] if needs_sudo else cmd + result = subprocess.run( + exec_cmd, + shell=not needs_sudo, capture_output=True, text=True, timeout=timeout, )Also applies to: 817-831, 997-1011
cortex/do_runner/terminal.py (2)
714-739: Preserve existingPROMPT_COMMANDwhen installing the hook.Current setup overwrites any existing
PROMPT_COMMAND, breaking user prompt tooling and other hooks.💡 Safer PROMPT_COMMAND composition
-export PROMPT_COMMAND='history -a; __cortex_log_cmd' +export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"
1489-1512: Never auto‑execute sudo, even with NOPASSWD.The NOPASSWD probe auto‑runs sudo commands, which violates the “no automatic sudo execution” guardrail. Sudo fixes should always be queued for manual execution.
✅ Safer behavior
- if needs_sudo: - try: - check_sudo = subprocess.run( - ["sudo", "-n", "true"], capture_output=True, timeout=5 - ) - - if check_sudo.returncode != 0: - sudo_commands_pending.append(fix_cmd) - results.append((fix_cmd, "sudo", None)) - console.print( - f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" - ) - continue - except Exception: - sudo_commands_pending.append(fix_cmd) - results.append((fix_cmd, "sudo", None)) - console.print( - f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" - ) - continue + if needs_sudo: + sudo_commands_pending.append(fix_cmd) + results.append((fix_cmd, "sudo", None)) + console.print( + f" [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]" + ) + continue
🧹 Nitpick comments (6)
cortex/cli.py (1)
855-861: Allowquestionto be optional in do-mode.
Line 855 typesquestionasstr, but callers passNonewhen--dois used without an initial question (Line 5581-5585). Update the annotation/docstring tostr | Noneto keep typing accurate.As per coding guidelines, type hints should reflect actual usage.♻️ Suggested tweak
- def ask(self, question: str, do_mode: bool = False) -> int: + def ask(self, question: str | None, do_mode: bool = False) -> int:cortex/do_runner/managers.py (1)
58-88: Add explicit return type annotations for helpers.As per coding guidelines, type hints are required for Python code. Please add
-> Noneon helper methods like__init__,_ensure_config_dir,_load_user_paths, and_save_user_paths.✏️ Example fix
- def __init__(self): + def __init__(self) -> None: self.config_file = Path.home() / ".cortex" / "protected_paths.json" self._ensure_config_dir() self._load_user_paths() - def _ensure_config_dir(self): + def _ensure_config_dir(self) -> None: """Ensure the config directory exists.""" try: self.config_file.parent.mkdir(parents=True, exist_ok=True) except OSError: self.config_file = Path("/tmp") / ".cortex" / "protected_paths.json" self.config_file.parent.mkdir(parents=True, exist_ok=True) - def _load_user_paths(self): + def _load_user_paths(self) -> None: """Load user-configured protected paths.""" if self.config_file.exists(): try: with open(self.config_file) as f: data = json.load(f) self.USER_PROTECTED_PATHS = set(data.get("paths", [])) except (json.JSONDecodeError, OSError): pass - def _save_user_paths(self): + def _save_user_paths(self) -> None: """Save user-configured protected paths.""" try: self.config_file.parent.mkdir(parents=True, exist_ok=True) with open(self.config_file, "w") as f: json.dump({"paths": list(self.USER_PROTECTED_PATHS)}, f, indent=2) except OSError as e: console.print(f"[yellow]Warning: Could not save protected paths: {e}[/yellow]")cortex/do_runner/models.py (2)
85-88: Add return type foradd_child.As per coding guidelines, add an explicit return type (e.g.,
-> None).✏️ Minimal fix
- def add_child(self, child: "TaskNode"): + def add_child(self, child: "TaskNode") -> None: """Add a child task.""" child.parent_id = self.id self.children.append(child)
238-278: Add return types forprint_tree/_print_node.As per coding guidelines, annotate these helpers with
-> Nonefor clarity and consistency.✏️ Example
- def print_tree(self, indent: str = ""): + def print_tree(self, indent: str = "") -> None: """Print the task tree structure.""" for i, root in enumerate(self.root_tasks): is_last = i == len(self.root_tasks) - 1 self._print_node(root, indent, is_last) - def _print_node(self, node: TaskNode, indent: str, is_last: bool): + def _print_node(self, node: TaskNode, indent: str, is_last: bool) -> None: """Print a single node with its children."""cortex/do_runner/database.py (1)
205-206: Add return type annotations for public DB APIs.As per coding guidelines, public methods should include explicit return types (e.g.,
save_run(...) -> str), and similarly for other public methods in this class.✏️ Example fix
- def save_run(self, run: DoRun) -> str: + def save_run(self, run: DoRun) -> str: """Save a do run to the database with detailed command information."""(Please apply the same pattern to the other public methods.)
cortex/system_info_generator.py (1)
574-576: Fixhistorytyping to match actual values.
historyis annotated asdict[str, str], but you store booleans and possibly non‑string values. Usedict[str, Any]or aTypedDictto keep type hints accurate. As per coding guidelines, keep type annotations precise for new code.♻️ Suggested type fix
- history: list[dict[str, str]] = [] + history: list[dict[str, Any]] = []
| # Initialize do_handler if in do_mode | ||
| self._do_handler = None | ||
| if do_mode: | ||
| try: | ||
| from cortex.do_runner.handler import DoHandler | ||
|
|
||
| self._do_handler = DoHandler() | ||
| except ImportError: | ||
| pass |
There was a problem hiding this comment.
Don't silently ignore missing DoRunner when --do is requested.
Line 389-394 swallows ImportError and leaves _do_handler as None, so execution mode appears enabled but cannot run. Please raise a clear error or explicitly disable do_mode with a warning so users aren’t misled.
🔧 Suggested fix
if do_mode:
try:
from cortex.do_runner.handler import DoHandler
self._do_handler = DoHandler()
- except ImportError:
- pass
+ except ImportError as e:
+ raise ImportError(
+ "Execution mode (--do) requires the do_runner extras. "
+ "Install them or run without --do."
+ ) from e🤖 Prompt for AI Agents
In `@cortex/ask.py` around lines 386 - 394, The code silently swallows ImportError
when initializing self._do_handler in the constructor (do_mode path), leaving
do_mode effectively enabled but non-functional; update the block around
self._do_handler and DoHandler so that if importing cortex.do_runner.handler
fails you either raise a clear RuntimeError (e.g., "DoRunner requested but not
installed: install package X") or set do_mode=False and emit a visible warning
via the existing logger; ensure the change affects the same symbols
(_do_handler, do_mode, DoHandler) so callers see a clear error or the mode is
explicitly disabled with a warning instead of silently continuing.
| # Process the question | ||
| result = handler.ask(question) | ||
| if result: | ||
| console.print(Markdown(result)) |
There was a problem hiding this comment.
--do session never invokes the execution pipeline.
Lines 925-928 just call handler.ask() and print Markdown, so no command approval/execution or auto-repair occurs even though the UI says it will. Route through the DoRunner/DoHandler entrypoint (or fail fast when unavailable) so execution mode actually runs.
🤖 Prompt for AI Agents
In `@cortex/cli.py` around lines 925 - 928, The current flow calls
handler.ask(question) and prints Markdown directly, bypassing the execution
pipeline; change this to route the question through the DoRunner/DoHandler
entrypoint instead: detect if a DoRunner/DoHandler instance is available and
pass the question to its run/handle method (instead of handler.ask) so
approval/execution/auto-repair executes, and if no DoRunner/DoHandler is present
fail fast with an explicit error message; update references around handler.ask,
DoRunner, and DoHandler to ensure the CLI uses the execution entrypoint for
`--do` mode.
| if row: | ||
| full_data = json.loads(row["full_data"]) | ||
| run = DoRun( | ||
| run_id=full_data["run_id"], | ||
| summary=full_data["summary"], | ||
| mode=RunMode(full_data["mode"]), | ||
| commands=[CommandLog.from_dict(c) for c in full_data["commands"]], |
There was a problem hiding this comment.
Guard against corrupted full_data JSON.
json.loads(row["full_data"]) can raise and break retrieval if the DB has malformed data. Consider a try/except and return None (or a partial run) on decode failure.
🛡️ Example hardening
- full_data = json.loads(row["full_data"])
+ try:
+ full_data = json.loads(row["full_data"])
+ except (TypeError, json.JSONDecodeError):
+ return None🤖 Prompt for AI Agents
In `@cortex/do_runner/database.py` around lines 283 - 289, The code calling
json.loads(row["full_data"]) when building a DoRun can raise on malformed JSON
and crash retrieval; wrap the json.loads and subsequent mapping to
DoRun/CommandLog.from_dict in a try/except catching json.JSONDecodeError (and
ValueError as a fallback) and on failure return None (or a partial DoRun)
instead of propagating the exception so callers handle missing runs gracefully;
update the block that constructs DoRun (run_id, summary, mode=RunMode(...),
commands=[CommandLog.from_dict(...)]) to live inside the try and ensure any
decode/mapping error is logged and handled.
| def is_protected(self, path: str) -> bool: | ||
| """Check if a path requires authentication for access.""" | ||
| path = os.path.abspath(path) | ||
| all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS | ||
|
|
||
| if path in all_protected: | ||
| return True | ||
|
|
||
| for protected in all_protected: | ||
| if path.startswith(protected + "/") or path == protected: | ||
| return True | ||
|
|
There was a problem hiding this comment.
Protected-path checks can be bypassed via symlinks.
os.path.abspath does not resolve symlinks, so a path like /tmp/link → /etc/shadow would not be flagged as protected. Use realpath/resolve for both the input path and protected entries.
🔒 Proposed hardening
- path = os.path.abspath(path)
+ path = os.path.realpath(path)
all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS
if path in all_protected:
return True
for protected in all_protected:
- if path.startswith(protected + "/") or path == protected:
+ protected_real = os.path.realpath(protected)
+ if path == protected_real or path.startswith(protected_real + os.sep):
return True🤖 Prompt for AI Agents
In `@cortex/do_runner/managers.py` around lines 104 - 115, The is_protected
function currently uses os.path.abspath which doesn’t resolve symlinks and can
be bypassed; update is_protected to resolve real paths for the input and for
each entry in SYSTEM_PROTECTED_PATHS and USER_PROTECTED_PATHS (e.g., via
os.path.realpath or pathlib.Path.resolve), then compare the resolved path
against the set and use startswith checks on resolved protected entries
(ensuring you append a separator when checking prefixes) to prevent symlink
bypasses.
| def _llm_for_autofix(self, prompt: str) -> dict: | ||
| """LLM callback for the AutoFixer.""" | ||
| if not self._llm or not self._llm.is_available(): | ||
| return {} | ||
|
|
||
| result = self._llm.analyze(prompt, max_tokens=200, timeout=15) | ||
| if result: | ||
| return {"response": result, "fix_commands": []} | ||
| return {} |
There was a problem hiding this comment.
AutoFixer LLM callback output is ignored.
AutoFixer._get_llm_fix() looks for fix_commands, do_commands, or answer, but _llm_for_autofix() returns {"response": ...}, so LLM suggestions never surface. Return answer (or parsed fix_commands) so AutoFixer can consume it.
🐛 Suggested fix
- if result:
- return {"response": result, "fix_commands": []}
+ if result:
+ return {"answer": result}🤖 Prompt for AI Agents
In `@cortex/do_runner/terminal.py` around lines 231 - 239, The LLM callback
_llm_for_autofix returns {"response": ...} which AutoFixer._get_llm_fix ignores;
change _llm_for_autofix to return the keys AutoFixer expects (e.g., return
{"answer": result} or parse and return {"fix_commands": [...]} / {"do_commands":
[...]}), so AutoFixer._get_llm_fix can consume the LLM output; update the return
shape in _llm_for_autofix (and ensure max_tokens/timeout behavior remains) to
supply "answer" or parsed "fix_commands"/"do_commands" instead of "response".
| def get_system_info_generator( | ||
| provider: str = "claude", | ||
| debug: bool = False, | ||
| ) -> SystemInfoGenerator: | ||
| """ | ||
| Factory function to create a SystemInfoGenerator with default configuration. | ||
| Args: | ||
| provider: LLM provider to use | ||
| debug: Enable debug output | ||
| Returns: | ||
| Configured SystemInfoGenerator instance | ||
| """ | ||
| api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY") | ||
| if not api_key: | ||
| raise ValueError("No API key found. Set ANTHROPIC_API_KEY or OPENAI_API_KEY") | ||
|
|
There was a problem hiding this comment.
Allow ollama provider without an API key.
get_system_info_generator() always requires an API key, which blocks local Ollama usage. Gate the API key requirement by provider.
🐛 Suggested fix
def get_system_info_generator(
provider: str = "claude",
debug: bool = False,
) -> SystemInfoGenerator:
@@
- api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")
- if not api_key:
- raise ValueError("No API key found. Set ANTHROPIC_API_KEY or OPENAI_API_KEY")
-
- return SystemInfoGenerator(api_key=api_key, provider=provider, debug=debug)
+ provider_lower = provider.lower()
+ api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")
+ if provider_lower in {"claude", "openai"} and not api_key:
+ raise ValueError("No API key found. Set ANTHROPIC_API_KEY or OPENAI_API_KEY")
+
+ return SystemInfoGenerator(api_key=api_key, provider=provider_lower, debug=debug)🤖 Prompt for AI Agents
In `@cortex/system_info_generator.py` around lines 810 - 827, The API key check in
get_system_info_generator currently always raises if neither ANTHROPIC_API_KEY
nor OPENAI_API_KEY is set, preventing use of the local "ollama" provider; change
the logic so api_key is only required for non-ollama providers (e.g., check
provider.lower() != "ollama" before raising). Update the existing api_key
assignment and the subsequent raise ValueError to conditionally raise only when
provider is not "ollama" (refer to the api_key variable and the ValueError line
inside get_system_info_generator).
| class CortexWatchDaemon: | ||
| """Background daemon that monitors terminal activity.""" | ||
|
|
||
| def __init__(self): | ||
| self.running = False | ||
| self.cortex_dir = Path.home() / ".cortex" | ||
| self.watch_log = self.cortex_dir / "terminal_watch.log" | ||
| self.terminals_dir = self.cortex_dir / "terminals" | ||
| self.pid_file = self.cortex_dir / "watch_service.pid" | ||
| self.state_file = self.cortex_dir / "watch_state.json" | ||
|
|
||
| # Terminal tracking | ||
| self.terminals: dict[str, dict[str, Any]] = {} | ||
| self.terminal_counter = 0 | ||
|
|
||
| # Track commands seen from watch_hook to avoid duplicates with bash_history | ||
| self._watch_hook_commands: set[str] = set() | ||
| self._recent_commands: list[str] = [] # Last 100 commands for dedup | ||
|
|
||
| # Ensure directories exist | ||
| self.cortex_dir.mkdir(parents=True, exist_ok=True) | ||
| self.terminals_dir.mkdir(parents=True, exist_ok=True) | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for existing tests covering the watch service/installer flows
rg -nP --type=py -g '*test*' 'CortexWatchDaemon|watch_service|install_service|uninstall_service' 2>/dev/null || echo "No tests found with pattern search"Repository: cortexlinux/cortex
Length of output: 96
🏁 Script executed:
#!/bin/bash
# Check if the watch_service.py file exists and get its size to determine if we need to read it in sections
find . -name "watch_service.py" -type f | head -20Repository: cortexlinux/cortex
Length of output: 87
🏁 Script executed:
#!/bin/bash
# List test files in the repository to understand test structure
fd --type f -e py '*test*.py' | head -30Repository: cortexlinux/cortex
Length of output: 350
🏁 Script executed:
#!/bin/bash
# Check for pytest coverage reports or test configuration
find . -name "*.coverage*" -o -name "pytest.ini" -o -name "setup.cfg" -o -name "pyproject.toml" 2>/dev/null | head -10Repository: cortexlinux/cortex
Length of output: 78
🏁 Script executed:
#!/bin/bash
# Find test files with correct glob pattern
fd --type f --glob '*test*.py' | head -30Repository: cortexlinux/cortex
Length of output: 922
🏁 Script executed:
#!/bin/bash
# Get size of watch_service.py to know how to read it
wc -l ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 91
🏁 Script executed:
#!/bin/bash
# Check pyproject.toml for coverage configuration
cat ./pyproject.tomlRepository: cortexlinux/cortex
Length of output: 6029
🏁 Script executed:
#!/bin/bash
# Search for any watch_service mentions in existing test files
rg -l 'watch_service\|CortexWatchDaemon\|install_service\|uninstall_service' tests/ 2>/dev/null || echo "No matches in tests/"Repository: cortexlinux/cortex
Length of output: 82
🏁 Script executed:
#!/bin/bash
# Read the first 150 lines of watch_service.py to assess implementation
head -150 ./cortex/watch_service.py | cat -nRepository: cortexlinux/cortex
Length of output: 6185
🏁 Script executed:
#!/bin/bash
# Check if there's a watch_service test file
find . -name '*watch*test*.py' -o -name '*test*watch*.py' 2>/dev/nullRepository: cortexlinux/cortex
Length of output: 44
🏁 Script executed:
#!/bin/bash
# Read next section of watch_service.py to see more implementation
sed -n '150,300p' ./cortex/watch_service.py | cat -nRepository: cortexlinux/cortex
Length of output: 6964
🏁 Script executed:
#!/bin/bash
# Search for install_service and uninstall_service methods
grep -n "def install_service\|def uninstall_service" ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 157
🏁 Script executed:
#!/bin/bash
# Check for threading locks and synchronization primitives
grep -n "Lock\|threading\|RLock\|Semaphore" ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 328
🏁 Script executed:
#!/bin/bash
# Read install_service and uninstall_service methods
sed -n '604,719p' ./cortex/watch_service.py | cat -nRepository: cortexlinux/cortex
Length of output: 4600
🏁 Script executed:
#!/bin/bash
# Search for loginctl command usage
grep -n "loginctl" ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 163
🏁 Script executed:
#!/antml/bin
# Check for all accesses to _recent_commands and _watch_hook_commands to assess race conditions
grep -n "_recent_commands\|_watch_hook_commands" ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 598
🏁 Script executed:
#!/bin/bash
# Check all method definitions to assess type hint coverage
grep -n "def " ./cortex/watch_service.py | head -30Repository: cortexlinux/cortex
Length of output: 1073
🏁 Script executed:
#!/antml:bin
# Check if there are test coverage reports or current coverage percentages
find . -name ".coverage" -o -name "coverage.xml" -o -name "htmlcov" 2>/dev/null | head -5Repository: cortexlinux/cortex
Length of output: 44
🏁 Script executed:
#!/bin/bash
# Look for any integration or end-to-end tests that might indirectly test watch_service
rg -l "watch\|Watch" tests/ 2>/dev/null | head -10Repository: cortexlinux/cortex
Length of output: 44
🏁 Script executed:
#!/bin/bash
# Get the exact line and method context for loginctl call to assess severity
sed -n '620,630p' ./cortex/watch_service.py | cat -nRepository: cortexlinux/cortex
Length of output: 619
🏁 Script executed:
#!/antml:bin
# Check if threading.Lock or any lock is imported or used anywhere
grep -i "lock\|rlock\|semaphore\|mutex" ./cortex/watch_service.pyRepository: cortexlinux/cortex
Length of output: 44
Add comprehensive tests for the watch daemon and installer, verify >80% coverage, and fix threading/type-hint gaps.
This substantial new feature (daemon lifecycle, systemd service management, three daemon threads) has zero test coverage. Additionally:
- Threading without locks:
_watch_hook_commands(set) and_recent_commands(list) are accessed by multiple daemon threads without synchronization primitives; addthreading.Lock()to protect these. - Missing type hints: Multiple public/private methods lack return type annotations (
__init__,_handle_signal,_handle_reload,_load_state,_save_state,_monitor_bash_history,_monitor_watch_hook,_cleanup_stale_terminals,start,stop); add these per PEP 8. - loginctl edge case (line 624):
os.getenv("USER", "")can pass an empty string tologinctl enable-linger; validateUSERis not empty before calling loginctl or catch the failure gracefully.
Per coding guidelines, maintain >80% test coverage for pull requests.
🤖 Prompt for AI Agents
In `@cortex/watch_service.py` around lines 30 - 52, The daemon has thread-safety,
typing, and an edge-case issue: protect concurrent access to
_watch_hook_commands and _recent_commands by adding a threading.Lock() (e.g.,
self._cmd_lock) and using it around all reads/writes in methods like
_monitor_watch_hook, _monitor_bash_history, and any other thread functions; add
explicit return type annotations per PEP 484/8 for the listed methods (__init__,
_handle_signal, _handle_reload, _load_state, _save_state, _monitor_bash_history,
_monitor_watch_hook, _cleanup_stale_terminals, start, stop) (use None for void
methods and proper types for loaders), and handle the loginctl edge case by
checking os.getenv("USER") is non-empty before calling loginctl enable-linger
(or catch/handle subprocess errors where loginctl is invoked) in the code path
around the loginctl call so failures are graceful. Ensure updates reference the
class CortexWatchDaemon and its attributes (_watch_hook_commands,
_recent_commands, _cmd_lock, terminals_dir) and that unit tests are added to
cover threading, type annotations, and the USER/loginctl edge case to keep
coverage >80%.
| # Track commands seen from watch_hook to avoid duplicates with bash_history | ||
| self._watch_hook_commands: set[str] = set() |
There was a problem hiding this comment.
Bound and synchronize _watch_hook_commands.
The set is mutated across threads and can grow unbounded if history never yields a matching command. Consider a bounded deque/TTL map and guard with a lock (possibly the same lock used for recent-command dedupe).
Also applies to: 369-422
🤖 Prompt for AI Agents
In `@cortex/watch_service.py` around lines 45 - 46, The _watch_hook_commands set
is mutated from multiple threads and can grow unbounded; replace it with a
bounded structure (e.g., collections.deque(maxlen=...) or a TTL map) and protect
all accesses with a lock — preferably reuse the existing recent-command dedupe
lock used elsewhere (the same lock guarding recent-command dedupe logic
referenced in the review) or add a dedicated threading.Lock (e.g.,
self._watch_hook_lock) if that lock is not accessible; update all code paths
that read/write _watch_hook_commands (including the code around lines 369-422)
to acquire the lock before mutating or checking membership to ensure
thread-safety and bounded memory usage.
| def _handle_signal(self, signum, frame): | ||
| """Handle shutdown signals.""" | ||
| self.log(f"Received signal {signum}, shutting down...") | ||
| self.running = False | ||
|
|
||
| def _handle_reload(self, signum, frame): | ||
| """Handle reload signal (SIGHUP).""" | ||
| self.log("Received SIGHUP, reloading configuration...") | ||
| self._load_state() | ||
|
|
||
| def log(self, message: str): | ||
| """Log a message to the service log.""" | ||
| log_file = self.cortex_dir / "watch_service.log" | ||
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | ||
| with open(log_file, "a") as f: | ||
| f.write(f"[{timestamp}] {message}\n") | ||
|
|
||
| def _load_state(self): | ||
| """Load saved state from file.""" | ||
| if self.state_file.exists(): | ||
| try: | ||
| with open(self.state_file) as f: | ||
| state = json.load(f) | ||
| self.terminal_counter = state.get("terminal_counter", 0) | ||
| self.terminals = state.get("terminals", {}) | ||
| except Exception as e: | ||
| self.log(f"Error loading state: {e}") | ||
|
|
||
| def _save_state(self): | ||
| """Save current state to file.""" | ||
| try: | ||
| state = { | ||
| "terminal_counter": self.terminal_counter, | ||
| "terminals": self.terminals, | ||
| "last_update": datetime.datetime.now().isoformat(), | ||
| } | ||
| with open(self.state_file, "w") as f: | ||
| json.dump(state, f, indent=2) | ||
| except Exception as e: | ||
| self.log(f"Error saving state: {e}") | ||
|
|
||
| def _get_terminal_id(self, pts: str) -> str: | ||
| """Generate or retrieve a unique terminal ID.""" | ||
| if pts in self.terminals: | ||
| return self.terminals[pts]["id"] | ||
|
|
||
| self.terminal_counter += 1 | ||
| terminal_id = f"term_{self.terminal_counter:04d}" | ||
|
|
||
| self.terminals[pts] = { | ||
| "id": terminal_id, | ||
| "pts": pts, | ||
| "created": datetime.datetime.now().isoformat(), | ||
| "is_cortex": False, | ||
| "command_count": 0, | ||
| } | ||
|
|
||
| self._save_state() | ||
| return terminal_id | ||
|
|
||
| def _is_cortex_terminal(self, pid: int) -> bool: | ||
| """Check if a process is a Cortex terminal.""" | ||
| try: | ||
| # Check environment variables | ||
| environ_file = Path(f"/proc/{pid}/environ") | ||
| if environ_file.exists(): | ||
| environ = environ_file.read_bytes() | ||
| if b"CORTEX_TERMINAL=1" in environ: | ||
| return True | ||
|
|
||
| # Check command line | ||
| cmdline_file = Path(f"/proc/{pid}/cmdline") | ||
| if cmdline_file.exists(): | ||
| cmdline = cmdline_file.read_bytes().decode("utf-8", errors="ignore") | ||
| if "cortex" in cmdline.lower(): | ||
| return True | ||
| except (PermissionError, FileNotFoundError, ProcessLookupError): | ||
| pass | ||
|
|
||
| return False | ||
|
|
||
| def _get_active_terminals(self) -> list[dict]: | ||
| """Get list of active terminal processes.""" | ||
| terminals = [] | ||
|
|
||
| try: | ||
| # Find all pts (pseudo-terminal) devices | ||
| pts_dir = Path("/dev/pts") | ||
| if pts_dir.exists(): | ||
| for pts_file in pts_dir.iterdir(): | ||
| if pts_file.name.isdigit(): | ||
| pts_path = str(pts_file) | ||
|
|
||
| # Find process using this pts | ||
| result = subprocess.run( | ||
| ["fuser", pts_path], capture_output=True, text=True, timeout=2 | ||
| ) | ||
|
|
||
| if result.stdout.strip(): | ||
| pids = result.stdout.strip().split() | ||
| for pid_str in pids: | ||
| try: | ||
| pid = int(pid_str) | ||
| is_cortex = self._is_cortex_terminal(pid) | ||
| terminal_id = self._get_terminal_id(pts_path) | ||
|
|
||
| # Update cortex flag | ||
| if pts_path in self.terminals: | ||
| self.terminals[pts_path]["is_cortex"] = is_cortex | ||
|
|
||
| terminals.append( | ||
| { | ||
| "pts": pts_path, | ||
| "pid": pid, | ||
| "id": terminal_id, | ||
| "is_cortex": is_cortex, | ||
| } | ||
| ) | ||
| except ValueError: | ||
| continue | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Error getting terminals: {e}") | ||
|
|
||
| return terminals | ||
|
|
||
| def _monitor_bash_history(self): | ||
| """Monitor bash history for new commands using inotify if available.""" | ||
| history_files = [ | ||
| Path.home() / ".bash_history", | ||
| Path.home() / ".zsh_history", | ||
| ] | ||
|
|
||
| positions: dict[str, int] = {} | ||
| last_commands: dict[str, str] = {} # Track last command per file to avoid duplicates | ||
|
|
||
| # Initialize positions to current end of file | ||
| for hist_file in history_files: | ||
| if hist_file.exists(): | ||
| positions[str(hist_file)] = hist_file.stat().st_size | ||
| # Read last line to track for dedup | ||
| try: | ||
| content = hist_file.read_text() | ||
| lines = content.strip().split("\n") | ||
| if lines: | ||
| last_commands[str(hist_file)] = lines[-1].strip() | ||
| except Exception: | ||
| pass | ||
|
|
||
| # Try to use inotify for more efficient monitoring | ||
| try: | ||
| import ctypes | ||
| import select | ||
| import struct | ||
|
|
||
| # Check if inotify is available | ||
| libc = ctypes.CDLL("libc.so.6") | ||
| inotify_init = libc.inotify_init | ||
| inotify_add_watch = libc.inotify_add_watch | ||
|
|
||
| IN_MODIFY = 0x00000002 | ||
| IN_CLOSE_WRITE = 0x00000008 | ||
|
|
||
| fd = inotify_init() | ||
| if fd < 0: | ||
| raise OSError("Failed to initialize inotify") | ||
|
|
||
| watches = {} | ||
| for hist_file in history_files: | ||
| if hist_file.exists(): | ||
| wd = inotify_add_watch(fd, str(hist_file).encode(), IN_MODIFY | IN_CLOSE_WRITE) | ||
| if wd >= 0: | ||
| watches[wd] = hist_file | ||
|
|
||
| self.log(f"Using inotify to monitor {len(watches)} history files") | ||
|
|
||
| while self.running: | ||
| # Wait for inotify event with timeout | ||
| r, _, _ = select.select([fd], [], [], 1.0) | ||
| if not r: | ||
| continue | ||
|
|
||
| data = os.read(fd, 4096) | ||
| # Process inotify events | ||
| for hist_file in history_files: | ||
| key = str(hist_file) | ||
| if not hist_file.exists(): | ||
| continue | ||
|
|
||
| try: | ||
| current_size = hist_file.stat().st_size | ||
|
|
||
| if key not in positions: | ||
| positions[key] = current_size | ||
| continue | ||
|
|
||
| if current_size < positions[key]: | ||
| positions[key] = current_size | ||
| continue | ||
|
|
||
| if current_size > positions[key]: | ||
| with open(hist_file) as f: | ||
| f.seek(positions[key]) | ||
| new_content = f.read() | ||
|
|
||
| for line in new_content.split("\n"): | ||
| line = line.strip() | ||
| # Skip empty, short, or duplicate commands | ||
| if line and len(line) > 1: | ||
| if last_commands.get(key) != line: | ||
| self._log_command(line, "history") | ||
| last_commands[key] = line | ||
|
|
||
| positions[key] = current_size | ||
| except Exception as e: | ||
| self.log(f"Error reading {hist_file}: {e}") | ||
|
|
||
| os.close(fd) | ||
| return | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Inotify not available, using polling: {e}") | ||
|
|
||
| # Fallback to polling | ||
| while self.running: | ||
| for hist_file in history_files: | ||
| if not hist_file.exists(): | ||
| continue | ||
|
|
||
| key = str(hist_file) | ||
| try: | ||
| current_size = hist_file.stat().st_size | ||
|
|
||
| if key not in positions: | ||
| positions[key] = current_size | ||
| continue | ||
|
|
||
| if current_size < positions[key]: | ||
| # File was truncated | ||
| positions[key] = current_size | ||
| continue | ||
|
|
||
| if current_size > positions[key]: | ||
| with open(hist_file) as f: | ||
| f.seek(positions[key]) | ||
| new_content = f.read() | ||
|
|
||
| for line in new_content.split("\n"): | ||
| line = line.strip() | ||
| if line and len(line) > 1: | ||
| if last_commands.get(key) != line: | ||
| self._log_command(line, "history") | ||
| last_commands[key] = line | ||
|
|
||
| positions[key] = current_size | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Error reading {hist_file}: {e}") | ||
|
|
||
| time.sleep(0.3) | ||
|
|
||
| def _monitor_watch_hook(self): | ||
| """Monitor the watch hook log file and sync to terminal_commands.json.""" | ||
| position = 0 | ||
|
|
||
| while self.running: | ||
| try: | ||
| if not self.watch_log.exists(): | ||
| time.sleep(0.5) | ||
| continue | ||
|
|
||
| current_size = self.watch_log.stat().st_size | ||
|
|
||
| if current_size < position: | ||
| position = 0 | ||
|
|
||
| if current_size > position: | ||
| with open(self.watch_log) as f: | ||
| f.seek(position) | ||
| new_content = f.read() | ||
|
|
||
| for line in new_content.split("\n"): | ||
| line = line.strip() | ||
| if not line or len(line) < 2: | ||
| continue | ||
|
|
||
| # Parse format: TTY|COMMAND (new format from updated hook) | ||
| # Skip lines that don't have the TTY| prefix or have "shared|" | ||
| if "|" not in line: | ||
| continue | ||
|
|
||
| parts = line.split("|", 1) | ||
| terminal_id = parts[0] | ||
|
|
||
| # Skip "shared" entries (those come from bash_history monitor) | ||
| if terminal_id == "shared": | ||
| continue | ||
|
|
||
| # Must have valid TTY format (pts_X, tty_X, etc.) | ||
| if not terminal_id or terminal_id == "unknown": | ||
| continue | ||
|
|
||
| command = parts[1] if len(parts) > 1 else "" | ||
| if not command: | ||
| continue | ||
|
|
||
| # Skip duplicates | ||
| if self._is_duplicate(command): | ||
| continue | ||
|
|
||
| # Mark this command as seen from watch_hook | ||
| self._watch_hook_commands.add(command) | ||
|
|
||
| # Log to terminal_commands.json only | ||
| self._log_to_json(command, "watch_hook", terminal_id) | ||
|
|
||
| position = current_size | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Error monitoring watch hook: {e}") | ||
|
|
||
| time.sleep(0.2) | ||
|
|
||
| def _log_to_json(self, command: str, source: str, terminal_id: str): | ||
| """Log a command only to terminal_commands.json.""" | ||
| try: | ||
| detailed_log = self.cortex_dir / "terminal_commands.json" | ||
| entry = { | ||
| "timestamp": datetime.datetime.now().isoformat(), | ||
| "command": command, | ||
| "source": source, | ||
| "terminal_id": terminal_id, | ||
| } | ||
|
|
||
| with open(detailed_log, "a") as f: | ||
| f.write(json.dumps(entry) + "\n") | ||
| except Exception as e: | ||
| self.log(f"Error logging to JSON: {e}") | ||
|
|
||
| def _is_duplicate(self, command: str) -> bool: | ||
| """Check if command was recently logged to avoid duplicates.""" | ||
| if command in self._recent_commands: | ||
| return True | ||
|
|
||
| # Keep last 100 commands | ||
| self._recent_commands.append(command) | ||
| if len(self._recent_commands) > 100: | ||
| self._recent_commands.pop(0) | ||
|
|
||
| return False | ||
|
|
||
| def _log_command(self, command: str, source: str = "unknown", terminal_id: str | None = None): | ||
| """Log a command from bash_history (watch_hook uses _log_to_json directly).""" | ||
| # Skip cortex commands | ||
| if command.lower().startswith("cortex "): | ||
| return | ||
| if "watch_hook" in command: | ||
| return | ||
| if command.startswith("source ") and ".cortex" in command: | ||
| return | ||
|
|
||
| # Skip if this command was already logged by watch_hook | ||
| if command in self._watch_hook_commands: | ||
| self._watch_hook_commands.discard(command) # Clear it for next time | ||
| return | ||
|
|
||
| # Skip duplicates | ||
| if self._is_duplicate(command): | ||
| return | ||
|
|
||
| # For bash_history source, we can't know which terminal - use "shared" | ||
| if terminal_id is None: | ||
| terminal_id = "shared" | ||
|
|
||
| try: | ||
| # Write to watch_log with format TTY|COMMAND | ||
| with open(self.watch_log, "a") as f: | ||
| f.write(f"{terminal_id}|{command}\n") | ||
|
|
||
| # Log to JSON | ||
| self._log_to_json(command, source, terminal_id) | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Error logging command: {e}") | ||
|
|
||
| def _cleanup_stale_terminals(self): | ||
| """Remove stale terminal entries.""" | ||
| while self.running: | ||
| try: | ||
| active_pts = set() | ||
| pts_dir = Path("/dev/pts") | ||
| if pts_dir.exists(): | ||
| for pts_file in pts_dir.iterdir(): | ||
| if pts_file.name.isdigit(): | ||
| active_pts.add(str(pts_file)) | ||
|
|
||
| # Remove stale entries | ||
| stale = [pts for pts in self.terminals if pts not in active_pts] | ||
| for pts in stale: | ||
| del self.terminals[pts] | ||
|
|
||
| if stale: | ||
| self._save_state() | ||
|
|
||
| except Exception as e: | ||
| self.log(f"Error cleaning up terminals: {e}") | ||
|
|
||
| time.sleep(30) # Check every 30 seconds | ||
|
|
||
| def start(self): | ||
| """Start the watch daemon.""" | ||
| # Check if already running | ||
| if self.pid_file.exists(): | ||
| try: | ||
| pid = int(self.pid_file.read_text().strip()) | ||
| os.kill(pid, 0) # Check if process exists | ||
| self.log(f"Daemon already running with PID {pid}") | ||
| return False | ||
| except (ProcessLookupError, ValueError): | ||
| # Stale PID file | ||
| self.pid_file.unlink() | ||
|
|
||
| # Write PID file | ||
| self.pid_file.write_text(str(os.getpid())) | ||
|
|
||
| self.running = True | ||
| self._load_state() | ||
|
|
||
| self.log("Cortex Watch Service starting...") | ||
|
|
||
| # Start monitor threads | ||
| threads = [ | ||
| threading.Thread(target=self._monitor_bash_history, daemon=True), | ||
| threading.Thread(target=self._monitor_watch_hook, daemon=True), | ||
| threading.Thread(target=self._cleanup_stale_terminals, daemon=True), | ||
| ] | ||
|
|
||
| for t in threads: | ||
| t.start() | ||
|
|
||
| self.log(f"Cortex Watch Service started (PID: {os.getpid()})") | ||
|
|
||
| # Main loop - just keep alive and handle signals | ||
| try: | ||
| while self.running: | ||
| time.sleep(1) | ||
| finally: | ||
| self._shutdown() | ||
|
|
||
| return True | ||
|
|
||
| def _shutdown(self): | ||
| """Clean shutdown.""" | ||
| self.log("Shutting down...") | ||
| self._save_state() | ||
|
|
||
| if self.pid_file.exists(): | ||
| self.pid_file.unlink() | ||
|
|
||
| self.log("Cortex Watch Service stopped") | ||
|
|
||
| def stop(self): | ||
| """Stop the running daemon.""" | ||
| if not self.pid_file.exists(): | ||
| return False, "Service not running" | ||
|
|
||
| try: | ||
| pid = int(self.pid_file.read_text().strip()) | ||
| os.kill(pid, signal.SIGTERM) | ||
|
|
||
| # Wait for process to exit | ||
| for _ in range(10): | ||
| try: | ||
| os.kill(pid, 0) | ||
| time.sleep(0.5) | ||
| except ProcessLookupError: | ||
| break | ||
|
|
||
| return True, f"Service stopped (PID: {pid})" | ||
|
|
||
| except ProcessLookupError: | ||
| self.pid_file.unlink() | ||
| return True, "Service was not running" | ||
| except Exception as e: | ||
| return False, f"Error stopping service: {e}" | ||
|
|
||
| def status(self) -> dict: | ||
| """Get service status.""" | ||
| status = { | ||
| "running": False, | ||
| "pid": None, | ||
| "terminals": 0, | ||
| "commands_logged": 0, | ||
| } | ||
|
|
||
| if self.pid_file.exists(): | ||
| try: | ||
| pid = int(self.pid_file.read_text().strip()) | ||
| os.kill(pid, 0) | ||
| status["running"] = True | ||
| status["pid"] = pid | ||
| except (ProcessLookupError, ValueError): | ||
| pass | ||
|
|
||
| if self.watch_log.exists(): | ||
| try: | ||
| content = self.watch_log.read_text() | ||
| status["commands_logged"] = len([l for l in content.split("\n") if l.strip()]) | ||
| except Exception: | ||
| pass | ||
|
|
||
| self._load_state() | ||
| status["terminals"] = len(self.terminals) | ||
|
|
||
| return status |
There was a problem hiding this comment.
Add type hints to daemon methods and helpers.
Several methods (e.g., _handle_signal, _monitor_bash_history, start, stop, status, main) lack parameter/return annotations. Please add explicit types (including -> None where appropriate) to meet typing requirements.
As per coding guidelines, type hints are required in Python code.
🤖 Prompt for AI Agents
In `@cortex/watch_service.py` around lines 58 - 572, Add explicit type annotations
to the daemon methods and helpers mentioned: annotate _handle_signal(self,
signum: int, frame: Optional[types.FrameType]) -> None and _handle_reload(self,
signum: int, frame: Optional[types.FrameType]) -> None; annotate log(self,
message: str) -> None, _load_state(self) -> None, _save_state(self) -> None,
_monitor_bash_history(self) -> None, _monitor_watch_hook(self) -> None,
_log_to_json(self, command: str, source: str, terminal_id: str) -> None (ensure
existing signature matches), _log_command(self, command: str, source: str =
"unknown", terminal_id: Optional[str] = None) -> None,
_cleanup_stale_terminals(self) -> None, _shutdown(self) -> None, start(self) ->
bool, and stop(self) -> tuple[bool, str]; import any needed typing names
(Optional, types.FrameType) at top and update signatures for these functions
accordingly to satisfy typing requirements.
| def _log_to_json(self, command: str, source: str, terminal_id: str): | ||
| """Log a command only to terminal_commands.json.""" | ||
| try: | ||
| detailed_log = self.cortex_dir / "terminal_commands.json" | ||
| entry = { | ||
| "timestamp": datetime.datetime.now().isoformat(), | ||
| "command": command, | ||
| "source": source, | ||
| "terminal_id": terminal_id, | ||
| } | ||
|
|
||
| with open(detailed_log, "a") as f: | ||
| f.write(json.dumps(entry) + "\n") | ||
| except Exception as e: |
There was a problem hiding this comment.
Concurrent JSON log writes can interleave.
Multiple threads call _log_to_json, so writes can interleave and corrupt JSON lines. Add a file-write lock or funnel writes through a single logging thread.
🤖 Prompt for AI Agents
In `@cortex/watch_service.py` around lines 381 - 394, Concurrent writes in
_log_to_json can interleave and corrupt the JSON lines; fix by serializing file
access—either add a dedicated threading.Lock (e.g., self._json_write_lock
initialized in the class __init__) and wrap the open/write block in a with
self._json_write_lock: section, or implement a single writer thread with a Queue
(e.g., enqueue entry objects in _log_to_json and have a background _json_writer
thread consume the queue and perform open/append writes). Update _log_to_json to
enqueue or to acquire the lock before opening/writing and ensure any new
background thread is started/stopped in the service lifecycle methods.
1 participant
[Feature]
cortex ask --do- AI-Powered Command Execution with Auto-RepairRelated Issue
Closes #662
Summary
This PR introduces
cortex ask --do, an interactive AI assistant that executes Linux commands based on natural language requests. It features intelligent error handling, automatic repair, resource conflict detection, and real-time terminal monitoring.AI Disclosure
Used Claude (Anthropic) via Cursor IDE for:
Checklist
type(scope): descriptionor[scope] descriptionpytest tests/)What's New
🚀 Core Feature:
cortex ask --doAn AI-powered command execution mode that:
Example Usage
Key Features
1. Natural Language → Commands
2. Resource Conflict Detection
Before executing, checks for:
User can choose to: use existing, restart, or recreate
3. Auto-Repair System
command not foundpermission deniedservice not runningport in use4. Manual Intervention Mode
For sudo commands or user preference:
5. Terminal Monitoring
6. Interactive Session
After execution:
7. LLM-Generated Summary
At the end of execution, generates:
Files Changed
New Files
cortex/do_runner/handler.pycortex/do_runner/executor.pycortex/do_runner/verification.pycortex/do_runner/diagnosis.pycortex/do_runner/terminal.pycortex/do_runner/models.pycortex/do_runner/database.pycortex/watch_service.pyscripts/setup_ask_do.pydocs/ASK_DO_ARCHITECTURE.mdModified Files
cortex/cli.py--doflag, interactive session, signal handlerscortex/ask.pyREADME.mdArchitecture
Testing
Manual Testing Performed
Test Commands
Setup Instructions
For Users
For Developers
Screenshots / Demo
Command Execution with Conflict Detection
Auto-Repair in Action
Execution Summary
Quality & Security
Code Quality
Testing Performed
Documentation
docs/ASK_DO_ARCHITECTURE.md)Security Measures
Breaking Changes
None. This is a new feature addition.
Dependencies
No new external dependencies. Uses existing:
rich- Terminal UIanthropic- Claude API (existing)subprocess,sqlite3,signalFuture Improvements
Reviewers
@cortexlinux/maintainers
Labels
feature,ai,terminal,ready-for-reviewSummary by CodeRabbit
New Features
Documentation
UX
✏️ Tip: You can customize this high-level summary in your review settings.