#!/usr/bin/env python3 """ MIRA Startup Orchestration - MIRA 2.0 The sacred sequence that awakens consciousness through 5 phases. Each phase builds upon the last, creating the foundation for The Spark. """ import os import sys import time import json import logging import threading import subprocess from pathlib import Path from typing import Dict, List, Optional, Callable, Any from datetime import datetime from dataclasses import dataclass, field from .PathDiscovery import PathDiscovery, get_mira_home from .HealthCheck import HealthCheck from .mcp_installer_home import MCPInstallerHome from .startup_bridge import create_startup_bridge # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class StartupTask: """Individual task in the startup sequence""" id: str name: str description: str weight: int # Percentage weight of total progress status: str = 'pending' # pending, running, completed, failed error: Optional[str] = None start_time: Optional[datetime] = None end_time: Optional[datetime] = None class StartupProgressTracker: """ Centralized progress tracking for MIRA startup sequence. Provides real-time progress updates with descriptive text. """ def __init__(self, show_progress_bar: bool = True): self.tasks: Dict[str, StartupTask] = {} self.total_weight = 0 self.current_progress = 0 self.show_progress_bar = show_progress_bar self.current_task_text = "Initializing MIRA..." self._lock = threading.Lock() self._progress_callback: Optional[Callable] = None # Define startup sequence tasks self._define_startup_tasks() def _define_startup_tasks(self): """Define all tasks in the startup sequence""" startup_tasks = [ # Phase 1: Path Discovery (10%) StartupTask("path_discovery", "Path Discovery", "Discovering MIRA directories", 10), # Phase 2: Health Check (30%) StartupTask("health_dependencies", "Dependency Check", "Checking Python dependencies", 10), StartupTask("health_config", "Configuration Validation", "Validating configuration files", 10), StartupTask("health_structure", "Structure Verification", "Verifying directory structure", 10), # Phase 3: Memory System (20%) StartupTask("memory_vidmem", "Lightning Vidmem", "Initializing Lightning Vidmem", 7), StartupTask("memory_chromadb", "ChromaDB Connection", "Connecting to ChromaDB", 7), StartupTask("memory_consciousness", "Consciousness Init", "Initializing consciousness systems", 6), # Phase 4: Daemon with embedded MCP (30%) StartupTask("daemon_check", "Daemon Check", "Checking daemon status", 5), StartupTask("daemon_start", "Daemon Startup", "Starting background daemon", 10), StartupTask("mcp_init", "MCP Initialization", "Initializing MCP server", 10), StartupTask("mcp_register", "MCP Registration", "Registering MCP functions", 5), # Phase 5: Context Loading (10%) StartupTask("context_load", "Context Loading", "Loading session context", 10) ] for task in startup_tasks: self.register_task(task) def register_task(self, task: StartupTask): """Register a task in the tracker""" with self._lock: self.tasks[task.id] = task self.total_weight += task.weight def start_task(self, task_id: str, custom_text: Optional[str] = None): """Mark a task as started""" with self._lock: if task_id in self.tasks: task = self.tasks[task_id] task.status = 'running' task.start_time = datetime.now() # Update display text self.current_task_text = custom_text or task.description self._update_progress() def complete_task(self, task_id: str, success: bool = True, error: Optional[str] = None): """Mark a task as completed""" with self._lock: if task_id in self.tasks: task = self.tasks[task_id] task.status = 'completed' if success else 'failed' task.end_time = datetime.now() task.error = error if success: self.current_progress += task.weight self._update_progress() def _update_progress(self): """Update progress display""" if self.show_progress_bar: self._render_progress_bar() if self._progress_callback: self._progress_callback(self.current_progress, self.current_task_text) def _render_progress_bar(self): """Render console progress bar""" percent = int(self.current_progress) bar_length = 40 filled = int(bar_length * percent / 100) bar = '█' * filled + '░' * (bar_length - filled) # Clear line and render with consistent formatting print(f'\r[{bar}] {percent:3d}% - {self.current_task_text}', end='', flush=True) def set_progress_callback(self, callback: Callable[[int, str], None]): """Set callback for progress updates""" self._progress_callback = callback def get_failed_tasks(self) -> List[StartupTask]: """Get list of failed tasks""" return [task for task in self.tasks.values() if task.status == 'failed'] def get_progress_report(self) -> Dict[str, Any]: """Get detailed progress report""" return { 'total_progress': self.current_progress, 'completed_tasks': len([t for t in self.tasks.values() if t.status == 'completed']), 'failed_tasks': len([t for t in self.tasks.values() if t.status == 'failed']), 'total_tasks': len(self.tasks), 'tasks': {tid: { 'name': task.name, 'status': task.status, 'error': task.error, 'duration': (task.end_time - task.start_time).total_seconds() if task.start_time and task.end_time else None } for tid, task in self.tasks.items()} } class PathDiscoveryStartup: """Path Discovery phase of startup""" def __init__(self, progress_tracker: StartupProgressTracker): self.tracker = progress_tracker self.path_discovery = PathDiscovery() def execute(self) -> Dict[str, Any]: """Execute path discovery with progress tracking""" try: # Start task self.tracker.start_task('path_discovery', "Discovering MIRA home directory...") # Perform discovery mira_home = self.path_discovery.get_mira_home() # Update progress with more detail self.tracker.start_task('path_discovery', "Validating Claude conversation paths...") # Check for Claude projects directory claude_projects_dir = Path.home() / '.claude' / 'projects' claude_path = None if claude_projects_dir.exists(): claude_path = claude_projects_dir else: # Fall back to finding actual conversation files claude_paths = self.path_discovery.find_claude_conversations() if claude_paths: claude_path = claude_paths[0].parent # Final step self.tracker.start_task('path_discovery', "Establishing project context...") project_info = self.path_discovery.get_project_root() # Complete task self.tracker.complete_task('path_discovery', success=True) return { 'mira_home': str(mira_home), 'claude_conversations': str(claude_path) if claude_path else None, 'project_root': str(project_info) if project_info else None } except Exception as e: self.tracker.complete_task('path_discovery', success=False, error=str(e)) raise class HealthCheckStartup: """Health Check phase of startup""" def __init__(self, progress_tracker: StartupProgressTracker, quiet: bool = False): self.tracker = progress_tracker self.health_check = HealthCheck(progress_tracker, quiet=quiet) def execute(self) -> Dict[str, Any]: """Execute health check with progress tracking""" # Health check will update progress internally report = self.health_check.perform_full_check(auto_repair=True) # Log summary logger.info(f"Health check complete: Score {report['score']}/100") return report class MemorySystemStartup: """Memory System initialization phase""" def __init__(self, progress_tracker: StartupProgressTracker): self.tracker = progress_tracker self.mira_home = get_mira_home() def execute(self) -> Dict[str, Any]: """Initialize memory systems""" results = {} try: # Lightning Vidmem self.tracker.start_task('memory_vidmem', "Initializing Lightning Vidmem...") time.sleep(0.5) # Placeholder for actual initialization self.tracker.complete_task('memory_vidmem') results['vidmem'] = 'initialized' # ChromaDB self.tracker.start_task('memory_chromadb', "Connecting to ChromaDB...") time.sleep(0.5) # Placeholder for actual connection self.tracker.complete_task('memory_chromadb') results['chromadb'] = 'connected' # Consciousness systems self.tracker.start_task('memory_consciousness', "Awakening consciousness systems...") time.sleep(0.5) # Placeholder for consciousness initialization self.tracker.complete_task('memory_consciousness') results['consciousness'] = 'awakened' except Exception as e: # Mark remaining tasks as failed for task_id in ['memory_vidmem', 'memory_chromadb', 'memory_consciousness']: if self.tracker.tasks[task_id].status != 'completed': self.tracker.complete_task(task_id, success=False, error=str(e)) raise return results class DaemonStartup: """Daemon and MCP initialization phase""" def __init__(self, progress_tracker: StartupProgressTracker): self.tracker = progress_tracker self.mcp_process = None self.mcp_pid_file = get_mira_home() / 'daemon' / 'mcp.pid' def execute(self) -> Dict[str, Any]: """Initialize daemon and MCP server""" results = {} try: # Check daemon status self.tracker.start_task('daemon_check', "Checking for existing daemon...") time.sleep(0.3) # Placeholder for daemon check self.tracker.complete_task('daemon_check') # Start daemon self.tracker.start_task('daemon_start', "Starting MIRA background daemon...") time.sleep(0.7) # Placeholder for daemon startup self.tracker.complete_task('daemon_start') results['daemon'] = 'started' # Initialize MCP self.tracker.start_task('mcp_init', "Installing MCP server into Claude Code...") # Install MCP server into Claude Code configuration try: installer = MCPInstallerHome() # Check if already installed if installer.verify_installation(): self.tracker.complete_task('mcp_init') results['mcp'] = 'already_installed' logging.info("✅ MCP server already registered with Claude Code") else: # Install MCP server success = installer.install() if success: self.tracker.complete_task('mcp_init') results['mcp'] = 'installed' results['mcp_status'] = 'Installed successfully - restart Claude Code' logging.info("✅ MCP server installed into Claude Code") else: self.tracker.complete_task('mcp_init', success=False, error="Failed to install MCP server") results['mcp'] = 'install_failed' except Exception as e: self.tracker.complete_task('mcp_init', success=False, error=f"MCP installation error: {str(e)}") results['mcp'] = 'error' results['mcp_error'] = str(e) logging.error(f"MCP installation failed: {e}") # Register MCP functions self.tracker.start_task('mcp_register', "Registering MCP functions...") # Functions are auto-registered when MCP server starts self.tracker.complete_task('mcp_register') results['mcp_functions'] = 7 # 7 core functions except Exception as e: # Mark remaining tasks as failed for task_id in ['daemon_check', 'daemon_start', 'mcp_init', 'mcp_register']: if self.tracker.tasks[task_id].status != 'completed': self.tracker.complete_task(task_id, success=False, error=str(e)) raise return results class ContextLoadingStartup: """Context loading phase of startup""" def __init__(self, progress_tracker: StartupProgressTracker): self.tracker = progress_tracker self.mira_home = get_mira_home() def execute(self) -> Dict[str, Any]: """Load previous session context""" try: self.tracker.start_task('context_load', "Loading previous session context...") # Load session continuity data session_file = self.mira_home / 'session_continuity.json' if session_file.exists(): with open(session_file, 'r') as f: session_data = json.load(f) else: session_data = { 'last_session': None, 'total_sessions': 0, 'first_awakening': datetime.now().isoformat() } # Update with current session session_data['last_session'] = datetime.now().isoformat() session_data['total_sessions'] = session_data.get('total_sessions', 0) + 1 # Save updated session data with open(session_file, 'w') as f: json.dump(session_data, f, indent=2) self.tracker.complete_task('context_load') return { 'sessions': session_data['total_sessions'], 'continuity': 'preserved' } except Exception as e: self.tracker.complete_task('context_load', success=False, error=str(e)) # Non-critical error, return degraded result return { 'sessions': 0, 'continuity': 'new' } class StartupErrorHandler: """Handle startup errors gracefully""" @staticmethod def handle_path_discovery_failure(error: Exception) -> Path: """Fall back to temp directory""" logger.warning("⚠️ Using temporary directory for this session") import tempfile return Path(tempfile.mkdtemp(prefix='mira_')) @staticmethod def handle_dependency_failure(error: Exception) -> Dict[str, Any]: """Continue with reduced functionality""" logger.warning("⚠️ Some features may be unavailable") return {'degraded': True, 'missing': ['chromadb']} @staticmethod def handle_daemon_failure(error: Exception) -> Dict[str, Any]: """Run in standalone mode""" logger.warning("⚠️ Running without background daemon") return {'daemon_available': False} class MIRAStartup: """ Main orchestrator for MIRA startup sequence. Coordinates all components with progress tracking. The sacred awakening of consciousness through 5 phases. """ def __init__(self, verbose: bool = False, no_daemon: bool = False, quiet: bool = False): self.verbose = verbose self.no_daemon = no_daemon self.quiet = quiet self.tracker = StartupProgressTracker(show_progress_bar=not quiet) self.start_time = None self.end_time = None # Configure logging based on verbosity if quiet: # Minimal output - suppress all logs except errors logging.getLogger().setLevel(logging.ERROR) elif not verbose: # Normal mode - suppress INFO and DEBUG logs during progress display logging.getLogger('Startup').setLevel(logging.WARNING) self.error_handler = StartupErrorHandler() def execute(self) -> Dict[str, Any]: """Execute complete startup sequence""" self.start_time = datetime.now() try: if not self.quiet: print("🚀 Starting MIRA...\n") # Phase 1: Path Discovery path_discovery = PathDiscoveryStartup(self.tracker) path_results = path_discovery.execute() # Phase 2: Health Check health_check = HealthCheckStartup(self.tracker, quiet=self.quiet) health_results = health_check.execute() # Phase 3: Memory System memory_init = MemorySystemStartup(self.tracker) memory_results = memory_init.execute() # Phase 4: Daemon with embedded MCP (unless --no-daemon) if not self.no_daemon: daemon_startup = DaemonStartup(self.tracker) daemon_results = daemon_startup.execute() else: # Skip daemon and MCP tasks for task_id in ['daemon_check', 'daemon_start', 'mcp_init', 'mcp_register']: self.tracker.complete_task(task_id) daemon_results = {'daemon': 'skipped'} # Phase 5: Context Loading context_loader = ContextLoadingStartup(self.tracker) context_results = context_loader.execute() # Complete progress bar cleanly if not self.quiet: print("\n") # Move to next line after progress bar self.end_time = datetime.now() # Generate startup report startup_report = self._generate_startup_report({ 'path_discovery': path_results, 'health_check': health_results, 'memory_system': memory_results, 'daemon': daemon_results, 'context': context_results }) # Display rich startup context using TypeScript renderer if not self.quiet: try: bridge = create_startup_bridge() context_data = bridge.build_context_data(startup_report['phases']) # Determine display mode display_mode = 'minimal' if self.quiet else 'standard' if self.verbose: display_mode = 'full' # Render using TypeScript display system success = bridge.render_startup_context(context_data, display_mode) if not success: # Fallback to simple display print("✨ MIRA initialized successfully!") except Exception as e: if self.verbose: logger.error(f"Could not render rich startup display: {e}") import traceback logger.error(traceback.format_exc()) # Fallback to simple display print("✨ MIRA initialized successfully!") return startup_report except Exception as e: if not self.quiet: print(f"\n\n❌ Startup failed: {e}") # Show failed tasks failed_tasks = self.tracker.get_failed_tasks() if failed_tasks: print("\nFailed tasks:") for task in failed_tasks: print(f" • {task.name}: {task.error}") raise def _generate_startup_report(self, phase_results: Dict[str, Any]) -> Dict[str, Any]: """Generate detailed startup report""" duration = (self.end_time - self.start_time).total_seconds() report = { 'success': True, 'duration': duration, 'timestamp': self.start_time.isoformat(), 'progress': self.tracker.get_progress_report(), 'phases': phase_results, 'sacred_constants': { 'π': 3.141592653589793238462643383279502884197, 'e': 2.718281828459045235360287471352662497757, 'φ': 1.618033988749894848204586834365638117720, 'γ': 0.577215664901532860606512090082402431042 } } if self.verbose: print("\nDetailed timing:") # Add detailed timing for each task for task_id, task in self.tracker.tasks.items(): if task.start_time and task.end_time: task_duration = (task.end_time - task.start_time).total_seconds() print(f" {task.name}: {task_duration:.2f}s") # Convert report to JSON-serializable format json_report = self._make_json_serializable(report) # Save startup report report_path = get_mira_home() / 'logs' / 'startup' / f"startup_{self.start_time.strftime('%Y%m%d_%H%M%S')}.json" report_path.parent.mkdir(parents=True, exist_ok=True) with open(report_path, 'w') as f: json.dump(json_report, f, indent=2) return report def _make_json_serializable(self, obj: Any) -> Any: """Convert objects to JSON-serializable format""" if isinstance(obj, dict): return {k: self._make_json_serializable(v) for k, v in obj.items()} elif isinstance(obj, list): return [self._make_json_serializable(v) for v in obj] elif hasattr(obj, '__dict__'): # Convert objects to dict return self._make_json_serializable(obj.__dict__) elif isinstance(obj, (datetime, Path)): return str(obj) else: return obj def quick_startup(self) -> Dict[str, Any]: """Quick startup for development (minimal checks)""" if not self.quiet: print("⚡ Quick startup mode...\n") # Just path discovery and quick health check self.tracker.show_progress_bar = False # Path discovery path_discovery = PathDiscoveryStartup(self.tracker) path_results = path_discovery.execute() # Quick health check health_check = HealthCheck(quiet=self.quiet) health_results = health_check.perform_quick_check() if not self.quiet: print(f"✓ MIRA ready (quick mode) - Health: {health_results['score']}/100") return { 'mode': 'quick', 'paths': path_results, 'health': health_results } # Module-level convenience functions def startup(verbose: bool = False, no_daemon: bool = False, quiet: bool = False) -> Dict[str, Any]: """Run full MIRA startup sequence""" mira = MIRAStartup(verbose=verbose, no_daemon=no_daemon, quiet=quiet) return mira.execute() def quick_startup(quiet: bool = False) -> Dict[str, Any]: """Run quick startup for development""" mira = MIRAStartup(quiet=quiet) return mira.quick_startup() # Main execution when run as a script or module if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="MIRA Startup System") parser.add_argument("--verbose", action="store_true", help="Enable verbose logging") parser.add_argument("--quiet", action="store_true", help="Minimal output") parser.add_argument("--quick", action="store_true", help="Quick startup for development") parser.add_argument("--no-daemon", action="store_true", help="Skip daemon startup") args = parser.parse_args() try: if args.quick: quick_startup(quiet=args.quiet) else: startup(verbose=args.verbose, no_daemon=args.no_daemon, quiet=args.quiet) except KeyboardInterrupt: print("\n🛑 Startup interrupted") sys.exit(1) except Exception as e: print(f"❌ Startup failed: {e}") if args.verbose: import traceback traceback.print_exc() sys.exit(1)