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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 129 additions & 5 deletions cortex/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,19 @@ def __init__(self, verbose: bool = False):
self.verbose = verbose
self.offline = False

def _build_prompt_with_stdin(self, user_prompt: str) -> str:
"""
Combine optional stdin context with user prompt.
"""
if getattr(self, "stdin_data", None):
return (
"Context (from stdin):\n"
f"{self.stdin_data}\n\n"
"User instruction:\n"
f"{user_prompt}"
)
return user_prompt

def _debug(self, message: str):
"""Print debug info only in verbose mode"""
if self.verbose:
Expand Down Expand Up @@ -325,6 +338,10 @@ def install(
if not is_valid:
self._print_error(error)
return 1
api_key = self._get_api_key()
if not api_key:
self._print_error("No API key configured")
return 1

# Special-case the ml-cpu stack:
# The LLM sometimes generates outdated torch==1.8.1+cpu installs
Expand All @@ -339,10 +356,6 @@ def install(
"pip3 install jupyter numpy pandas"
)

api_key = self._get_api_key()
if not api_key:
return 1

provider = self._get_provider()
self._debug(f"Using provider: {provider}")
self._debug(f"API key: {api_key[:10]}...{api_key[-4:]}")
Expand All @@ -355,17 +368,84 @@ def install(
try:
self._print_status("🧠", "Understanding request...")

api_key = self._get_api_key()
if not api_key:
self._print_error("No API key configured")
return 1

interpreter = CommandInterpreter(
api_key=api_key, provider=provider, offline=self.offline
)
intent = interpreter.extract_intent(software)
# ---------- Extract install mode from intent ----------
install_mode = intent.get("install_mode", "system")

# ---------- NORMALIZE INTENT (ADD THIS) ----------
action = intent.get("action", "unknown")
domain = intent.get("domain", "unknown")

if not isinstance(action, str):
action = "unknown"
if not isinstance(domain, str):
domain = "unknown"

raw_confidence = intent.get("confidence", 0.0)
try:
confidence = float(raw_confidence)
except (TypeError, ValueError):
confidence = 0.0

ambiguous = bool(intent.get("ambiguous", False))
# Normalize unstable model output
if isinstance(action, str) and "|" in action:
action = action.split("|")[0].strip()

# Policy: known domain ⇒ not ambiguous
if domain != "unknown":
ambiguous = False
# ----------------------------------------------

print("\n🤖 I understood your request as:")
print(f"• Action : {action}")
print(f"• Domain : {domain}")
print(f"• Description : {intent.get('description')}")
print(f"• Confidence : {confidence}")

# Handle ambiguous intent
if ambiguous and domain == "unknown":
print("\n❓ Your request is ambiguous.")
print("Please clarify what you want to install.")
return 0

# Handle low confidence
if confidence < 0.4 and execute:
print("\n🤔 I'm not confident I understood your request.")
print("Please rephrase with more details.")
return 1

print() # spacing
# -------------------------------------------

self._print_status("📦", "Planning installation...")

for _ in range(10):
self._animate_spinner("Analyzing system requirements...")
self._clear_line()

commands = interpreter.parse(f"install {software}")
# ---------- Build command-generation prompt ----------
if install_mode == "python":
base_prompt = (
f"install {software}. "
"Use pip and Python virtual environments. "
"Do NOT use sudo or system package managers."
)
else:
base_prompt = f"install {software}"

prompt = self._build_prompt_with_stdin(base_prompt)
# ---------------------------------------------------

commands = interpreter.parse(prompt)

if not commands:
self._print_error(
Expand All @@ -387,6 +467,50 @@ def install(
for i, cmd in enumerate(commands, 1):
print(f" {i}. {cmd}")

# ---------- User confirmation ----------
if execute:
print("\nDo you want to proceed with these commands?")
print(" [y] Yes, execute")
print(" [e] Edit commands")
print(" [n] No, cancel")

choice = input("Enter choice [y/e/n]: ").strip().lower()

if choice == "n":
print("❌ Installation cancelled by user.")
return 0

elif choice == "e":
print("\nEnter edited commands (one per line).")
print("Press ENTER on an empty line to finish:\n")

edited_commands = []
while True:
line = input("> ").strip()
if not line:
break
edited_commands.append(line)

if not edited_commands:
print("❌ No commands provided. Cancelling.")
return 1

commands = edited_commands

print("\n✅ Updated commands:")
for i, cmd in enumerate(commands, 1):
print(f" {i}. {cmd}")

confirm = input("\nExecute edited commands? [y/n]: ").strip().lower()
if confirm != "y":
print("❌ Installation cancelled.")
return 0

elif choice != "y":
print("❌ Invalid choice. Cancelling.")
return 1
# -------------------------------------

if dry_run:
print("\n(Dry run mode - commands not executed)")
if install_id:
Expand Down
Loading
Loading