feat: fixed package manager issues with retry circuit logic

Author: haseeb-heaven

.gitignore

@@ -22,6 +22,7 @@ venv/
 myvenv/
 heaven_venv/
 .venv/
+node_modules/
 
 # Interactive environment data
 interp/

libs/interpreter_lib.py

@@ -32,7 +32,6 @@
 import shlex
 import shutil
 from rich.console import Console
-from rich.text import Text
 
 litellm.set_verbose = False
 litellm.suppress_debug_info = True
@@ -1242,7 +1241,7 @@ def interpreter_main(self,  version):
 				elif any(command in task.lower() for command in ['/mode ']):
 					mode = task.split(' ')[1]
 					if mode:
-						if not mode.lower() in ['code','script','command','vision','chat']:
+						if mode.lower() not in ['code','script','command','vision','chat']:
 							mode = 'code'
 							display_markdown_message(f"The input mode is not supported. Mode changed to {mode},"
 													 "\nUse '/list' command to get the list of supported modes.")
@@ -1290,22 +1289,46 @@ def interpreter_main(self,  version):
 					continue
 
 				# INSTALL - Command section.
-				elif any(command in task.lower() for command in ['/install']):
-					# get the package name after the command 
-					package_name = task.split(' ')[1]
+				elif task.lower().startswith('/install'):
+					parts = task.split(' ')
+					if len(parts) >= 3:
+						# Case: /install <language> <package>
+						language = parts[1].lower()
+						package_name = parts[2]
+						
+						# Validate language
+						if language in ['python', 'py']:
+							language = 'python'
+						elif language in ['javascript', 'js', 'node']:
+							language = 'javascript'
+						else:
+							# If unknown language, treat as package name to be safe
+							package_name = parts[1]
+							language = self.INTERPRETER_LANGUAGE
+					elif len(parts) == 2:
+						package_name = parts[1]
+						language = self.INTERPRETER_LANGUAGE
+					else:
+						display_markdown_message("Usage: **/install [language] <package_name>**")
+						continue
 
 					# check if package name is not system module.
 					system_modules = self.package_manager.get_system_modules()
 
-					# Skip installing system modules.
 					if package_name in system_modules:
 						self.logger.info(f"Package {package_name} is a system module.")
 						display_markdown_message(f"Package {package_name} is a system module.")
-						raise Exception(f"Package {package_name} is a system module.")
+						continue
 
 					if package_name:
-						self.logger.info(f"Installing package {package_name} on interpreter {self.INTERPRETER_LANGUAGE}")
-						self.package_manager.install_package(package_name, self.INTERPRETER_LANGUAGE)
+						self.logger.info(f"Installing package {package_name} for {language}")
+						display_markdown_message(f"Installing package **{package_name}** for **{language}**...")
+						try:
+							self.package_manager.install_package(package_name, language)
+							display_markdown_message(f"Successfully installed **{package_name}**")
+						except Exception as ex:
+							self.logger.error(f"Manual installation failed: {ex}")
+							display_markdown_message(f"Failed to install **{package_name}**: {ex}")
 					continue
 
 				# Get the prompt based on the mode.
@@ -1492,21 +1515,32 @@ def interpreter_main(self,  version):
 							raise Exception(f"Package {package_name} is a system module.")
 
 						if package_name:
-							self.logger.info(f"Installing package {package_name} on interpreter {self.INTERPRETER_LANGUAGE}")
-							self.package_manager.install_package(package_name, self.INTERPRETER_LANGUAGE)
-
-							# Wait and Execute the code again.
-							time.sleep(3)
-							code_output, code_error = self._execute_generated_output(code_snippet, os_name, force_execute=True)
-							if code_output:
-								self.logger.info(f"{self.INTERPRETER_LANGUAGE} code executed successfully.")
-								display_code(code_output)
-								self.logger.info(f"Output: {code_output[:100]}")
-							elif code_error:
-								self.logger.info(f"{self.INTERPRETER_LANGUAGE} code executed with error.")
-								display_markdown_message(f"Error: {code_error}")
-							else:
-								display_markdown_message("Execution completed successfully. No stdout was produced.")
+							for attempt in range(1, 4):
+								try:
+									self.logger.info(f"Installing package {package_name} on interpreter {self.INTERPRETER_LANGUAGE} (Attempt {attempt}/3)")
+									self.package_manager.install_package(package_name, self.INTERPRETER_LANGUAGE)
+
+									# Wait and Execute the code again.
+									time.sleep(3)
+									code_output, code_error = self._execute_generated_output(code_snippet, os_name, force_execute=True)
+									if code_output:
+										self.logger.info(f"{self.INTERPRETER_LANGUAGE} code executed successfully.")
+										display_code(code_output)
+										self.logger.info(f"Output: {code_output[:100]}")
+									elif code_error:
+										self.logger.info(f"{self.INTERPRETER_LANGUAGE} code executed with error.")
+										display_markdown_message(f"Error: {code_error}")
+									else:
+										display_markdown_message("Execution completed successfully. No stdout was produced.")
+									break  # Exit retry loop on success
+								except Exception as ex:
+									if attempt < 3:
+										self.logger.warning(f"Attempt {attempt} to install package {package_name} failed: {ex}")
+										display_markdown_message(f"Attempt {attempt}/3 to install package **{package_name}** failed. Retrying in 2 seconds...")
+										time.sleep(2)
+									else:
+										self.logger.error(f"Failed to install package {package_name} after 3 attempts: {ex}")
+										display_markdown_message(f"Failed to install package **{package_name}** after 3 attempts. Error: {ex}. Proceeding with repair logic...")
 
 					if code_error and not code_error.startswith("Safety blocked:"):
 						code_snippet, repaired_output, repaired_error = self._attempt_repair_after_failure(

libs/package_manager.py

@@ -1,8 +1,8 @@
 import subprocess
 import re
-import logging
 import requests
 import stdlib_list
+import os
 from libs.logger import Logger
 
 
@@ -15,6 +15,15 @@ def __init__(self):
 		self.npm_command = "npm"
 		self.logger = Logger.initialize("logs/interpreter.log")
 
+	def _run_command(self, args):
+		"""Run a shell command with OS-appropriate settings."""
+		try:
+			# On Windows, shell=True is needed to resolve script-based commands like npm or pip
+			use_shell = os.name == 'nt'
+			return subprocess.check_call(args, shell=use_shell)
+		except subprocess.CalledProcessError as e:
+			raise e
+
 	def install_package(self, package_name, language):
 		if language == "python":
 			if not self._check_package_exists_pip(package_name):
@@ -77,23 +86,23 @@ def get_system_modules(self):
 
 	def _install_package_with_pip(self,  package_name):
 		try:
-			subprocess.check_call([self.pip_command, "install", package_name])
+			self._run_command([self.pip_command, "install", package_name])
 			self.logger.info(f"Successfully installed package with pip: {package_name}")
 		except subprocess.CalledProcessError as exception:
 			self.logger.error(f"Failed to install package with pip: {package_name}")
 			raise exception
 
 	def _install_package_with_pip3(self,  package_name):
 		try:
-			subprocess.check_call([self.pip3_command, "install", package_name])
+			self._run_command([self.pip3_command, "install", package_name])
 			self.logger.info(f"Successfully installed package with pip3: {package_name}")
 		except subprocess.CalledProcessError as exception:
 			self.logger.error(f"Failed to install package with pip3: {package_name}")
 			raise exception
 
 	def _install_package_with_npm(self,  package_name):
 		try:
-			subprocess.check_call([self.npm_command, "install", package_name])
+			self._run_command([self.npm_command, "install", package_name])
 			self.logger.info(f"Successfully installed package with npm: {package_name}")
 		except subprocess.CalledProcessError as exception:
 			self.logger.error(f"Failed to install package with npm: {package_name}")
@@ -102,20 +111,20 @@ def _install_package_with_npm(self,  package_name):
 	def _is_package_installed(self,  package_name, package_manager):
 		if package_manager == "pip":
 			try:
-				subprocess.check_call([self.pip_command, "show", package_name])
+				self._run_command([self.pip_command, "show", package_name])
 				self.logger.info(f"Package {package_name} is installed")
 				return True
 			except subprocess.CalledProcessError:
 				try:
-					subprocess.check_call([self.pip3_command, "show", package_name])
+					self._run_command([self.pip3_command, "show", package_name])
 					self.logger.info(f"Package {package_name} is installed")
 					return True
 				except subprocess.CalledProcessError:
 					self.logger.info(f"Package {package_name} is not installed")
 					return False
 		elif package_manager == "npm":
 			try:
-				subprocess.check_call([self.npm_command, "list", "-g", package_name])
+				self._run_command([self.npm_command, "list", "-g", package_name])
 				self.logger.info(f"Package {package_name} is installed")
 				return True
 			except subprocess.CalledProcessError:
@@ -169,12 +178,12 @@ def _check_package_exists_pip(self, package_name):
 
 	def _check_package_exists_npm(self,  package_name):
 		try:
-			response = requests.get(f"https://www.npmjs.com/package/{package_name}")
+			response = requests.get(f"https://registry.npmjs.org/{package_name}")
 			if response.status_code == 200:
-				self.logger.info(f"Package {package_name} exists on npm website")
+				self.logger.info(f"Package {package_name} exists on npm registry")
 				return True
 			else:
-				self.logger.info(f"Package {package_name} does not exist on npm website")
+				self.logger.info(f"Package {package_name} does not exist on npm registry")
 				return False
 		except requests.exceptions.RequestException as exception:
 			self.logger.error(f"Failed to check package existence on npm website: {exception}")