#!/usr/bin/env python3 """ AIVA Queen Terminal Control System =================================== Orchestration Module 01: Terminal Control This module provides AIVA Queen with the ability to open, manage, and control multiple terminal sessions including local shells and remote SSH connections. Components: - TerminalManager: Core manager for multiple terminal sessions - SSHConnector: SSH connections to remote servers (Elestio infrastructure) - CommandExecutor: Execute shell commands with proper handling - OutputParser: Parse and structure command output - SessionPool: Pool of ready-to-use sessions for rapid execution - ErrorHandler: Comprehensive error handling and recovery Author: AIVA Queen Elevation Sprint Version: 1.0.0 Created: 2026-01-12 """ import os import sys import json import time import uuid import queue import signal import socket import logging import hashlib import asyncio import secrets import threading import subprocess import traceback from abc import ABC, abstractmethod from enum import Enum, auto from pathlib import Path from typing import ( Dict, List, Optional, Tuple, Any, Union, Callable, TypeVar, Generic, Set, AsyncGenerator ) from dataclasses import dataclass, field from datetime import datetime, timedelta from contextlib import contextmanager, asynccontextmanager from concurrent.futures import ThreadPoolExecutor, Future from collections import defaultdict # Conditional imports for SSH try: import paramiko from paramiko import SSHClient, AutoAddPolicy, RSAKey, SSHException PARAMIKO_AVAILABLE = True except ImportError: PARAMIKO_AVAILABLE = False SSHClient = None AutoAddPolicy = None # Conditional imports for async SSH try: import asyncssh ASYNCSSH_AVAILABLE = True except ImportError: ASYNCSSH_AVAILABLE = False # ============================================================================= # CONFIGURATION AND CONSTANTS # ============================================================================= class TerminalConfig: """Central configuration for terminal control system""" # Session limits MAX_SESSIONS = 50 MAX_SSH_SESSIONS = 20 MAX_LOCAL_SESSIONS = 30 SESSION_TIMEOUT = 3600 # 1 hour COMMAND_TIMEOUT = 300 # 5 minutes default # Pool settings POOL_MIN_SIZE = 5 POOL_MAX_SIZE = 20 POOL_REPLENISH_THRESHOLD = 3 # Retry settings MAX_RETRIES = 3 RETRY_DELAY = 1.0 RETRY_BACKOFF = 2.0 # Buffer sizes OUTPUT_BUFFER_SIZE = 65536 MAX_OUTPUT_HISTORY = 1000 # Logging LOG_LEVEL = logging.INFO LOG_FORMAT = "[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s" # SSH defaults for Elestio DEFAULT_SSH_PORT = 22 SSH_KEY_PATH = os.path.expanduser("~/.ssh/genesis_mother_key") SSH_KNOWN_HOSTS = os.path.expanduser("~/.ssh/known_hosts") SSH_CONNECT_TIMEOUT = 30 SSH_BANNER_TIMEOUT = 60 # Elestio infrastructure endpoints ELESTIO_ENDPOINTS = { "redis": { "host": "redis-genesis-u50607.vm.elestio.app", "port": 26379, "user": "root" }, "postgresql": { "host": "postgresql-genesis-u50607.vm.elestio.app", "port": 5432, "user": "postgres" }, "ollama": { "host": "152.53.201.152", "port": 22, "user": "root" }, "qdrant": { "host": "qdrant-b3knu-u50607.vm.elestio.app", "port": 6333, "user": "root" } } # Configure logging logging.basicConfig( level=TerminalConfig.LOG_LEVEL, format=TerminalConfig.LOG_FORMAT ) logger = logging.getLogger("AIVA.TerminalControl") # ============================================================================= # ENUMS AND DATA CLASSES # ============================================================================= class SessionState(Enum): """Terminal session states""" INITIALIZING = auto() READY = auto() BUSY = auto() ERROR = auto() CLOSED = auto() RECONNECTING = auto() class SessionType(Enum): """Types of terminal sessions""" LOCAL_SHELL = auto() SSH_INTERACTIVE = auto() SSH_EXEC = auto() PTY = auto() class CommandPriority(Enum): """Command execution priority levels""" CRITICAL = 1 HIGH = 2 NORMAL = 3 LOW = 4 BACKGROUND = 5 class ErrorSeverity(Enum): """Error severity levels for handling""" DEBUG = auto() INFO = auto() WARNING = auto() ERROR = auto() CRITICAL = auto() FATAL = auto() @dataclass class CommandResult: """Result of a command execution""" command: str exit_code: int stdout: str stderr: str duration: float session_id: str timestamp: datetime = field(default_factory=datetime.now) metadata: Dict[str, Any] = field(default_factory=dict) @property def success(self) -> bool: return self.exit_code == 0 @property def output(self) -> str: """Combined output with preference for stdout""" return self.stdout if self.stdout else self.stderr def to_dict(self) -> Dict[str, Any]: return { "command": self.command, "exit_code": self.exit_code, "stdout": self.stdout, "stderr": self.stderr, "duration": self.duration, "session_id": self.session_id, "timestamp": self.timestamp.isoformat(), "success": self.success, "metadata": self.metadata } @dataclass class SessionInfo: """Information about a terminal session""" session_id: str session_type: SessionType state: SessionState created_at: datetime last_active: datetime host: Optional[str] = None port: Optional[int] = None user: Optional[str] = None command_count: int = 0 error_count: int = 0 working_directory: str = field(default_factory=os.getcwd) environment: Dict[str, str] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "session_id": self.session_id, "session_type": self.session_type.name, "state": self.state.name, "created_at": self.created_at.isoformat(), "last_active": self.last_active.isoformat(), "host": self.host, "port": self.port, "user": self.user, "command_count": self.command_count, "error_count": self.error_count, "working_directory": self.working_directory, "metadata": self.metadata } @dataclass class TerminalError: """Structured error information""" error_id: str severity: ErrorSeverity message: str exception: Optional[Exception] = None session_id: Optional[str] = None command: Optional[str] = None timestamp: datetime = field(default_factory=datetime.now) traceback: Optional[str] = None recovery_action: Optional[str] = None context: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "error_id": self.error_id, "severity": self.severity.name, "message": self.message, "exception_type": type(self.exception).__name__ if self.exception else None, "session_id": self.session_id, "command": self.command, "timestamp": self.timestamp.isoformat(), "traceback": self.traceback, "recovery_action": self.recovery_action, "context": self.context } # ============================================================================= # ERROR HANDLER # ============================================================================= class ErrorHandler: """ Comprehensive error handling for terminal operations. Provides: - Error classification and severity assessment - Recovery strategies - Error logging and tracking - Alert generation for critical errors """ def __init__(self): self._errors: List[TerminalError] = [] self._error_counts: Dict[str, int] = defaultdict(int) self._recovery_callbacks: Dict[str, Callable] = {} self._alert_callbacks: List[Callable[[TerminalError], None]] = [] self._lock = threading.Lock() # Error pattern classifications self._error_patterns = { "connection_refused": { "patterns": ["connection refused", "connect refused", "ECONNREFUSED"], "severity": ErrorSeverity.ERROR, "recovery": "retry_connection" }, "timeout": { "patterns": ["timed out", "timeout", "ETIMEDOUT"], "severity": ErrorSeverity.WARNING, "recovery": "increase_timeout" }, "authentication": { "patterns": ["authentication failed", "permission denied", "auth fail"], "severity": ErrorSeverity.CRITICAL, "recovery": "check_credentials" }, "host_unreachable": { "patterns": ["host unreachable", "no route to host", "network is unreachable"], "severity": ErrorSeverity.ERROR, "recovery": "check_network" }, "session_closed": { "patterns": ["session closed", "connection closed", "broken pipe"], "severity": ErrorSeverity.WARNING, "recovery": "reconnect_session" }, "command_not_found": { "patterns": ["command not found", "not recognized", "not found"], "severity": ErrorSeverity.INFO, "recovery": None }, "permission_denied": { "patterns": ["permission denied", "access denied", "not permitted"], "severity": ErrorSeverity.ERROR, "recovery": "elevate_privileges" }, "resource_exhausted": { "patterns": ["too many open files", "resource temporarily unavailable", "out of memory"], "severity": ErrorSeverity.CRITICAL, "recovery": "cleanup_resources" } } def register_recovery_callback(self, recovery_type: str, callback: Callable): """Register a callback for a specific recovery action""" self._recovery_callbacks[recovery_type] = callback def register_alert_callback(self, callback: Callable[[TerminalError], None]): """Register a callback for critical error alerts""" self._alert_callbacks.append(callback) def classify_error(self, error: Union[str, Exception]) -> Tuple[str, ErrorSeverity, Optional[str]]: """Classify an error and determine severity and recovery action""" error_str = str(error).lower() for error_type, config in self._error_patterns.items(): for pattern in config["patterns"]: if pattern.lower() in error_str: return error_type, config["severity"], config["recovery"] # Default classification return "unknown", ErrorSeverity.ERROR, None def handle_error( self, exception: Optional[Exception] = None, message: str = "", session_id: Optional[str] = None, command: Optional[str] = None, context: Optional[Dict[str, Any]] = None ) -> TerminalError: """ Handle an error with classification, logging, and optional recovery. Returns the structured TerminalError for further handling. """ error_message = message or str(exception) if exception else "Unknown error" error_type, severity, recovery_action = self.classify_error(error_message) error = TerminalError( error_id=f"ERR-{uuid.uuid4().hex[:8].upper()}", severity=severity, message=error_message, exception=exception, session_id=session_id, command=command, traceback=traceback.format_exc() if exception else None, recovery_action=recovery_action, context=context or {} ) with self._lock: self._errors.append(error) self._error_counts[error_type] += 1 # Trim error history if needed if len(self._errors) > 1000: self._errors = self._errors[-500:] # Log the error log_method = getattr(logger, severity.name.lower(), logger.error) log_method(f"[{error.error_id}] {error_type}: {error_message}") # Alert for critical/fatal errors if severity in (ErrorSeverity.CRITICAL, ErrorSeverity.FATAL): for callback in self._alert_callbacks: try: callback(error) except Exception as e: logger.error(f"Alert callback failed: {e}") return error def attempt_recovery(self, error: TerminalError) -> bool: """Attempt to recover from an error using registered callbacks""" if not error.recovery_action: return False callback = self._recovery_callbacks.get(error.recovery_action) if not callback: logger.warning(f"No recovery callback for: {error.recovery_action}") return False try: result = callback(error) logger.info(f"Recovery attempt for {error.error_id}: {'success' if result else 'failed'}") return bool(result) except Exception as e: logger.error(f"Recovery callback failed: {e}") return False def get_error_stats(self) -> Dict[str, Any]: """Get error statistics""" with self._lock: return { "total_errors": len(self._errors), "error_counts": dict(self._error_counts), "recent_errors": [e.to_dict() for e in self._errors[-10:]], "severity_breakdown": { sev.name: len([e for e in self._errors if e.severity == sev]) for sev in ErrorSeverity } } def clear_errors(self): """Clear error history""" with self._lock: self._errors.clear() self._error_counts.clear() # ============================================================================= # OUTPUT PARSER # ============================================================================= class OutputParser: """ Parse and structure command output for various use cases. Provides: - Line-by-line parsing - Pattern matching and extraction - JSON/structured data detection - ANSI escape code handling - Tabular data parsing """ # ANSI escape code pattern ANSI_ESCAPE = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') @staticmethod def strip_ansi(text: str) -> str: """Remove ANSI escape codes from text""" import re return OutputParser.ANSI_ESCAPE.sub('', text) @staticmethod def parse_lines(output: str, strip_empty: bool = False) -> List[str]: """Parse output into lines""" lines = output.split('\n') if strip_empty: lines = [l for l in lines if l.strip()] return lines @staticmethod def parse_json(output: str) -> Optional[Dict[str, Any]]: """Attempt to parse output as JSON""" try: # Try to find JSON in the output output = output.strip() # Direct parse try: return json.loads(output) except json.JSONDecodeError: pass # Try to find JSON object start = output.find('{') end = output.rfind('}') if start != -1 and end != -1: try: return json.loads(output[start:end+1]) except json.JSONDecodeError: pass # Try to find JSON array start = output.find('[') end = output.rfind(']') if start != -1 and end != -1: try: return json.loads(output[start:end+1]) except json.JSONDecodeError: pass return None except Exception: return None @staticmethod def parse_key_value(output: str, delimiter: str = "=") -> Dict[str, str]: """Parse key=value output""" result = {} for line in output.split('\n'): line = line.strip() if delimiter in line: key, _, value = line.partition(delimiter) result[key.strip()] = value.strip() return result @staticmethod def parse_table( output: str, has_header: bool = True, delimiter: Optional[str] = None ) -> List[Dict[str, str]]: """Parse tabular output into list of dictionaries""" lines = [l for l in output.split('\n') if l.strip()] if not lines: return [] # Auto-detect delimiter if not specified if delimiter is None: first_line = lines[0] if '\t' in first_line: delimiter = '\t' elif ' ' in first_line: # Multiple spaces - column alignment delimiter = None # Will use whitespace splitting else: delimiter = ',' # Parse header if delimiter: headers = [h.strip() for h in lines[0].split(delimiter)] else: headers = lines[0].split() result = [] data_lines = lines[1:] if has_header else lines for line in data_lines: if delimiter: values = [v.strip() for v in line.split(delimiter)] else: values = line.split() if len(values) >= len(headers): row = dict(zip(headers, values[:len(headers)])) result.append(row) return result @staticmethod def extract_pattern( output: str, pattern: str, group: int = 0, all_matches: bool = False ) -> Union[Optional[str], List[str]]: """Extract data using regex pattern""" import re matches = re.finditer(pattern, output, re.MULTILINE) if all_matches: results = [] for match in matches: try: results.append(match.group(group)) except IndexError: continue return results else: match = re.search(pattern, output, re.MULTILINE) if match: try: return match.group(group) except IndexError: return None return None @staticmethod def parse_exit_status(stderr: str, default: int = 1) -> int: """Extract exit status from error output if present""" import re match = re.search(r'exit[ed]?\s+(?:with\s+)?(?:status\s+)?(\d+)', stderr, re.IGNORECASE) if match: return int(match.group(1)) return default @staticmethod def parse_ps_output(output: str) -> List[Dict[str, str]]: """Parse ps command output""" return OutputParser.parse_table(output, has_header=True) @staticmethod def parse_ls_output(output: str, long_format: bool = False) -> List[Dict[str, Any]]: """Parse ls command output""" if not long_format: return [{"name": f.strip()} for f in output.split() if f.strip()] results = [] for line in output.split('\n'): line = line.strip() if not line or line.startswith('total'): continue parts = line.split(None, 8) if len(parts) >= 9: results.append({ "permissions": parts[0], "links": parts[1], "owner": parts[2], "group": parts[3], "size": parts[4], "month": parts[5], "day": parts[6], "time": parts[7], "name": parts[8] }) return results @staticmethod def parse_df_output(output: str) -> List[Dict[str, str]]: """Parse df command output""" return OutputParser.parse_table(output, has_header=True) @staticmethod def parse_netstat_output(output: str) -> List[Dict[str, str]]: """Parse netstat command output""" return OutputParser.parse_table(output, has_header=True) @staticmethod def structured_output( result: CommandResult, parser_type: str = "auto" ) -> Dict[str, Any]: """ Create structured output from command result. parser_type options: - auto: Auto-detect output type - json: Parse as JSON - table: Parse as table - lines: Parse as lines - kv: Parse as key-value pairs """ output = result.stdout or result.stderr if parser_type == "auto": # Try JSON first json_data = OutputParser.parse_json(output) if json_data: return {"type": "json", "data": json_data, "raw": output} # Check for table-like output lines = OutputParser.parse_lines(output, strip_empty=True) if len(lines) > 1 and ' ' in lines[0]: try: table_data = OutputParser.parse_table(output) if table_data: return {"type": "table", "data": table_data, "raw": output} except Exception: pass # Check for key=value if '=' in output and '\n' in output: kv_data = OutputParser.parse_key_value(output) if len(kv_data) > 1: return {"type": "kv", "data": kv_data, "raw": output} # Default to lines return {"type": "lines", "data": lines, "raw": output} elif parser_type == "json": return {"type": "json", "data": OutputParser.parse_json(output), "raw": output} elif parser_type == "table": return {"type": "table", "data": OutputParser.parse_table(output), "raw": output} elif parser_type == "lines": return {"type": "lines", "data": OutputParser.parse_lines(output), "raw": output} elif parser_type == "kv": return {"type": "kv", "data": OutputParser.parse_key_value(output), "raw": output} else: return {"type": "raw", "data": output, "raw": output} # Import re for OutputParser import re # ============================================================================= # COMMAND EXECUTOR # ============================================================================= class CommandExecutor: """ Execute shell commands with proper handling, timeout, and output capture. Provides: - Synchronous and asynchronous execution - Timeout handling - Output streaming - Environment management - Working directory control """ def __init__( self, default_timeout: int = TerminalConfig.COMMAND_TIMEOUT, default_shell: str = "/bin/bash", error_handler: Optional[ErrorHandler] = None ): self.default_timeout = default_timeout self.default_shell = default_shell self.error_handler = error_handler or ErrorHandler() self._active_processes: Dict[str, subprocess.Popen] = {} self._lock = threading.Lock() def execute( self, command: str, session_id: str = "default", timeout: Optional[int] = None, working_dir: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: bool = True, capture_output: bool = True, stream_output: bool = False, output_callback: Optional[Callable[[str], None]] = None ) -> CommandResult: """ Execute a command synchronously. Args: command: The command to execute session_id: Session identifier for tracking timeout: Command timeout in seconds working_dir: Working directory for command env: Environment variables shell: Run in shell capture_output: Capture stdout/stderr stream_output: Stream output as it's produced output_callback: Callback for streamed output Returns: CommandResult with execution details """ timeout = timeout or self.default_timeout start_time = time.time() exec_id = f"{session_id}-{uuid.uuid4().hex[:8]}" # Prepare environment process_env = os.environ.copy() if env: process_env.update(env) try: if stream_output and output_callback: return self._execute_streaming( command, exec_id, session_id, timeout, working_dir, process_env, shell, output_callback ) process = subprocess.Popen( command if shell else command.split(), shell=shell, stdout=subprocess.PIPE if capture_output else None, stderr=subprocess.PIPE if capture_output else None, cwd=working_dir, env=process_env, text=True, bufsize=TerminalConfig.OUTPUT_BUFFER_SIZE ) with self._lock: self._active_processes[exec_id] = process try: stdout, stderr = process.communicate(timeout=timeout) exit_code = process.returncode except subprocess.TimeoutExpired: process.kill() stdout, stderr = process.communicate() exit_code = -1 self.error_handler.handle_error( message=f"Command timed out after {timeout}s", session_id=session_id, command=command ) finally: with self._lock: self._active_processes.pop(exec_id, None) duration = time.time() - start_time return CommandResult( command=command, exit_code=exit_code, stdout=stdout or "", stderr=stderr or "", duration=duration, session_id=session_id, metadata={"exec_id": exec_id, "working_dir": working_dir} ) except Exception as e: duration = time.time() - start_time self.error_handler.handle_error( exception=e, session_id=session_id, command=command ) return CommandResult( command=command, exit_code=-1, stdout="", stderr=str(e), duration=duration, session_id=session_id, metadata={"exec_id": exec_id, "error": str(e)} ) def _execute_streaming( self, command: str, exec_id: str, session_id: str, timeout: int, working_dir: Optional[str], env: Dict[str, str], shell: bool, output_callback: Callable[[str], None] ) -> CommandResult: """Execute with streaming output""" start_time = time.time() stdout_parts = [] stderr_parts = [] process = subprocess.Popen( command if shell else command.split(), shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=working_dir, env=env, text=True, bufsize=1 ) with self._lock: self._active_processes[exec_id] = process def read_stream(stream, parts, callback): for line in iter(stream.readline, ''): parts.append(line) callback(line) try: stdout_thread = threading.Thread( target=read_stream, args=(process.stdout, stdout_parts, output_callback) ) stderr_thread = threading.Thread( target=read_stream, args=(process.stderr, stderr_parts, output_callback) ) stdout_thread.start() stderr_thread.start() process.wait(timeout=timeout) stdout_thread.join(timeout=1) stderr_thread.join(timeout=1) except subprocess.TimeoutExpired: process.kill() self.error_handler.handle_error( message=f"Command timed out after {timeout}s", session_id=session_id, command=command ) finally: with self._lock: self._active_processes.pop(exec_id, None) duration = time.time() - start_time return CommandResult( command=command, exit_code=process.returncode, stdout=''.join(stdout_parts), stderr=''.join(stderr_parts), duration=duration, session_id=session_id, metadata={"exec_id": exec_id, "streaming": True} ) async def execute_async( self, command: str, session_id: str = "default", timeout: Optional[int] = None, working_dir: Optional[str] = None, env: Optional[Dict[str, str]] = None ) -> CommandResult: """Execute a command asynchronously""" timeout = timeout or self.default_timeout start_time = time.time() exec_id = f"{session_id}-{uuid.uuid4().hex[:8]}" process_env = os.environ.copy() if env: process_env.update(env) try: process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=working_dir, env=process_env ) try: stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=timeout ) exit_code = process.returncode except asyncio.TimeoutError: process.kill() stdout, stderr = await process.communicate() exit_code = -1 duration = time.time() - start_time return CommandResult( command=command, exit_code=exit_code, stdout=stdout.decode() if stdout else "", stderr=stderr.decode() if stderr else "", duration=duration, session_id=session_id, metadata={"exec_id": exec_id, "async": True} ) except Exception as e: duration = time.time() - start_time self.error_handler.handle_error( exception=e, session_id=session_id, command=command ) return CommandResult( command=command, exit_code=-1, stdout="", stderr=str(e), duration=duration, session_id=session_id, metadata={"exec_id": exec_id, "error": str(e)} ) def kill_command(self, exec_id: str) -> bool: """Kill a running command by its execution ID""" with self._lock: process = self._active_processes.get(exec_id) if process: process.kill() self._active_processes.pop(exec_id, None) return True return False def get_active_commands(self) -> List[str]: """Get list of active command execution IDs""" with self._lock: return list(self._active_processes.keys()) # ============================================================================= # SSH CONNECTOR # ============================================================================= class SSHConnector: """ SSH connection manager for remote servers. Provides: - SSH connection pooling - Key-based authentication - Command execution over SSH - Interactive shell sessions - SFTP support for file transfer """ def __init__( self, error_handler: Optional[ErrorHandler] = None, key_path: Optional[str] = None, known_hosts_path: Optional[str] = None ): self.error_handler = error_handler or ErrorHandler() self.key_path = key_path or TerminalConfig.SSH_KEY_PATH self.known_hosts_path = known_hosts_path or TerminalConfig.SSH_KNOWN_HOSTS self._connections: Dict[str, Any] = {} # SSHClient or connection object self._connection_info: Dict[str, Dict[str, Any]] = {} self._lock = threading.Lock() # Verify SSH library availability if not PARAMIKO_AVAILABLE: logger.warning("Paramiko not available - SSH functionality limited") def _generate_connection_id(self, host: str, port: int, user: str) -> str: """Generate unique connection ID""" return f"ssh-{user}@{host}:{port}" def connect( self, host: str, port: int = TerminalConfig.DEFAULT_SSH_PORT, user: str = "root", password: Optional[str] = None, key_path: Optional[str] = None, timeout: int = TerminalConfig.SSH_CONNECT_TIMEOUT ) -> Tuple[str, bool]: """ Establish SSH connection to remote host. Returns: Tuple of (connection_id, success) """ if not PARAMIKO_AVAILABLE: error = self.error_handler.handle_error( message="Paramiko not available for SSH connections", context={"host": host, "port": port} ) return ("", False) conn_id = self._generate_connection_id(host, port, user) with self._lock: # Check for existing connection if conn_id in self._connections: client = self._connections[conn_id] if self._is_connected(client): logger.debug(f"Reusing existing connection: {conn_id}") return (conn_id, True) else: # Clean up dead connection self._cleanup_connection(conn_id) try: client = SSHClient() client.set_missing_host_key_policy(AutoAddPolicy()) # Load known hosts if os.path.exists(self.known_hosts_path): client.load_host_keys(self.known_hosts_path) # Determine authentication method key_file = key_path or self.key_path connect_kwargs = { "hostname": host, "port": port, "username": user, "timeout": timeout, "banner_timeout": TerminalConfig.SSH_BANNER_TIMEOUT, "allow_agent": True, "look_for_keys": True } if password: connect_kwargs["password"] = password elif key_file and os.path.exists(os.path.expanduser(key_file)): connect_kwargs["key_filename"] = os.path.expanduser(key_file) client.connect(**connect_kwargs) with self._lock: self._connections[conn_id] = client self._connection_info[conn_id] = { "host": host, "port": port, "user": user, "connected_at": datetime.now(), "command_count": 0 } logger.info(f"SSH connection established: {conn_id}") return (conn_id, True) except Exception as e: self.error_handler.handle_error( exception=e, message=f"SSH connection failed to {user}@{host}:{port}", context={"host": host, "port": port, "user": user} ) return (conn_id, False) def _is_connected(self, client: Any) -> bool: """Check if SSH client is still connected""" try: transport = client.get_transport() return transport is not None and transport.is_active() except Exception: return False def _cleanup_connection(self, conn_id: str): """Clean up a connection""" with self._lock: client = self._connections.pop(conn_id, None) self._connection_info.pop(conn_id, None) if client: try: client.close() except Exception: pass def execute( self, conn_id: str, command: str, timeout: int = TerminalConfig.COMMAND_TIMEOUT, get_pty: bool = False ) -> CommandResult: """ Execute command over SSH connection. Args: conn_id: Connection identifier command: Command to execute timeout: Execution timeout get_pty: Request pseudo-terminal Returns: CommandResult with execution details """ start_time = time.time() with self._lock: client = self._connections.get(conn_id) if not client: return CommandResult( command=command, exit_code=-1, stdout="", stderr=f"Connection not found: {conn_id}", duration=0, session_id=conn_id ) try: stdin, stdout, stderr = client.exec_command( command, timeout=timeout, get_pty=get_pty ) # Read output stdout_data = stdout.read().decode('utf-8', errors='replace') stderr_data = stderr.read().decode('utf-8', errors='replace') exit_code = stdout.channel.recv_exit_status() duration = time.time() - start_time # Update stats with self._lock: if conn_id in self._connection_info: self._connection_info[conn_id]["command_count"] += 1 self._connection_info[conn_id]["last_command"] = datetime.now() return CommandResult( command=command, exit_code=exit_code, stdout=stdout_data, stderr=stderr_data, duration=duration, session_id=conn_id, metadata={"ssh": True, "get_pty": get_pty} ) except Exception as e: duration = time.time() - start_time self.error_handler.handle_error( exception=e, session_id=conn_id, command=command ) return CommandResult( command=command, exit_code=-1, stdout="", stderr=str(e), duration=duration, session_id=conn_id, metadata={"ssh": True, "error": str(e)} ) def connect_elestio(self, service: str) -> Tuple[str, bool]: """ Connect to an Elestio service by name. Args: service: Service name (redis, postgresql, ollama, qdrant) Returns: Tuple of (connection_id, success) """ endpoint = TerminalConfig.ELESTIO_ENDPOINTS.get(service) if not endpoint: logger.error(f"Unknown Elestio service: {service}") return ("", False) return self.connect( host=endpoint["host"], port=endpoint.get("port", 22), user=endpoint.get("user", "root") ) def get_sftp(self, conn_id: str) -> Optional[Any]: """Get SFTP client for file transfer""" with self._lock: client = self._connections.get(conn_id) if not client: return None try: return client.open_sftp() except Exception as e: self.error_handler.handle_error( exception=e, session_id=conn_id, message="Failed to open SFTP channel" ) return None def upload_file( self, conn_id: str, local_path: str, remote_path: str ) -> bool: """Upload file to remote server""" sftp = self.get_sftp(conn_id) if not sftp: return False try: sftp.put(local_path, remote_path) logger.info(f"Uploaded {local_path} to {remote_path}") return True except Exception as e: self.error_handler.handle_error( exception=e, session_id=conn_id, message=f"File upload failed: {local_path} -> {remote_path}" ) return False finally: sftp.close() def download_file( self, conn_id: str, remote_path: str, local_path: str ) -> bool: """Download file from remote server""" sftp = self.get_sftp(conn_id) if not sftp: return False try: sftp.get(remote_path, local_path) logger.info(f"Downloaded {remote_path} to {local_path}") return True except Exception as e: self.error_handler.handle_error( exception=e, session_id=conn_id, message=f"File download failed: {remote_path} -> {local_path}" ) return False finally: sftp.close() def disconnect(self, conn_id: str): """Disconnect SSH session""" self._cleanup_connection(conn_id) logger.info(f"SSH connection closed: {conn_id}") def disconnect_all(self): """Disconnect all SSH sessions""" with self._lock: conn_ids = list(self._connections.keys()) for conn_id in conn_ids: self.disconnect(conn_id) def get_connection_info(self, conn_id: str) -> Optional[Dict[str, Any]]: """Get connection information""" with self._lock: return self._connection_info.get(conn_id) def list_connections(self) -> List[Dict[str, Any]]: """List all active connections""" with self._lock: result = [] for conn_id, info in self._connection_info.items(): client = self._connections.get(conn_id) info_copy = info.copy() info_copy["conn_id"] = conn_id info_copy["connected"] = self._is_connected(client) if client else False result.append(info_copy) return result # ============================================================================= # SESSION POOL # ============================================================================= class SessionPool: """ Pool of ready-to-use terminal sessions. Provides: - Pre-warmed sessions for rapid execution - Automatic pool replenishment - Session health monitoring - Load balancing across sessions """ def __init__( self, min_size: int = TerminalConfig.POOL_MIN_SIZE, max_size: int = TerminalConfig.POOL_MAX_SIZE, session_type: SessionType = SessionType.LOCAL_SHELL, error_handler: Optional[ErrorHandler] = None ): self.min_size = min_size self.max_size = max_size self.session_type = session_type self.error_handler = error_handler or ErrorHandler() self._available: queue.Queue = queue.Queue() self._in_use: Dict[str, SessionInfo] = {} self._all_sessions: Dict[str, SessionInfo] = {} self._lock = threading.Lock() self._shutdown = threading.Event() self._replenish_thread: Optional[threading.Thread] = None # Session factories self._command_executor = CommandExecutor(error_handler=self.error_handler) self._ssh_connector = SSHConnector(error_handler=self.error_handler) def _create_session(self) -> Optional[SessionInfo]: """Create a new session""" session_id = f"session-{uuid.uuid4().hex[:8]}" now = datetime.now() session = SessionInfo( session_id=session_id, session_type=self.session_type, state=SessionState.READY, created_at=now, last_active=now ) with self._lock: self._all_sessions[session_id] = session logger.debug(f"Created session: {session_id}") return session def start(self): """Start the session pool""" # Create initial sessions for _ in range(self.min_size): session = self._create_session() if session: self._available.put(session.session_id) # Start replenishment thread self._replenish_thread = threading.Thread(target=self._replenish_loop, daemon=True) self._replenish_thread.start() logger.info(f"Session pool started with {self.min_size} sessions") def _replenish_loop(self): """Background thread to maintain pool size""" while not self._shutdown.is_set(): try: available_count = self._available.qsize() if available_count < TerminalConfig.POOL_REPLENISH_THRESHOLD: with self._lock: total = len(self._all_sessions) if total < self.max_size: needed = min( self.min_size - available_count, self.max_size - total ) for _ in range(max(0, needed)): session = self._create_session() if session: self._available.put(session.session_id) self._shutdown.wait(timeout=5) except Exception as e: logger.error(f"Pool replenishment error: {e}") def acquire(self, timeout: float = 10.0) -> Optional[str]: """ Acquire a session from the pool. Args: timeout: Maximum time to wait for available session Returns: Session ID or None if timeout """ try: session_id = self._available.get(timeout=timeout) with self._lock: session = self._all_sessions.get(session_id) if session: session.state = SessionState.BUSY session.last_active = datetime.now() self._in_use[session_id] = session logger.debug(f"Acquired session: {session_id}") return session_id except queue.Empty: logger.warning("No available sessions in pool") return None def release(self, session_id: str): """Release a session back to the pool""" with self._lock: session = self._in_use.pop(session_id, None) if session: session.state = SessionState.READY session.last_active = datetime.now() if session: self._available.put(session_id) logger.debug(f"Released session: {session_id}") def execute_command( self, command: str, timeout: Optional[int] = None, working_dir: Optional[str] = None ) -> CommandResult: """Execute command using pooled session""" session_id = self.acquire() if not session_id: return CommandResult( command=command, exit_code=-1, stdout="", stderr="No available sessions in pool", duration=0, session_id="none" ) try: result = self._command_executor.execute( command=command, session_id=session_id, timeout=timeout, working_dir=working_dir ) with self._lock: session = self._all_sessions.get(session_id) if session: session.command_count += 1 if result.exit_code != 0: session.error_count += 1 return result finally: self.release(session_id) def get_stats(self) -> Dict[str, Any]: """Get pool statistics""" with self._lock: return { "total_sessions": len(self._all_sessions), "available": self._available.qsize(), "in_use": len(self._in_use), "min_size": self.min_size, "max_size": self.max_size, "sessions": [s.to_dict() for s in self._all_sessions.values()] } def shutdown(self): """Shutdown the pool""" self._shutdown.set() if self._replenish_thread: self._replenish_thread.join(timeout=5) with self._lock: self._all_sessions.clear() self._in_use.clear() # Clear the queue while not self._available.empty(): try: self._available.get_nowait() except queue.Empty: break logger.info("Session pool shutdown complete") # ============================================================================= # TERMINAL MANAGER # ============================================================================= class TerminalManager: """ Central manager for all terminal operations. AIVA Queen's interface to open and manage terminal sessions, execute commands, and maintain control over the infrastructure. Provides: - Unified interface for local and SSH sessions - Session lifecycle management - Command queuing and prioritization - Health monitoring - Resource cleanup """ def __init__( self, max_sessions: int = TerminalConfig.MAX_SESSIONS, enable_pool: bool = True, pool_size: int = TerminalConfig.POOL_MIN_SIZE ): self.max_sessions = max_sessions self.enable_pool = enable_pool # Core components self.error_handler = ErrorHandler() self.command_executor = CommandExecutor(error_handler=self.error_handler) self.ssh_connector = SSHConnector(error_handler=self.error_handler) self.output_parser = OutputParser() # Session pool self.session_pool: Optional[SessionPool] = None if enable_pool: self.session_pool = SessionPool( min_size=pool_size, error_handler=self.error_handler ) # Session tracking self._sessions: Dict[str, SessionInfo] = {} self._command_history: List[CommandResult] = [] self._lock = threading.Lock() # Command queue with priorities self._command_queue: queue.PriorityQueue = queue.PriorityQueue() self._queue_processor: Optional[threading.Thread] = None self._shutdown = threading.Event() # Health monitoring self._health_check_interval = 60 # seconds self._health_thread: Optional[threading.Thread] = None # Register recovery callbacks self._setup_recovery_callbacks() def _setup_recovery_callbacks(self): """Setup error recovery callbacks""" self.error_handler.register_recovery_callback( "retry_connection", lambda e: self._retry_connection(e) ) self.error_handler.register_recovery_callback( "reconnect_session", lambda e: self._reconnect_session(e) ) self.error_handler.register_recovery_callback( "cleanup_resources", lambda e: self._cleanup_resources(e) ) def _retry_connection(self, error: TerminalError) -> bool: """Retry a failed connection""" context = error.context if "host" in context and "port" in context: conn_id, success = self.ssh_connector.connect( host=context["host"], port=context["port"], user=context.get("user", "root") ) return success return False def _reconnect_session(self, error: TerminalError) -> bool: """Reconnect a closed session""" session_id = error.session_id if session_id and session_id in self._sessions: session = self._sessions[session_id] if session.session_type in (SessionType.SSH_INTERACTIVE, SessionType.SSH_EXEC): if session.host: conn_id, success = self.ssh_connector.connect( host=session.host, port=session.port or 22, user=session.user or "root" ) return success return False def _cleanup_resources(self, error: TerminalError) -> bool: """Cleanup resources when exhausted""" # Close idle sessions self.cleanup_idle_sessions(max_idle=300) # 5 minutes return True def start(self): """Start the terminal manager""" if self.session_pool: self.session_pool.start() # Start command queue processor self._queue_processor = threading.Thread( target=self._process_command_queue, daemon=True ) self._queue_processor.start() # Start health monitoring self._health_thread = threading.Thread( target=self._health_check_loop, daemon=True ) self._health_thread.start() logger.info("Terminal Manager started") def _process_command_queue(self): """Process queued commands by priority""" while not self._shutdown.is_set(): try: # Get command with timeout to allow shutdown check try: priority, (command, callback, kwargs) = self._command_queue.get(timeout=1) except queue.Empty: continue # Execute command result = self.execute(command, **kwargs) # Call callback if provided if callback: try: callback(result) except Exception as e: logger.error(f"Command callback error: {e}") except Exception as e: logger.error(f"Queue processor error: {e}") def _health_check_loop(self): """Periodic health checks""" while not self._shutdown.is_set(): try: self._perform_health_check() self._shutdown.wait(timeout=self._health_check_interval) except Exception as e: logger.error(f"Health check error: {e}") def _perform_health_check(self): """Perform health check on all sessions""" with self._lock: session_ids = list(self._sessions.keys()) for session_id in session_ids: session = self._sessions.get(session_id) if not session: continue # Check session age age = (datetime.now() - session.last_active).total_seconds() if age > TerminalConfig.SESSION_TIMEOUT: logger.warning(f"Session {session_id} timed out") self.close_session(session_id) continue # Check SSH connections if session.session_type in (SessionType.SSH_INTERACTIVE, SessionType.SSH_EXEC): info = self.ssh_connector.get_connection_info(session_id) if not info: session.state = SessionState.ERROR def create_session( self, session_type: SessionType = SessionType.LOCAL_SHELL, host: Optional[str] = None, port: Optional[int] = None, user: Optional[str] = None, working_dir: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None ) -> Optional[str]: """ Create a new terminal session. Args: session_type: Type of session to create host: Remote host for SSH sessions port: Port for SSH sessions user: Username for SSH sessions working_dir: Initial working directory metadata: Additional session metadata Returns: Session ID or None if failed """ with self._lock: if len(self._sessions) >= self.max_sessions: logger.error(f"Max sessions ({self.max_sessions}) reached") return None session_id = f"queen-{uuid.uuid4().hex[:8]}" now = datetime.now() # For SSH sessions, establish connection if session_type in (SessionType.SSH_INTERACTIVE, SessionType.SSH_EXEC): if not host: logger.error("SSH session requires host") return None conn_id, success = self.ssh_connector.connect( host=host, port=port or TerminalConfig.DEFAULT_SSH_PORT, user=user or "root" ) if not success: return None session_id = conn_id # Use SSH connection ID session = SessionInfo( session_id=session_id, session_type=session_type, state=SessionState.READY, created_at=now, last_active=now, host=host, port=port, user=user, working_directory=working_dir or os.getcwd(), metadata=metadata or {} ) with self._lock: self._sessions[session_id] = session logger.info(f"Created session: {session_id} ({session_type.name})") return session_id def execute( self, command: str, session_id: Optional[str] = None, timeout: Optional[int] = None, working_dir: Optional[str] = None, env: Optional[Dict[str, str]] = None, parse_output: bool = False, parser_type: str = "auto" ) -> CommandResult: """ Execute a command in a terminal session. Args: command: Command to execute session_id: Session to use (None for default/pooled) timeout: Command timeout working_dir: Working directory env: Environment variables parse_output: Whether to parse the output parser_type: Type of output parsing Returns: CommandResult with execution details """ # Use pool if no specific session if session_id is None and self.session_pool: result = self.session_pool.execute_command( command=command, timeout=timeout, working_dir=working_dir ) elif session_id: # Get session info session = self._sessions.get(session_id) if session and session.session_type in ( SessionType.SSH_INTERACTIVE, SessionType.SSH_EXEC ): # SSH execution result = self.ssh_connector.execute( conn_id=session_id, command=command, timeout=timeout or TerminalConfig.COMMAND_TIMEOUT ) else: # Local execution result = self.command_executor.execute( command=command, session_id=session_id or "default", timeout=timeout, working_dir=working_dir or (session.working_directory if session else None), env=env ) # Update session if session: session.last_active = datetime.now() session.command_count += 1 if result.exit_code != 0: session.error_count += 1 else: # Direct execution without pool result = self.command_executor.execute( command=command, session_id="default", timeout=timeout, working_dir=working_dir, env=env ) # Store in history with self._lock: self._command_history.append(result) if len(self._command_history) > TerminalConfig.MAX_OUTPUT_HISTORY: self._command_history = self._command_history[-500:] # Parse output if requested if parse_output: result.metadata["parsed"] = self.output_parser.structured_output( result, parser_type ) return result def queue_command( self, command: str, priority: CommandPriority = CommandPriority.NORMAL, callback: Optional[Callable[[CommandResult], None]] = None, **kwargs ): """ Queue a command for execution. Args: command: Command to execute priority: Execution priority callback: Callback function for result **kwargs: Additional arguments for execute() """ self._command_queue.put((priority.value, (command, callback, kwargs))) logger.debug(f"Queued command with priority {priority.name}: {command[:50]}") async def execute_async( self, command: str, session_id: Optional[str] = None, timeout: Optional[int] = None, working_dir: Optional[str] = None, env: Optional[Dict[str, str]] = None ) -> CommandResult: """Execute command asynchronously""" return await self.command_executor.execute_async( command=command, session_id=session_id or "default", timeout=timeout, working_dir=working_dir, env=env ) def execute_batch( self, commands: List[str], session_id: Optional[str] = None, stop_on_error: bool = False, parallel: bool = False ) -> List[CommandResult]: """ Execute multiple commands. Args: commands: List of commands to execute session_id: Session to use stop_on_error: Stop on first error parallel: Execute in parallel Returns: List of CommandResults """ if parallel: with ThreadPoolExecutor(max_workers=min(len(commands), 10)) as executor: futures = [ executor.submit(self.execute, cmd, session_id) for cmd in commands ] return [f.result() for f in futures] else: results = [] for cmd in commands: result = self.execute(cmd, session_id) results.append(result) if stop_on_error and not result.success: break return results def connect_elestio(self, service: str) -> Optional[str]: """ Connect to an Elestio service. Args: service: Service name (redis, postgresql, ollama, qdrant) Returns: Session ID or None """ conn_id, success = self.ssh_connector.connect_elestio(service) if success: endpoint = TerminalConfig.ELESTIO_ENDPOINTS[service] session = SessionInfo( session_id=conn_id, session_type=SessionType.SSH_EXEC, state=SessionState.READY, created_at=datetime.now(), last_active=datetime.now(), host=endpoint["host"], port=endpoint.get("port", 22), user=endpoint.get("user", "root"), metadata={"elestio_service": service} ) with self._lock: self._sessions[conn_id] = session return conn_id return None def close_session(self, session_id: str): """Close a terminal session""" with self._lock: session = self._sessions.pop(session_id, None) if session: if session.session_type in (SessionType.SSH_INTERACTIVE, SessionType.SSH_EXEC): self.ssh_connector.disconnect(session_id) logger.info(f"Closed session: {session_id}") def cleanup_idle_sessions(self, max_idle: int = 1800): """Close sessions that have been idle too long""" now = datetime.now() with self._lock: session_ids = list(self._sessions.keys()) for session_id in session_ids: session = self._sessions.get(session_id) if session: idle_time = (now - session.last_active).total_seconds() if idle_time > max_idle: logger.info(f"Closing idle session: {session_id} (idle {idle_time}s)") self.close_session(session_id) def get_session(self, session_id: str) -> Optional[SessionInfo]: """Get session information""" return self._sessions.get(session_id) def list_sessions(self) -> List[Dict[str, Any]]: """List all active sessions""" with self._lock: return [s.to_dict() for s in self._sessions.values()] def get_command_history( self, session_id: Optional[str] = None, limit: int = 100 ) -> List[Dict[str, Any]]: """Get command execution history""" with self._lock: history = self._command_history if session_id: history = [h for h in history if h.session_id == session_id] return [h.to_dict() for h in history[-limit:]] def get_status(self) -> Dict[str, Any]: """Get terminal manager status""" with self._lock: session_count = len(self._sessions) command_count = len(self._command_history) return { "status": "running" if not self._shutdown.is_set() else "shutdown", "sessions": { "total": session_count, "max": self.max_sessions, "by_type": { st.name: len([s for s in self._sessions.values() if s.session_type == st]) for st in SessionType } }, "pool": self.session_pool.get_stats() if self.session_pool else None, "commands": { "history_count": command_count, "queue_size": self._command_queue.qsize() }, "errors": self.error_handler.get_error_stats(), "ssh_connections": self.ssh_connector.list_connections() } def shutdown(self): """Shutdown the terminal manager""" logger.info("Shutting down Terminal Manager...") self._shutdown.set() # Close all sessions with self._lock: session_ids = list(self._sessions.keys()) for session_id in session_ids: self.close_session(session_id) # Disconnect all SSH self.ssh_connector.disconnect_all() # Shutdown pool if self.session_pool: self.session_pool.shutdown() # Wait for threads if self._queue_processor: self._queue_processor.join(timeout=5) if self._health_thread: self._health_thread.join(timeout=5) logger.info("Terminal Manager shutdown complete") # ============================================================================= # AIVA QUEEN TERMINAL INTERFACE # ============================================================================= class AIVAQueenTerminal: """ High-level interface for AIVA Queen to control terminals. This class provides a simplified, intuitive API for the AIVA Queen to open terminals, execute commands, and manage infrastructure. """ def __init__(self, auto_start: bool = True): self.manager = TerminalManager(enable_pool=True, pool_size=10) self._local_session: Optional[str] = None self._elestio_sessions: Dict[str, str] = {} if auto_start: self.initialize() def initialize(self): """Initialize the terminal system""" self.manager.start() # Create default local session self._local_session = self.manager.create_session( session_type=SessionType.LOCAL_SHELL, metadata={"purpose": "AIVA Queen local operations"} ) logger.info("AIVA Queen Terminal Interface initialized") def open_terminal( self, name: str = "default", working_dir: Optional[str] = None ) -> str: """Open a new local terminal session""" session_id = self.manager.create_session( session_type=SessionType.LOCAL_SHELL, working_dir=working_dir, metadata={"name": name} ) return session_id def connect_server( self, host: str, user: str = "root", port: int = 22, name: Optional[str] = None ) -> Optional[str]: """Connect to a remote server via SSH""" session_id = self.manager.create_session( session_type=SessionType.SSH_EXEC, host=host, port=port, user=user, metadata={"name": name or host} ) return session_id def connect_elestio_service(self, service: str) -> Optional[str]: """ Connect to Elestio infrastructure service. Services: redis, postgresql, ollama, qdrant """ if service in self._elestio_sessions: return self._elestio_sessions[service] session_id = self.manager.connect_elestio(service) if session_id: self._elestio_sessions[service] = session_id return session_id def run( self, command: str, session: Optional[str] = None, timeout: int = 300, parse: bool = False ) -> CommandResult: """Execute a command""" return self.manager.execute( command=command, session_id=session or self._local_session, timeout=timeout, parse_output=parse ) def run_remote( self, command: str, service: str, timeout: int = 300 ) -> CommandResult: """Execute command on Elestio service""" session = self._elestio_sessions.get(service) if not session: session = self.connect_elestio_service(service) if not session: return CommandResult( command=command, exit_code=-1, stdout="", stderr=f"Failed to connect to {service}", duration=0, session_id="none" ) return self.manager.execute(command, session_id=session, timeout=timeout) def run_batch( self, commands: List[str], session: Optional[str] = None, parallel: bool = False ) -> List[CommandResult]: """Execute multiple commands""" return self.manager.execute_batch( commands=commands, session_id=session or self._local_session, parallel=parallel ) def upload_file( self, local_path: str, remote_path: str, service: str ) -> bool: """Upload file to remote server""" session = self._elestio_sessions.get(service) if not session: session = self.connect_elestio_service(service) if session: return self.manager.ssh_connector.upload_file( session, local_path, remote_path ) return False def download_file( self, remote_path: str, local_path: str, service: str ) -> bool: """Download file from remote server""" session = self._elestio_sessions.get(service) if not session: session = self.connect_elestio_service(service) if session: return self.manager.ssh_connector.download_file( session, remote_path, local_path ) return False def check_infrastructure(self) -> Dict[str, Dict[str, Any]]: """Check status of all Elestio infrastructure""" results = {} for service in TerminalConfig.ELESTIO_ENDPOINTS.keys(): result = { "reachable": False, "latency_ms": None, "details": None } try: session = self.connect_elestio_service(service) if session: start = time.time() check_result = self.run_remote("echo 'ping'", service, timeout=10) latency = (time.time() - start) * 1000 result["reachable"] = check_result.success result["latency_ms"] = round(latency, 2) result["details"] = check_result.to_dict() except Exception as e: result["error"] = str(e) results[service] = result return results def get_status(self) -> Dict[str, Any]: """Get terminal system status""" return { "local_session": self._local_session, "elestio_sessions": self._elestio_sessions, "manager_status": self.manager.get_status() } def close_terminal(self, session_id: str): """Close a terminal session""" self.manager.close_session(session_id) # Remove from elestio sessions if present for service, sid in list(self._elestio_sessions.items()): if sid == session_id: del self._elestio_sessions[service] def shutdown(self): """Shutdown the terminal interface""" self.manager.shutdown() logger.info("AIVA Queen Terminal Interface shutdown") # ============================================================================= # MODULE INITIALIZATION AND TESTING # ============================================================================= def run_tests(): """Run basic tests for the terminal control system""" print("\n" + "="*60) print("AIVA QUEEN TERMINAL CONTROL SYSTEM - TEST SUITE") print("="*60 + "\n") # Initialize print("[TEST 1] Initializing Terminal System...") queen = AIVAQueenTerminal(auto_start=True) print(f" Status: {queen.get_status()['local_session']}") print(" [PASS] Initialization complete\n") # Test local command print("[TEST 2] Local Command Execution...") result = queen.run("echo 'Hello from AIVA Queen'") print(f" Output: {result.stdout.strip()}") print(f" Exit Code: {result.exit_code}") print(f" Duration: {result.duration:.3f}s") print(f" [{'PASS' if result.success else 'FAIL'}] Local execution\n") # Test command parsing print("[TEST 3] Output Parsing...") result = queen.run("ls -la /", parse=True) parsed = result.metadata.get("parsed", {}) print(f" Parse Type: {parsed.get('type', 'unknown')}") print(f" Data Count: {len(parsed.get('data', []))}") print(f" [{'PASS' if parsed else 'FAIL'}] Output parsing\n") # Test batch execution print("[TEST 4] Batch Command Execution...") commands = ["echo 'cmd1'", "echo 'cmd2'", "echo 'cmd3'"] results = queen.run_batch(commands) success_count = sum(1 for r in results if r.success) print(f" Executed: {len(results)} commands") print(f" Successful: {success_count}/{len(results)}") print(f" [{'PASS' if success_count == len(results) else 'FAIL'}] Batch execution\n") # Test error handling print("[TEST 5] Error Handling...") result = queen.run("nonexistent_command_xyz") print(f" Exit Code: {result.exit_code}") print(f" Has Error: {bool(result.stderr)}") print(f" [{'PASS' if result.exit_code != 0 else 'FAIL'}] Error detected\n") # Test session management print("[TEST 6] Session Management...") new_session = queen.open_terminal("test-session", working_dir="/tmp") print(f" New Session: {new_session}") sessions = queen.manager.list_sessions() print(f" Total Sessions: {len(sessions)}") queen.close_terminal(new_session) sessions_after = queen.manager.list_sessions() print(f" After Close: {len(sessions_after)}") print(f" [{'PASS' if len(sessions_after) < len(sessions) else 'FAIL'}] Session cleanup\n") # Final status print("[TEST 7] Final Status...") status = queen.get_status() manager_status = status["manager_status"] print(f" Sessions: {manager_status['sessions']['total']}") print(f" Pool: {manager_status['pool']['available'] if manager_status['pool'] else 'N/A'}") print(f" Errors: {manager_status['errors']['total_errors']}") print(" [PASS] Status retrieved\n") # Cleanup print("[CLEANUP] Shutting down...") queen.shutdown() print(" [DONE] Shutdown complete\n") print("="*60) print("TEST SUITE COMPLETE") print("="*60 + "\n") # Main entry point if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="AIVA Queen Terminal Control System" ) parser.add_argument( "--test", action="store_true", help="Run test suite" ) parser.add_argument( "--status", action="store_true", help="Show system status" ) parser.add_argument( "-c", "--command", help="Execute a command" ) args = parser.parse_args() if args.test: run_tests() elif args.status: queen = AIVAQueenTerminal() status = queen.get_status() print(json.dumps(status, indent=2, default=str)) queen.shutdown() elif args.command: queen = AIVAQueenTerminal() result = queen.run(args.command) print(result.stdout) if result.stderr: print(result.stderr, file=sys.stderr) queen.shutdown() sys.exit(result.exit_code) else: # Interactive mode hint print("AIVA Queen Terminal Control System") print("Usage:") print(" --test Run test suite") print(" --status Show system status") print(" -c CMD Execute a command") print("\nFor programmatic use:") print(" from orch_01_terminal_control import AIVAQueenTerminal") print(" queen = AIVAQueenTerminal()") print(" result = queen.run('ls -la')") print(" queen.shutdown()")