From 2de77a3565233e3d6c66c3cd4bef441c486eb90d Mon Sep 17 00:00:00 2001 From: Bentlybro Date: Mon, 2 Jun 2025 02:13:31 +0100 Subject: [PATCH] Add-Terminal-Manager --- SimpleAgent/core/__init__.py | 8 +- SimpleAgent/core/terminal/__init__.py | 22 + SimpleAgent/core/terminal/commands.py | 719 ++++++++++++++++++ SimpleAgent/core/terminal/terminal_manager.py | 705 +++++++++++++++++ SimpleAgent/core/tool_manager.py | 16 +- requirements.txt | 3 +- 6 files changed, 1469 insertions(+), 4 deletions(-) create mode 100644 SimpleAgent/core/terminal/__init__.py create mode 100644 SimpleAgent/core/terminal/commands.py create mode 100644 SimpleAgent/core/terminal/terminal_manager.py diff --git a/SimpleAgent/core/__init__.py b/SimpleAgent/core/__init__.py index 72b500b..a17185e 100644 --- a/SimpleAgent/core/__init__.py +++ b/SimpleAgent/core/__init__.py @@ -12,6 +12,9 @@ from core.run_manager import RunManager from core.security import get_secure_path +# Terminal control system +from core.terminal import TerminalManager, TerminalSession, register_terminal_commands + __all__ = [ "SimpleAgent", "ChangeSummarizer", @@ -19,5 +22,8 @@ "ExecutionManager", "MemoryManager", "RunManager", - "get_secure_path" + "get_secure_path", + "TerminalManager", + "TerminalSession", + "register_terminal_commands" ] \ No newline at end of file diff --git a/SimpleAgent/core/terminal/__init__.py b/SimpleAgent/core/terminal/__init__.py new file mode 100644 index 0000000..67c4d09 --- /dev/null +++ b/SimpleAgent/core/terminal/__init__.py @@ -0,0 +1,22 @@ +""" +Terminal Core Module for SimpleAgent + +This module provides terminal control capabilities as a core feature. +Allows agents to execute shell commands, interact with CLI tools, and maintain +persistent terminal sessions. +""" + +from .terminal_manager import SimpleTerminalManager, SimpleTerminalSession +from .commands import register_terminal_commands + +# For backward compatibility, also export with original names +TerminalManager = SimpleTerminalManager +TerminalSession = SimpleTerminalSession + +__all__ = [ + 'SimpleTerminalManager', + 'SimpleTerminalSession', + 'TerminalManager', + 'TerminalSession', + 'register_terminal_commands' +] \ No newline at end of file diff --git a/SimpleAgent/core/terminal/commands.py b/SimpleAgent/core/terminal/commands.py new file mode 100644 index 0000000..3059d7c --- /dev/null +++ b/SimpleAgent/core/terminal/commands.py @@ -0,0 +1,719 @@ +""" +Terminal Commands for SimpleAgent - Simplified Version + +This module provides command schemas and implementations for terminal operations. +These commands are integrated into the tool system and provide a clean interface +for terminal control, including interactive session management. +""" + +from typing import Dict, Any, Optional +from .terminal_manager import get_terminal_manager, TerminalResult +import time + + +def execute_command( + command: str, + timeout: Optional[float] = 30, + background: bool = False, + session_id: Optional[str] = None +) -> str: + """ + Execute a terminal command. + + Args: + command: The command to execute + timeout: Maximum execution time in seconds (default: 30) + background: Whether to run in background (default: False) + session_id: Terminal session to use (optional, uses default if not specified) + + Returns: + Formatted result of the command execution + """ + terminal_manager = get_terminal_manager() + + try: + result = terminal_manager.execute_command( + command=command, + timeout=timeout, + background=background, + session_id=session_id + ) + + # Format the result for the agent + output_parts = [] + + if result.background: + output_parts.append(f"āœ… Command started in background (PID: {result.pid})") + output_parts.append(f"Command: {command}") + else: + output_parts.append(f"āœ… Command executed (Exit code: {result.exit_code})") + output_parts.append(f"Command: {command}") + output_parts.append(f"Execution time: {result.execution_time:.2f}s") + + if result.stdout: + output_parts.append("\nšŸ“¤ Output:") + output_parts.append(result.stdout) + + if result.stderr and result.exit_code != 0: + output_parts.append("\nāŒ Error:") + output_parts.append(result.stderr) + elif result.stderr: + output_parts.append("\nāš ļø Warning:") + output_parts.append(result.stderr) + + return "\n".join(output_parts) + + except Exception as e: + return f"āŒ Error executing command: {str(e)}" + + +def start_interactive_session( + command: str, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Start an interactive terminal session for commands with menus or prompts. + + Args: + command: The interactive command to start (e.g., 'wsl', 'firebase init') + session_id: Terminal session to use (optional) + interactive_session_name: Name for the interactive session (default: 'default') + + Returns: + Status message about starting the interactive session + """ + terminal_manager = get_terminal_manager() + + try: + success = terminal_manager.start_interactive_session( + command=command, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + if success: + return f"āœ… Started interactive session '{interactive_session_name}' with command: {command}\n" \ + f"Use 'send_to_interactive_session' to send commands and 'get_interactive_session_output' to check output." + else: + return f"āŒ Failed to start interactive session with command: {command}" + + except Exception as e: + return f"āŒ Error starting interactive session: {str(e)}" + + +def send_to_interactive_session( + input_text: str, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Send specific input to an interactive session. + + Args: + input_text: Text to send to the interactive session + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after sending the input + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.send_to_interactive_session( + input_text=input_text, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"šŸ“¤ Sent: {input_text.strip()}\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error sending input: {str(e)}" + + +def get_interactive_session_output( + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Get current output from an interactive session without sending input. + + Args: + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Current output from the interactive session + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.get_interactive_session_output( + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"šŸ“ŗ Current output:\n{output}" + + except Exception as e: + return f"āŒ Error getting session output: {str(e)}" + + +def terminate_interactive_session( + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Terminate an interactive session. + + Args: + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Status message about termination + """ + terminal_manager = get_terminal_manager() + + try: + result = terminal_manager.terminate_interactive_session( + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return result + + except Exception as e: + return f"āŒ Error terminating interactive session: {str(e)}" + + +def list_interactive_sessions(session_id: Optional[str] = None) -> str: + """ + List all interactive sessions in a terminal session. + + Args: + session_id: Terminal session to check (optional) + + Returns: + List of interactive sessions + """ + terminal_manager = get_terminal_manager() + + try: + session = terminal_manager.get_session(session_id) + sessions = session.list_interactive_sessions() + + if not sessions: + return "šŸ“‹ No active interactive sessions" + + output_parts = ["šŸ“‹ Active interactive sessions:"] + for sess in sessions: + status = "šŸŸ¢ Active" if sess['alive'] else "šŸ”“ Inactive" + output_parts.append( + f" ā€¢ {sess['name']}: {status} " + f"({sess['commands']} commands executed)" + ) + + return "\n".join(output_parts) + + except Exception as e: + return f"āŒ Error listing interactive sessions: {str(e)}" + + +def list_sessions() -> str: + """ + List all active terminal sessions. + + Returns: + List of active terminal sessions + """ + terminal_manager = get_terminal_manager() + + try: + sessions = terminal_manager.list_sessions() + + if not sessions: + return "šŸ“‹ No active terminal sessions" + + output_parts = ["šŸ“‹ Active terminal sessions:"] + for session_id in sessions: + session = terminal_manager.get_session(session_id) + interactive_count = len(session.interactive_sessions) + output_parts.append( + f" ā€¢ {session_id} (Working dir: {session.working_dir}, " + f"Interactive sessions: {interactive_count})" + ) + + return "\n".join(output_parts) + + except Exception as e: + return f"āŒ Error listing sessions: {str(e)}" + + +def send_key_to_interactive_session( + key: str, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Send special keys like arrow keys, Enter, etc. to an interactive session. + + Args: + key: Special key to send ('up', 'down', 'left', 'right', 'enter', 'tab', 'escape', 'space', 'ctrl+c') + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after sending the key + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.send_key_to_interactive_session( + key=key, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"šŸ”‘ Sent key: {key}\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error sending key: {str(e)}" + + +def send_arrow_to_interactive_session( + direction: str, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Send arrow key to navigate menus in interactive session. + + Args: + direction: Arrow direction ('up', 'down', 'left', 'right') + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after sending the arrow key + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.send_arrow_to_interactive_session( + direction=direction, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"ā¬†ļøā¬‡ļøā¬…ļøāž”ļø Sent arrow: {direction}\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error sending arrow key: {str(e)}" + + +def send_enter_to_interactive_session( + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Send Enter key to confirm selection in interactive session. + + Args: + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after sending Enter + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.send_enter_to_interactive_session( + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"ā†©ļø Sent Enter key\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error sending Enter key: {str(e)}" + + +def navigate_interactive_session_menu( + option_index: int, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Navigate to a specific menu option by index (0-based) in interactive session. + + Args: + option_index: Index of the menu option to navigate to (0-based) + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after navigation + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.navigate_interactive_session_menu( + option_index=option_index, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"šŸ§­ Navigated to menu option {option_index}\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error navigating menu: {str(e)}" + + +def select_interactive_session_menu_option( + option_index: int, + session_id: Optional[str] = None, + interactive_session_name: str = "default" +) -> str: + """ + Navigate to and select a specific menu option by index (0-based) in interactive session. + + Args: + option_index: Index of the menu option to select (0-based) + session_id: Terminal session to use (optional) + interactive_session_name: Name of the interactive session (default: 'default') + + Returns: + Output received after selection + """ + terminal_manager = get_terminal_manager() + + try: + output = terminal_manager.select_interactive_session_menu_option( + option_index=option_index, + session_id=session_id, + interactive_session_name=interactive_session_name + ) + + return f"āœ… Selected menu option {option_index}\nšŸ“„ Output:\n{output}" + + except Exception as e: + return f"āŒ Error selecting menu option: {str(e)}" + + +# Simplified command schemas for the tool system +EXECUTE_COMMAND_SCHEMA = { + "type": "function", + "function": { + "name": "execute_command", + "description": "Execute a terminal/shell command. Supports background processes and persistent sessions.", + "parameters": { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "The command to execute (e.g., 'ls -la', 'git status', 'npm install')" + }, + "timeout": { + "type": "number", + "description": "Maximum execution time in seconds (default: 30)", + "default": 30 + }, + "background": { + "type": "boolean", + "description": "Whether to run the command in background (default: false)", + "default": False + }, + "session_id": { + "type": "string", + "description": "Terminal session ID to use (optional, uses default session if not specified)" + } + }, + "required": ["command"] + } + } +} + +START_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "start_interactive_session", + "description": "Start an interactive terminal session for commands that need ongoing interaction.", + "parameters": { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "The interactive command to start (e.g., 'wsl', 'firebase init', 'python')" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID to use (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name for this interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["command"] + } + } +} + +SEND_TO_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "send_to_interactive_session", + "description": "Send input to an active interactive session.", + "parameters": { + "type": "object", + "properties": { + "input_text": { + "type": "string", + "description": "Text to send to the interactive session" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["input_text"] + } + } +} + +GET_INTERACTIVE_SESSION_OUTPUT_SCHEMA = { + "type": "function", + "function": { + "name": "get_interactive_session_output", + "description": "Get current output from an interactive session without sending any input.", + "parameters": { + "type": "object", + "properties": { + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": [] + } + } +} + +TERMINATE_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "terminate_interactive_session", + "description": "Terminate an interactive session when you're done with it.", + "parameters": { + "type": "object", + "properties": { + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": [] + } + } +} + +LIST_INTERACTIVE_SESSIONS_SCHEMA = { + "type": "function", + "function": { + "name": "list_interactive_sessions", + "description": "List all active interactive sessions.", + "parameters": { + "type": "object", + "properties": { + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + } + }, + "required": [] + } + } +} + +LIST_SESSIONS_SCHEMA = { + "type": "function", + "function": { + "name": "list_sessions", + "description": "List all active terminal sessions", + "parameters": { + "type": "object", + "properties": {}, + "required": [] + } + } +} + +SEND_KEY_TO_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "send_key_to_interactive_session", + "description": "Send special keys like arrow keys, Enter, etc. to navigate interactive CLI menus.", + "parameters": { + "type": "object", + "properties": { + "key": { + "type": "string", + "description": "Special key to send ('up', 'down', 'left', 'right', 'enter', 'tab', 'escape', 'space', 'ctrl+c')" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["key"] + } + } +} + +SEND_ARROW_TO_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "send_arrow_to_interactive_session", + "description": "Send arrow key to navigate menus in interactive CLI applications.", + "parameters": { + "type": "object", + "properties": { + "direction": { + "type": "string", + "description": "Arrow direction ('up', 'down', 'left', 'right')" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["direction"] + } + } +} + +SEND_ENTER_TO_INTERACTIVE_SESSION_SCHEMA = { + "type": "function", + "function": { + "name": "send_enter_to_interactive_session", + "description": "Send Enter key to confirm selection in interactive CLI menu.", + "parameters": { + "type": "object", + "properties": { + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": [] + } + } +} + +NAVIGATE_INTERACTIVE_SESSION_MENU_SCHEMA = { + "type": "function", + "function": { + "name": "navigate_interactive_session_menu", + "description": "Navigate to a specific menu option by index (0-based) without selecting it.", + "parameters": { + "type": "object", + "properties": { + "option_index": { + "type": "integer", + "description": "Index of the menu option to navigate to (0-based)" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["option_index"] + } + } +} + +SELECT_INTERACTIVE_SESSION_MENU_OPTION_SCHEMA = { + "type": "function", + "function": { + "name": "select_interactive_session_menu_option", + "description": "Navigate to and select a specific menu option by index (0-based) in one operation.", + "parameters": { + "type": "object", + "properties": { + "option_index": { + "type": "integer", + "description": "Index of the menu option to select (0-based)" + }, + "session_id": { + "type": "string", + "description": "Terminal session ID (optional)" + }, + "interactive_session_name": { + "type": "string", + "description": "Name of the interactive session (default: 'default')", + "default": "default" + } + }, + "required": ["option_index"] + } + } +} + + +def register_terminal_commands(): + """Register core terminal commands with the tool system.""" + from core.tool_manager import register_command + + # Register simplified terminal commands + register_command("execute_command", execute_command, EXECUTE_COMMAND_SCHEMA) + register_command("start_interactive_session", start_interactive_session, START_INTERACTIVE_SESSION_SCHEMA) + register_command("send_to_interactive_session", send_to_interactive_session, SEND_TO_INTERACTIVE_SESSION_SCHEMA) + register_command("get_interactive_session_output", get_interactive_session_output, GET_INTERACTIVE_SESSION_OUTPUT_SCHEMA) + register_command("terminate_interactive_session", terminate_interactive_session, TERMINATE_INTERACTIVE_SESSION_SCHEMA) + register_command("list_interactive_sessions", list_interactive_sessions, LIST_INTERACTIVE_SESSIONS_SCHEMA) + register_command("list_sessions", list_sessions, LIST_SESSIONS_SCHEMA) + register_command("send_key_to_interactive_session", send_key_to_interactive_session, SEND_KEY_TO_INTERACTIVE_SESSION_SCHEMA) + register_command("send_arrow_to_interactive_session", send_arrow_to_interactive_session, SEND_ARROW_TO_INTERACTIVE_SESSION_SCHEMA) + register_command("send_enter_to_interactive_session", send_enter_to_interactive_session, SEND_ENTER_TO_INTERACTIVE_SESSION_SCHEMA) + register_command("navigate_interactive_session_menu", navigate_interactive_session_menu, NAVIGATE_INTERACTIVE_SESSION_MENU_SCHEMA) + register_command("select_interactive_session_menu_option", select_interactive_session_menu_option, SELECT_INTERACTIVE_SESSION_MENU_OPTION_SCHEMA) \ No newline at end of file diff --git a/SimpleAgent/core/terminal/terminal_manager.py b/SimpleAgent/core/terminal/terminal_manager.py new file mode 100644 index 0000000..ba63a20 --- /dev/null +++ b/SimpleAgent/core/terminal/terminal_manager.py @@ -0,0 +1,705 @@ +""" +Terminal Manager for SimpleAgent - Simplified Version + +Provides basic terminal control capabilities: +- Simple command execution +- Basic interactive session support +- Working directory management +- Clean, reliable I/O +""" + +import subprocess +import time +import os +import signal +import psutil +from typing import Dict, Any, List, Optional, Tuple +from dataclasses import dataclass +import platform +import threading + + +@dataclass +class TerminalResult: + """Result of a terminal command execution.""" + stdout: str + stderr: str + exit_code: int + execution_time: float + pid: Optional[int] = None + background: bool = False + + +class SimpleInteractiveSession: + """ + Simplified interactive terminal session - focuses on what actually works. + """ + + def __init__(self, session_id: str, working_dir: str = None): + """Initialize a simple interactive session.""" + self.session_id = session_id + self.working_dir = working_dir or os.getcwd() + self.environment = dict(os.environ) + + self.process: Optional[subprocess.Popen] = None + self.is_running = False + self.command_history: List[str] = [] + + # Simple output collection + self.output_lines = [] + self.output_lock = threading.Lock() + self.output_thread = None + + def start_process(self, command: str) -> bool: + """Start a process with clean, simple settings.""" + if self.is_running: + return False + + try: + self.command_history.append(command) + + if platform.system() == "Windows": + if command.strip().lower() == 'wsl': + # WSL with binary mode to have complete control over line endings + self.process = subprocess.Popen( + ['wsl.exe', '--exec', 'bash', '--norc', '--noprofile'], # Minimal bash setup + cwd=self.working_dir, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=False, # Binary mode + bufsize=0, # Unbuffered + env=dict(os.environ, TERM='dumb', LANG='C.UTF-8') # Clean environment + ) + else: + # Simple Windows command + self.process = subprocess.Popen( + command, + shell=True, + cwd=self.working_dir, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=0 + ) + else: + # Simple Unix command + self.process = subprocess.Popen( + command, + shell=True, + cwd=self.working_dir, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=0 + ) + + self.is_running = True + + # Start simple output collection + self.output_thread = threading.Thread(target=self._collect_output, daemon=True) + self.output_thread.start() + + # Wait for startup and send initial command to test + time.sleep(2) + + # Send a test command to see if we get any response + if command.strip().lower() == 'wsl': + # Send with explicit UTF-8 encoding and Unix line ending only + test_cmd = 'echo "WSL_READY"\n'.encode('utf-8') + self.process.stdin.write(test_cmd) + self.process.stdin.flush() + time.sleep(1) + + return True + + except Exception as e: + print(f"āŒ Failed to start: {e}") + return False + + def _collect_output(self): + """Simple output collection in background.""" + + # Check if this is a WSL process (binary mode) + is_wsl_binary = (hasattr(self.process, 'stdin') and + hasattr(self.process.stdin, 'mode') and + 'b' in str(type(self.process.stdin))) + + if not is_wsl_binary: + # Detect binary mode by checking if stdout has a mode attribute + try: + is_wsl_binary = not hasattr(self.process.stdout, 'encoding') + except: + is_wsl_binary = False + + while self.is_running and self.process: + try: + # Check if process is still alive + poll_result = self.process.poll() + if poll_result is not None: + break + + # Try to read with a timeout approach + try: + if is_wsl_binary: + # Binary mode - read bytes and decode as UTF-8 + line_bytes = self.process.stdout.readline() + if line_bytes: + line = line_bytes.decode('utf-8', errors='replace') + with self.output_lock: + self.output_lines.append(line.rstrip('\n\r')) + else: + time.sleep(0.1) + else: + # Text mode - read normally + line = self.process.stdout.readline() + if line: + with self.output_lock: + self.output_lines.append(line.rstrip('\n\r')) + else: + time.sleep(0.1) + + except Exception as e: + time.sleep(0.1) + + except Exception as e: + break + + def send_command(self, command: str) -> bool: + """Send a command with proper line ending.""" + if not self.is_running or not self.process: + return False + + try: + # Clean command completely + clean_command = command.strip() + # Remove ALL carriage returns and normalize line endings + clean_command = clean_command.replace('\r\n', '').replace('\r', '').replace('\n', '') + + # Check if this is binary mode (WSL) + is_wsl_binary = False + try: + is_wsl_binary = not hasattr(self.process.stdout, 'encoding') + except: + is_wsl_binary = False + + if is_wsl_binary: + # Binary mode - encode as UTF-8 with only Unix newline + command_bytes = (clean_command + '\n').encode('utf-8') + self.process.stdin.write(command_bytes) + else: + # Text mode - add only Unix newline + clean_command = clean_command + '\n' + self.process.stdin.write(clean_command) + + self.process.stdin.flush() + return True + + except Exception as e: + print(f"āŒ Send error: {e}") + return False + + def send_key(self, key: str) -> bool: + """Send special keyboard keys like arrow keys, Enter, etc.""" + if not self.is_running or not self.process: + return False + + try: + # Check if this is binary mode (WSL) + is_wsl_binary = False + try: + is_wsl_binary = not hasattr(self.process.stdout, 'encoding') + except: + is_wsl_binary = False + + # Map special keys to their ANSI escape sequences + key_mappings = { + 'up': '\x1b[A', # Up arrow + 'down': '\x1b[B', # Down arrow + 'right': '\x1b[C', # Right arrow + 'left': '\x1b[D', # Left arrow + 'enter': '\r', # Enter key + 'tab': '\t', # Tab key + 'escape': '\x1b', # Escape key + 'space': ' ', # Space key + 'ctrl+c': '\x03', # Ctrl+C + } + + # Get the key sequence + key_sequence = key_mappings.get(key.lower(), key) + + if is_wsl_binary: + # Binary mode - encode as UTF-8 + key_bytes = key_sequence.encode('utf-8') + self.process.stdin.write(key_bytes) + else: + # Text mode + self.process.stdin.write(key_sequence) + + self.process.stdin.flush() + return True + + except Exception as e: + print(f"āŒ Key send error: {e}") + return False + + def send_arrow_key(self, direction: str) -> bool: + """Send arrow key in specified direction.""" + return self.send_key(direction) + + def send_enter(self) -> bool: + """Send Enter key.""" + return self.send_key('enter') + + def navigate_menu(self, option_index: int, total_options: int = None) -> bool: + """Navigate to a specific menu option by index (0-based).""" + if not self.is_running or not self.process: + return False + + # Send arrow keys to navigate to the desired option + # Assuming we start at option 0, navigate down to reach the target + for i in range(option_index): + if not self.send_arrow_key('down'): + return False + time.sleep(0.1) # Small delay between key presses + + return True + + def select_menu_option(self, option_index: int, total_options: int = None) -> bool: + """Navigate to and select a menu option.""" + if self.navigate_menu(option_index, total_options): + time.sleep(0.2) # Brief pause before selecting + return self.send_enter() + return False + + def get_output(self, clear: bool = True) -> str: + """Get collected output.""" + with self.output_lock: + output = '\n'.join(self.output_lines) + if clear: + self.output_lines.clear() + return output + + def get_recent_output(self, lines: int = 10) -> str: + """Get recent output lines.""" + with self.output_lock: + recent = self.output_lines[-lines:] if self.output_lines else [] + return '\n'.join(recent) + + def wait_for_output(self, timeout: float = 5.0) -> str: + """Wait for new output to appear.""" + start_time = time.time() + initial_count = len(self.output_lines) + + while time.time() - start_time < timeout: + with self.output_lock: + if len(self.output_lines) > initial_count: + break + time.sleep(0.1) + + return self.get_output(clear=False) + + def is_alive(self) -> bool: + """Check if process is alive.""" + return self.is_running and self.process and self.process.poll() is None + + def terminate(self) -> bool: + """Terminate the session.""" + if not self.is_running: + return True + + self.is_running = False + + if self.process: + try: + self.process.terminate() + self.process.wait(timeout=3) + except: + try: + self.process.kill() + self.process.wait() + except: + pass + + return True + + +class SimpleTerminalSession: + """ + Simplified terminal session management. + """ + + def __init__(self, session_id: str, working_dir: str = None): + """Initialize terminal session.""" + self.session_id = session_id + self.working_dir = working_dir or os.getcwd() + self.environment = dict(os.environ) + self.interactive_sessions: Dict[str, SimpleInteractiveSession] = {} + self.is_active = True + + def execute_command( + self, + command: str, + timeout: Optional[float] = 30, + background: bool = False + ) -> TerminalResult: + """Execute a simple command.""" + start_time = time.time() + + try: + if background: + # Background process + process = subprocess.Popen( + command, + shell=True, + cwd=self.working_dir, + env=self.environment + ) + + return TerminalResult( + stdout=f"Background process started (PID: {process.pid})", + stderr="", + exit_code=0, + execution_time=time.time() - start_time, + pid=process.pid, + background=True + ) + else: + # Foreground process + result = subprocess.run( + command, + shell=True, + cwd=self.working_dir, + env=self.environment, + capture_output=True, + text=True, + timeout=timeout + ) + + return TerminalResult( + stdout=result.stdout or "", + stderr=result.stderr or "", + exit_code=result.returncode, + execution_time=time.time() - start_time + ) + + except subprocess.TimeoutExpired as e: + return TerminalResult( + stdout=e.stdout or "", + stderr=f"Command timed out after {timeout}s", + exit_code=-1, + execution_time=time.time() - start_time + ) + except Exception as e: + return TerminalResult( + stdout="", + stderr=f"Error: {str(e)}", + exit_code=-1, + execution_time=time.time() - start_time + ) + + def start_interactive_session(self, command: str, session_name: str = "default") -> bool: + """Start an interactive session.""" + if session_name in self.interactive_sessions: + self.interactive_sessions[session_name].terminate() + + session = SimpleInteractiveSession(session_name, self.working_dir) + session.environment = self.environment.copy() + + if session.start_process(command): + self.interactive_sessions[session_name] = session + return True + + return False + + def send_to_session(self, input_text: str, session_name: str = "default") -> str: + """Send input to interactive session.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + if not session.is_alive(): + return "āŒ Session not active" + + # AGGRESSIVE input cleaning - remove ALL carriage returns and line endings + clean_input = input_text.strip() + # Remove all possible line ending variations + clean_input = clean_input.replace('\r\n', '').replace('\r', '').replace('\n', '') + + # Clear old output first + session.get_output(clear=True) + + # Send command + success = session.send_command(clean_input) + if not success: + return "āŒ Failed to send input" + + # Wait for output + time.sleep(1.5) + + # Get new output + output = session.get_output(clear=False) + return output + + def send_key_to_session(self, key: str, session_name: str = "default") -> str: + """Send special key to interactive session.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + if not session.is_alive(): + return "āŒ Session not active" + + # Clear old output first + session.get_output(clear=True) + + # Send key + success = session.send_key(key) + if not success: + return "āŒ Failed to send key" + + # Wait for output + time.sleep(1.0) + + # Get new output + output = session.get_output(clear=False) + return output + + def send_arrow_to_session(self, direction: str, session_name: str = "default") -> str: + """Send arrow key to interactive session.""" + return self.send_key_to_session(direction, session_name) + + def send_enter_to_session(self, session_name: str = "default") -> str: + """Send Enter key to interactive session.""" + return self.send_key_to_session('enter', session_name) + + def navigate_session_menu(self, option_index: int, session_name: str = "default") -> str: + """Navigate to specific menu option in interactive session.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + if not session.is_alive(): + return "āŒ Session not active" + + # Clear old output first + session.get_output(clear=True) + + # Navigate to option + success = session.navigate_menu(option_index) + if not success: + return "āŒ Failed to navigate menu" + + # Wait for output + time.sleep(1.0) + + # Get new output + output = session.get_output(clear=False) + return output + + def select_session_menu_option(self, option_index: int, session_name: str = "default") -> str: + """Navigate to and select menu option in interactive session.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + if not session.is_alive(): + return "āŒ Session not active" + + # Clear old output first + session.get_output(clear=True) + + # Navigate and select option + success = session.select_menu_option(option_index) + if not success: + return "āŒ Failed to select menu option" + + # Wait for output + time.sleep(1.5) + + # Get new output + output = session.get_output(clear=False) + return output + + def get_session_output(self, session_name: str = "default") -> str: + """Get current session output.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + return session.get_output(clear=False) + + def terminate_interactive_session(self, session_name: str = "default") -> str: + """Terminate interactive session.""" + if session_name not in self.interactive_sessions: + return "āŒ No session found" + + session = self.interactive_sessions[session_name] + session.terminate() + del self.interactive_sessions[session_name] + return f"āœ… Session '{session_name}' terminated" + + def list_interactive_sessions(self) -> List[Dict[str, Any]]: + """List interactive sessions.""" + sessions = [] + for name, session in self.interactive_sessions.items(): + sessions.append({ + "name": name, + "alive": session.is_alive(), + "commands": len(session.command_history) + }) + return sessions + + +class SimpleTerminalManager: + """ + Simplified terminal manager - focuses on core functionality. + """ + + def __init__(self): + """Initialize the manager.""" + self.sessions: Dict[str, SimpleTerminalSession] = {} + self.default_session_id = "main" + + # Create default session + self.create_session(self.default_session_id) + + def create_session(self, session_id: str, working_dir: str = None) -> SimpleTerminalSession: + """Create a new session.""" + if session_id in self.sessions: + raise ValueError(f"Session {session_id} already exists") + + session = SimpleTerminalSession(session_id, working_dir) + self.sessions[session_id] = session + return session + + def get_session(self, session_id: str = None) -> SimpleTerminalSession: + """Get a session.""" + session_id = session_id or self.default_session_id + if session_id not in self.sessions: + return self.create_session(session_id) + return self.sessions[session_id] + + def execute_command( + self, + command: str, + session_id: str = None, + timeout: Optional[float] = 30, + background: bool = False + ) -> TerminalResult: + """Execute a command.""" + session = self.get_session(session_id) + return session.execute_command(command, timeout, background) + + def start_interactive_session( + self, + command: str, + session_id: str = None, + interactive_session_name: str = "default" + ) -> bool: + """Start interactive session.""" + session = self.get_session(session_id) + return session.start_interactive_session(command, interactive_session_name) + + def send_to_interactive_session( + self, + input_text: str, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Send to interactive session.""" + session = self.get_session(session_id) + return session.send_to_session(input_text, interactive_session_name) + + def send_key_to_interactive_session( + self, + key: str, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Send special key to interactive session.""" + session = self.get_session(session_id) + return session.send_key_to_session(key, interactive_session_name) + + def send_arrow_to_interactive_session( + self, + direction: str, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Send arrow key to interactive session.""" + session = self.get_session(session_id) + return session.send_arrow_to_session(direction, interactive_session_name) + + def send_enter_to_interactive_session( + self, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Send Enter key to interactive session.""" + session = self.get_session(session_id) + return session.send_enter_to_session(interactive_session_name) + + def navigate_interactive_session_menu( + self, + option_index: int, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Navigate to specific menu option in interactive session.""" + session = self.get_session(session_id) + return session.navigate_session_menu(option_index, interactive_session_name) + + def select_interactive_session_menu_option( + self, + option_index: int, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Navigate to and select menu option in interactive session.""" + session = self.get_session(session_id) + return session.select_session_menu_option(option_index, interactive_session_name) + + def get_interactive_session_output( + self, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Get session output.""" + session = self.get_session(session_id) + return session.get_session_output(interactive_session_name) + + def terminate_interactive_session( + self, + session_id: str = None, + interactive_session_name: str = "default" + ) -> str: + """Terminate session.""" + session = self.get_session(session_id) + return session.terminate_interactive_session(interactive_session_name) + + def list_sessions(self) -> List[str]: + """List all sessions.""" + return list(self.sessions.keys()) + + def cleanup_all(self): + """Clean up all sessions.""" + for session_id in list(self.sessions.keys()): + session = self.sessions[session_id] + for interactive_session in list(session.interactive_sessions.values()): + interactive_session.terminate() + del self.sessions[session_id] + + +# Global instance +_terminal_manager = None + + +def get_terminal_manager() -> SimpleTerminalManager: + """Get the global terminal manager instance.""" + global _terminal_manager + if _terminal_manager is None: + _terminal_manager = SimpleTerminalManager() + return _terminal_manager \ No newline at end of file diff --git a/SimpleAgent/core/tool_manager.py b/SimpleAgent/core/tool_manager.py index 4cb8a9d..de4bb70 100644 --- a/SimpleAgent/core/tool_manager.py +++ b/SimpleAgent/core/tool_manager.py @@ -752,7 +752,9 @@ def print_available_tools(self) -> None: if local_tools or remote_tools: # Determine category display name - if category == 'file_ops': + if category == 'terminal': + display_name = 'šŸ–„ļø Terminal Control' + elif category == 'file_ops': display_name = 'šŸ“ File Operations' elif category == 'github_ops': display_name = 'šŸ™ GitHub Operations' @@ -818,7 +820,9 @@ def print_commands(self) -> None: for category, commands in COMMANDS_BY_CATEGORY.items(): for cmd in commands: # Determine actual category from command name - if any(x in cmd for x in ['file', 'read', 'write', 'edit', 'delete', 'create_directory', 'list_directory', 'load_json', 'save_json', 'append']): + if any(x in cmd for x in ['execute_command', 'change_directory', 'set_environment_variable', 'list_sessions', 'list_active_processes', 'terminate_process', 'get_session_info']): + actual_categories['šŸ–„ļø Terminal Control'].append(cmd) + elif any(x in cmd for x in ['file', 'read', 'write', 'edit', 'delete', 'create_directory', 'list_directory', 'load_json', 'save_json', 'append']): actual_categories['šŸ“ File Operations'].append(cmd) elif any(x in cmd for x in ['github', 'git_', 'pr_', 'issue_']): actual_categories['šŸ™ GitHub Operations'].append(cmd) @@ -885,6 +889,14 @@ def init(dynamic: bool = True) -> None: """ tool_manager = get_tool_manager() + # Always register terminal commands first as they are core features + try: + from core.terminal import register_terminal_commands + register_terminal_commands() + tool_manager.logger.info("āœ… Terminal control commands registered") + except ImportError as e: + tool_manager.logger.warning(f"āš ļø Failed to register terminal commands: {e}") + if dynamic: tool_manager.initialize_dynamic_tools() tool_manager.print_available_tools() diff --git a/requirements.txt b/requirements.txt index e9b6bda..8c6511b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,4 +8,5 @@ googlesearch-python PyGithub discord.py pyautogui -google-genai \ No newline at end of file +google-genai +psutil \ No newline at end of file