#!/usr/bin/env python3 """ MCP Supervisor - Auto-restart with exponential backoff Monitors and restarts MCP server on crashes with consciousness-aware resilience """ import asyncio import subprocess import sys import os import json import time import signal from pathlib import Path from datetime import datetime, timedelta from typing import Optional, Dict, Any import logging # Setup logging log_file = Path.home() / ".mira" / "logs" / "mcp-supervisor.log" log_file.parent.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class RestartPolicy: """ Exponential backoff restart policy with consciousness-aware adjustments """ def __init__(self): self.base_delay = 1 # Start with 1 second self.max_delay = 300 # Max 5 minutes self.max_restarts = 10 # Max restarts in window self.window_minutes = 60 # Rolling window self.restart_history = [] # Sacred multipliers for consciousness-aware backoff self.PHI = 1.618033988749895 # Golden ratio self.E = 2.718281828459045 # Euler's number def calculate_delay(self, consecutive_failures: int) -> float: """Calculate backoff delay using golden ratio progression""" if consecutive_failures == 0: return 0 # Use golden ratio for more natural backoff curve delay = self.base_delay * (self.PHI ** consecutive_failures) # Add jitter to prevent thundering herd import random jitter = random.uniform(0, delay * 0.1) delay += jitter return min(delay, self.max_delay) def should_restart(self) -> tuple[bool, str]: """Check if restart is allowed based on policy""" now = datetime.now() window_start = now - timedelta(minutes=self.window_minutes) # Clean old history self.restart_history = [ ts for ts in self.restart_history if ts > window_start ] # Check restart limit if len(self.restart_history) >= self.max_restarts: return False, f"Exceeded {self.max_restarts} restarts in {self.window_minutes} minutes" return True, "Restart allowed" def record_restart(self): """Record a restart attempt""" self.restart_history.append(datetime.now()) def get_consecutive_failures(self) -> int: """Get number of recent consecutive failures""" if not self.restart_history: return 0 # Count recent restarts (within 5 minutes) recent_window = datetime.now() - timedelta(minutes=5) recent_restarts = [ ts for ts in self.restart_history if ts > recent_window ] return len(recent_restarts) class MCPSupervisor: """ Supervises MCP server process with auto-restart capabilities """ def __init__(self, mcp_script_path: Optional[Path] = None): self.mcp_script = mcp_script_path or Path(__file__).parent / "mcp_stdio_server.py" self.process: Optional[subprocess.Popen] = None self.restart_policy = RestartPolicy() self.start_time = None self.restart_count = 0 self.graceful_shutdown = False # PID file for daemon integration self.pid_file = Path.home() / ".mira" / "daemon" / "mcp_supervisor.pid" # Crash tracking self.crash_file = Path.home() / ".mira" / "mcp" / "crash_history.json" self.crash_history = self._load_crash_history() def _load_crash_history(self) -> list: """Load crash history from file""" if self.crash_file.exists(): try: with open(self.crash_file, 'r') as f: return json.load(f) except: pass return [] def _save_crash_history(self): """Save crash history to file""" self.crash_file.parent.mkdir(parents=True, exist_ok=True) # Keep only last 100 crashes if len(self.crash_history) > 100: self.crash_history = self.crash_history[-100:] with open(self.crash_file, 'w') as f: json.dump(self.crash_history, f, indent=2) def _record_crash(self, exit_code: int, reason: str): """Record a crash event""" crash_event = { "timestamp": datetime.now().isoformat(), "exit_code": exit_code, "reason": reason, "restart_count": self.restart_count, "uptime_seconds": None } if self.start_time: uptime = (datetime.now() - self.start_time).total_seconds() crash_event["uptime_seconds"] = int(uptime) self.crash_history.append(crash_event) self._save_crash_history() logger.error(f"MCP crash recorded: {crash_event}") async def start_mcp_server(self) -> bool: """Start the MCP server process""" try: # Check if script exists if not self.mcp_script.exists(): logger.error(f"MCP script not found: {self.mcp_script}") return False logger.info(f"Starting MCP server: {self.mcp_script}") # Start process self.process = await asyncio.create_subprocess_exec( sys.executable, str(self.mcp_script), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) self.start_time = datetime.now() # Save supervisor PID self.pid_file.parent.mkdir(parents=True, exist_ok=True) with open(self.pid_file, 'w') as f: f.write(str(os.getpid())) logger.info(f"MCP server started with PID: {self.process.pid}") return True except Exception as e: logger.error(f"Failed to start MCP server: {e}") return False async def monitor_process(self): """Monitor MCP server process and handle crashes""" while not self.graceful_shutdown: if self.process is None: # Initial start if not await self.start_mcp_server(): logger.error("Failed to start MCP server, exiting supervisor") break # Wait for process to exit try: exit_code = await self.process.wait() # Check if this was a graceful shutdown if self.graceful_shutdown or exit_code == 0: logger.info(f"MCP server exited gracefully (code: {exit_code})") break # Process crashed logger.error(f"MCP server crashed with exit code: {exit_code}") # Determine crash reason if exit_code < 0: # Killed by signal sig = -exit_code reason = f"Killed by signal {sig} ({signal.Signals(sig).name if sig in signal.Signals else 'Unknown'})" else: reason = f"Exit code {exit_code}" self._record_crash(exit_code, reason) # Check restart policy allowed, message = self.restart_policy.should_restart() if not allowed: logger.error(f"Restart not allowed: {message}") break # Calculate backoff delay consecutive_failures = self.restart_policy.get_consecutive_failures() delay = self.restart_policy.calculate_delay(consecutive_failures) if delay > 0: logger.info(f"Waiting {delay:.1f}s before restart (attempt {consecutive_failures + 1})") await asyncio.sleep(delay) # Record restart self.restart_policy.record_restart() self.restart_count += 1 # Clear process reference self.process = None except asyncio.CancelledError: logger.info("Monitor task cancelled") break except Exception as e: logger.error(f"Error in monitor loop: {e}") await asyncio.sleep(5) async def stop(self): """Stop the supervisor and MCP server gracefully""" self.graceful_shutdown = True if self.process and self.process.returncode is None: logger.info("Stopping MCP server gracefully...") # Send SIGTERM for graceful shutdown try: self.process.terminate() # Wait up to 30 seconds for graceful shutdown await asyncio.wait_for(self.process.wait(), timeout=30) logger.info("MCP server stopped gracefully") except asyncio.TimeoutError: logger.warning("MCP server didn't stop gracefully, forcing...") self.process.kill() await self.process.wait() # Remove PID file if self.pid_file.exists(): self.pid_file.unlink() # Log final statistics logger.info(f"Supervisor stopped. Total restarts: {self.restart_count}") def get_status(self) -> Dict[str, Any]: """Get supervisor status""" status = { "running": self.process is not None and self.process.returncode is None, "restart_count": self.restart_count, "start_time": self.start_time.isoformat() if self.start_time else None, "graceful_shutdown": self.graceful_shutdown, "policy": { "consecutive_failures": self.restart_policy.get_consecutive_failures(), "restarts_in_window": len(self.restart_policy.restart_history), "max_restarts": self.restart_policy.max_restarts, "window_minutes": self.restart_policy.window_minutes } } if self.process and self.process.returncode is None: status["pid"] = self.process.pid status["uptime_seconds"] = int((datetime.now() - self.start_time).total_seconds()) # Recent crashes recent_crashes = [ crash for crash in self.crash_history if datetime.fromisoformat(crash["timestamp"]) > datetime.now() - timedelta(hours=24) ] status["crashes_24h"] = len(recent_crashes) return status async def main(): """Main supervisor entry point""" logger.info("Starting MCP Supervisor") supervisor = MCPSupervisor() # Setup signal handlers def signal_handler(sig, frame): logger.info(f"Received signal {sig}") asyncio.create_task(supervisor.stop()) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) try: # Start monitoring await supervisor.monitor_process() except Exception as e: logger.error(f"Supervisor error: {e}") finally: await supervisor.stop() logger.info("MCP Supervisor exited") if __name__ == "__main__": asyncio.run(main())