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
34 changes: 34 additions & 0 deletions interpreter.py
100755 → 100644
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,18 @@


def build_parser():
"""
Create and return the command-line ArgumentParser configured for the Code Interpreter CLI.

The parser includes flags for execution and output control (--exec, --save_code, --display_code),
mode selection (--mode with choices 'code', 'script', 'command', 'vision', 'chat'), model override (--model),
language selection (--lang), history memory (--history), unsafe mode (--unsafe), upgrade workflow (--upgrade),
and optional input file (--file, uses 'prompt.txt' when provided without a value). It also exposes a version
option and enforces mutual exclusion between --cli and --tui UI selection flags.

Returns:
argparse.ArgumentParser: A fully configured ArgumentParser for the interpreter CLI.
"""
parser = argparse.ArgumentParser(description='Code - Interpreter')
parser.add_argument('--exec', '-e', action='store_true', default=False, help='Execute the code')
parser.add_argument('--save_code', '-s', action='store_true', default=False, help='Save the generated code')
Expand All @@ -49,10 +61,26 @@ def build_parser():


def _get_default_model():
"""
Get the default model name used when no model is specified.

Returns:
str: The default model name to use for code generation.
"""
return UtilityManager.get_default_model_name()


def prepare_args(args, argv):
"""
Finalize CLI/TUI selection and populate missing defaults on the parsed arguments.

Parameters:
args (argparse.Namespace): Parsed command-line arguments to finalize; may be modified in-place.
argv (Sequence[str]): Original program argv used to detect whether runtime arguments were provided.

Returns:
The finalized argument namespace, or the value returned by TerminalUI().launch(args) when TUI mode is launched.
"""
no_runtime_args = len(argv) <= 1
if no_runtime_args and not args.cli and not args.tui:
args.tui = True
Expand All @@ -69,6 +97,12 @@ def prepare_args(args, argv):


def main(argv=None):
"""
Parse command-line arguments, prepare runtime settings, and start the interpreter process.

Parameters:
argv (list[str] | None): Optional argument vector to parse; when `None`, `sys.argv` is used. This allows overriding CLI input for testing or embedding.
"""
argv = argv or sys.argv
parser = build_parser()
args = parser.parse_args(argv[1:])
Expand Down
106 changes: 103 additions & 3 deletions libs/code_interpreter.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,27 @@
class CodeInterpreter:

def __init__(self):
"""
Initialize the CodeInterpreter instance and configure its logger.

Sets self.logger to a Logger initialized with the file "logs/code-interpreter.log".
"""
self.logger = Logger.initialize("logs/code-interpreter.log")

def _get_subprocess_security_kwargs(self, sandbox_context=None):
"""
Builds subprocess keyword arguments applying working directory, environment, and OS-specific process isolation flags.

Parameters:
sandbox_context (optional): An object that may have `cwd` and `env` attributes; those values (or `None` if absent) are used to populate the corresponding subprocess kwargs.

Returns:
dict: A mapping suitable for passing to subprocess functions containing:
- `cwd`: the working directory from `sandbox_context.cwd` or `None`.
- `env`: the environment mapping from `sandbox_context.env` or `None`.
- On Windows (`os.name == "nt"`): `creationflags` (int) combining available flags such as `CREATE_NO_WINDOW` and `CREATE_NEW_PROCESS_GROUP`.
- On non-Windows: `start_new_session` set to `True`.
"""
kwargs = {
"cwd": getattr(sandbox_context, "cwd", None),
"env": getattr(sandbox_context, "env", None),
Expand All @@ -34,6 +52,15 @@ def _get_subprocess_security_kwargs(self, sandbox_context=None):
return kwargs

def _build_command_invocation(self, command: str):
"""
Constructs a platform-appropriate command invocation list suitable for passing to subprocess functions.

Parameters:
command (str): The shell command string to execute.

Returns:
list: A list of program and argument tokens that invoke the given command on the current OS (Windows uses `cmd.exe /d /c`, Linux/macOS prefers `/bin/bash --noprofile --norc -lc` when available, otherwise `sh -c`).
"""
if os.name == "nt":
return ["cmd.exe", "/d", "/c", command]
bash_path = "/bin/bash" if os.path.exists("/bin/bash") else None
Expand All @@ -42,6 +69,17 @@ def _build_command_invocation(self, command: str):
return ["sh", "-c", command]

def _execute_script(self, script: str, shell: str, sandbox_context=None):
"""
Execute a script using the specified shell and return its captured output and error text.

Parameters:
script (str): The script text to execute.
shell (str): The shell to use; expected values are `"bash"`, `"powershell"`, or `"applescript"`.
sandbox_context (optional): An object that may provide `cwd`, `env`, and `timeout_seconds` to control the subprocess environment and timeout.

Returns:
(tuple): A pair `(stdout, stderr)` where `stdout` is the trimmed standard output string or `None` if no output, and `stderr` is the trimmed standard error string or `None` if no error. On timeout, `stderr` will be `"Execution timed out."`. If an invalid `shell` is provided, returns `(None, "Invalid shell selected: <shell>")`.
"""
stdout = stderr = None
try:
popen_kwargs = {
Expand Down Expand Up @@ -124,9 +162,19 @@ def save_code(self, filename='output/code_generated.py', code=None):

def extract_code(self, code: str, start_sep='```', end_sep='```', skip_first_line=False, code_mode=False):
"""
Extracts the code from the provided string.
If the string contains the start and end separators, it extracts the code between them.
Otherwise, it returns the original string.
Extracts a code snippet delimited by the provided start and end separators from a text block.

If the input contains triple backticks ("```") but the provided separators are single backticks, the function treats the separators as triple backticks. When a matching fenced region is found, the content between the separators is returned with optional adjustments described below; if no matching separators are present, the original `code` string is returned.

Parameters:
code (str): The input text containing code or plain text. If `None`, the function returns `None`.
start_sep (str): Opening separator that marks the start of the code block (default: "```").
end_sep (str): Closing separator that marks the end of the code block (default: "```").
skip_first_line (bool): When True and `code_mode` is True, skip the first line of the fenced block if the opening separator is not immediately followed by a newline.
code_mode (bool): When True, treat the extracted content as code (affects `skip_first_line` behavior). When False, non-code cleanup is applied (see returns).

Returns:
str or None: The extracted code block (possibly adjusted), the original `code` string if no matching separators are found, or `None` if the input `code` is `None`.
"""
try:
if code is None:
Expand Down Expand Up @@ -172,6 +220,24 @@ def extract_code(self, code: str, start_sep='```', end_sep='```', skip_first_lin
raise Exception(f"Error occurred while extracting code: {exception}")

def execute_code(self, code, language, sandbox_context=None):
"""
Execute the provided source code in the specified language and return its captured output and errors.

Executes `code` using a subprocess for the given `language` and returns the subprocess stdout and stderr as decoded UTF-8 strings. Supports "python" (runs `python -c`) and "javascript" (runs `node -e`). Applies optional sandboxing parameters from `sandbox_context` (cwd, env, and timeout_seconds) to the subprocess invocation.

Parameters:
code (str): Source code to execute.
language (str): Programming language name (e.g., "python", "javascript").
sandbox_context (optional): An object that may provide `cwd`, `env`, and `timeout_seconds` to control subprocess execution and timeout.

Returns:
tuple: `(stdout, stderr)` where each is a decoded UTF-8 string containing the subprocess standard output and standard error.
str: If the provided `code` is empty or only whitespace, returns the message "Code is empty. Cannot execute an empty code."
tuple: `(None, "Execution timed out.")` if the subprocess exceeds the configured timeout.

Raises:
Exception: If required compilers/interpreters are not found, if the language is unsupported, or on other execution errors.
"""
try:
language = language.lower()
self.logger.info(f"Running code: {code[:100]} in language: {language}")
Expand Down Expand Up @@ -226,6 +292,20 @@ def execute_code(self, code, language, sandbox_context=None):
raise exception

def execute_script(self, script:str, os_type:str='macos', sandbox_context=None):
"""
Execute a platform-specific script and return its captured output and error.

Parameters:
script (str): The script content to run.
os_type (str): Target operating system; recognized values include 'macos', 'linux', and 'windows' (case-insensitive).
sandbox_context (optional): Sandbox configuration object (e.g., providing `cwd`, `env`, and `timeout_seconds`) applied to the subprocess invocation.

Returns:
tuple: (stdout, stderr) where `stdout` is the script's standard output string or None, and `stderr` is the script's standard error string or None.

Raises:
ValueError: If `script` or `os_type` is missing, or if `os_type` is not one of 'macos', 'linux', or 'windows'.
"""
output = error = None
try:
if not script:
Expand Down Expand Up @@ -254,6 +334,26 @@ def execute_script(self, script:str, os_type:str='macos', sandbox_context=None):
return output, error

def execute_command(self, command:str, sandbox_context=None):
"""
Execute a shell command in a subprocess and return its captured stdout and stderr.

Parameters:
command (str): The command string to execute; must be provided.
sandbox_context (optional): Optional object that may supply execution parameters:
- cwd: working directory for the subprocess
- env: environment variables mapping for the subprocess
- timeout_seconds: execution timeout in seconds (defaults to 30)
Additionally used to determine OS-specific subprocess kwargs (e.g., creationflags or start_new_session).

Returns:
tuple: (stdout, stderr)
- stdout (str or None): UTF-8 decoded standard output from the command, or None if execution timed out.
- stderr (str): UTF-8 decoded standard error from the command, or the string "Execution timed out." if the process exceeded the timeout.

Raises:
ValueError: If `command` is empty or not provided.
Exception: Re-raises any unexpected exceptions encountered during execution.
"""
try:
if not command:
raise ValueError("Command must be provided.")
Expand Down
71 changes: 69 additions & 2 deletions libs/history_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,12 @@

class History:
def __init__(self, history_file: str):
"""
Initialize the History instance, set up its logger, and ensure the history file and its parent directory exist (creating them if necessary). If the history file is newly created, write an empty JSON array into it.

Parameters:
history_file (str): Path to the JSON file used to store history; parent directories will be created if they do not exist.
"""
self.history_file = history_file
self.logger = Logger.initialize("logs/interpreter.log")
history_dir = os.path.dirname(self.history_file)
Expand All @@ -16,6 +22,21 @@ def __init__(self, history_file: str):
json.dump([], history_file)

def save_history_json(self, task, mode, os_name, language, prompt, code_snippet, code_output, model_name):
"""
Append a structured history entry to the JSON array stored at the instance's history_file.

Builds an entry containing assistant metadata (`task`, `mode`, `os`, `language`, `model`), the user `prompt`, and system `code` and `output`, then appends it to the JSON array in the history file. If the file does not exist or is empty, a new JSON array is created containing the entry. On failure the error is logged and the original exception is re-raised.

Parameters:
task (str): High-level task or intent for the assistant.
mode (str): Mode or context identifier for the session.
os_name (str): Operating system name or target environment.
language (str): Programming or natural language associated with the entry.
prompt (str): User prompt or input text.
code_snippet (str): Code produced or executed in the session.
code_output (str): Output or result produced by the code.
model_name (str): Name of the model used by the assistant.
"""
try:
history_entry = {
"assistant": {
Expand Down Expand Up @@ -46,7 +67,15 @@ def save_history_json(self, task, mode, os_name, language, prompt, code_snippet,
raise

def _get_data_for_key(self, key: str) -> List[Any]:
"""Returns a list of all values for the specified key in the history data."""
"""
Collects all values associated with a given key from stored history entries.

Parameters:
key (str): The key to look up within each history entry (searched first in the entry's 'assistant' object, then in 'system').

Returns:
values (List[Any]): A list of values found for `key` across all history entries. Returns an empty list if the history file is missing, empty, or no entries contain the key.
"""
try:
if not os.path.exists(self.history_file):
return []
Expand All @@ -69,7 +98,17 @@ def _get_data_for_key(self, key: str) -> List[Any]:
raise

def _get_last_entries(self, count: int) -> List[dict]:
"""Returns the last n entries from the history data."""
"""
Retrieve the most recent history entries.

Returns up to `count` of the most recent history records from the history file; returns an empty list if the history file is missing or empty.

Parameters:
count (int): Maximum number of entries to return. If fewer entries exist, all available entries are returned.

Returns:
last_entries (List[dict]): A list of history entry dictionaries (up to `count`), ordered from oldest to newest within the returned slice; empty list if no entries are available.
"""
try:
if not os.path.exists(self.history_file) or os.path.getsize(self.history_file) == 0:
return []
Expand Down Expand Up @@ -99,6 +138,16 @@ def _get_last_entries_for_key(self, key: str, count: int) -> List[Any]:
raise

def _get_last_entries_for_keys(self, count: int, *keys: str) -> List[dict]:
"""
Assembles up to `count` session dictionaries where each requested key maps to its corresponding most-recent value or `None`.

Parameters:
count (int): Maximum number of sessions to return.
*keys (str): One or more history keys to include in each session.

Returns:
List[dict]: A list of up to `count` dictionaries. Each dictionary maps each requested key to the value at that position in the key's recent-values list or `None` if no value exists for that position. Returns an empty list if none of the requested keys have any entries.
"""
last_entries = []
try:
entries = {key: self._get_last_entries_for_key(key, count) for key in keys}
Expand All @@ -118,7 +167,25 @@ def get_chat_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "task", "output")

def get_code_history(self, count: int) -> List[dict]:
"""
Retrieve the most recent code sessions with their corresponding outputs.

Parameters:
count (int): Maximum number of recent sessions to return.

Returns:
List[dict]: A list of up to `count` session dictionaries where each dictionary contains the keys `"code"` and `"output"` mapped to their most recent values; missing values are `None`.
"""
return self._get_last_entries_for_keys(count, "code", "output")

def get_full_history(self, count: int) -> List[dict]:
"""
Return the most recent sessions containing task, code, and output entries.

Parameters:
count (int): Maximum number of recent sessions to include.

Returns:
history (List[dict]): A list with up to `count` session dictionaries. Each session maps the keys `"task"`, `"code"`, and `"output"` to their most recent values (or `None` if a value is missing).
"""
return self._get_last_entries_for_keys(count, "task", "code", "output")
Loading
Loading