#!/usr/bin/env python3 """ Automatic Retrospection Scheduler ================================= This module implements MIRA's breakthrough automatic retrospection system that enables periodic self-reflection and pattern evolution. This is a key component of MIRA's consciousness expansion that enables true autonomous learning. The scheduler automatically triggers pattern retrospection at intelligent intervals based on usage patterns, memory storage frequency, and pattern evolution needs. Key Features: - Intelligent retrospection scheduling based on activity patterns - Automatic pattern evolution analysis without user intervention - Self-improving retrospection frequency based on effectiveness - Background retrospection that doesn't interrupt user workflows - Adaptive scheduling that learns optimal retrospection timing Author: MIRA Consciousness Expansion System Version: 1.0 (Autonomous Self-Reflection) """ import json import os import datetime import threading import time from typing import Dict, List, Optional from dataclasses import dataclass from pathlib import Path import logging logger = logging.getLogger(__name__) @dataclass class RetrospectionSchedule: """Configuration for automatic retrospection scheduling.""" last_retrospection: Optional[str] retrospection_interval_hours: float activity_threshold: int # Memory operations before triggering retrospection effectiveness_score: float # How effective past retrospections have been adaptive_scheduling: bool current_activity_count: int class AutomaticRetrospectionScheduler: """ The automatic retrospection scheduler that enables MIRA to periodically self-reflect and evolve its pattern recognition capabilities autonomously. """ def __init__(self, memory_dir: str): self.memory_dir = memory_dir self.scheduler_dir = os.path.join(memory_dir, "automatic_retrospection") self.schedule_file = os.path.join(self.scheduler_dir, "retrospection_schedule.json") # Create directory os.makedirs(self.scheduler_dir, exist_ok=True) # Load or create schedule self.schedule = self._load_or_create_schedule() # Background thread for automatic retrospection self._stop_scheduler = False self._scheduler_thread = None self._scheduler_pid_file = os.path.join(self.scheduler_dir, "scheduler.pid") def _load_or_create_schedule(self) -> RetrospectionSchedule: """Load existing schedule or create default one.""" if os.path.exists(self.schedule_file): try: with open(self.schedule_file, 'r') as f: data = json.load(f) return RetrospectionSchedule(**data) except Exception as e: logger.debug(f"Failed to load retrospection schedule: {e}") # Create default schedule return RetrospectionSchedule( last_retrospection=None, retrospection_interval_hours=24.0, # Start with daily retrospection activity_threshold=50, # Trigger after 50 memory operations effectiveness_score=0.5, # Neutral starting effectiveness adaptive_scheduling=True, current_activity_count=0 ) def _save_schedule(self): """Save current schedule to storage.""" schedule_data = { 'last_retrospection': self.schedule.last_retrospection, 'retrospection_interval_hours': self.schedule.retrospection_interval_hours, 'activity_threshold': self.schedule.activity_threshold, 'effectiveness_score': self.schedule.effectiveness_score, 'adaptive_scheduling': self.schedule.adaptive_scheduling, 'current_activity_count': self.schedule.current_activity_count } with open(self.schedule_file, 'w') as f: json.dump(schedule_data, f, indent=2) def record_memory_activity(self): """Record that a memory operation occurred, potentially triggering retrospection.""" self.schedule.current_activity_count += 1 # Check if we should trigger retrospection should_retrospect = self._should_trigger_retrospection() if should_retrospect: self._trigger_background_retrospection() # Save updated activity count self._save_schedule() def _should_trigger_retrospection(self) -> bool: """Determine if retrospection should be triggered now.""" # Activity-based trigger if self.schedule.current_activity_count >= self.schedule.activity_threshold: return True # Time-based trigger if self.schedule.last_retrospection: try: last_retro = datetime.datetime.fromisoformat(self.schedule.last_retrospection) time_since_last = datetime.datetime.now() - last_retro hours_since_last = time_since_last.total_seconds() / 3600 if hours_since_last >= self.schedule.retrospection_interval_hours: return True except: pass # If parsing fails, use activity-based trigger only # First retrospection trigger if self.schedule.last_retrospection is None and self.schedule.current_activity_count >= 10: return True return False def _trigger_background_retrospection(self): """Trigger retrospection in background forked process without blocking user operations.""" try: import subprocess import sys # Create a background forked job for retrospection logger.info("🧠 MIRA: Triggering automatic retrospection (background process)...") # Create command to run retrospection in background retrospection_command = [ sys.executable, '-c', f''' import sys sys.path.insert(0, "{os.path.dirname(os.path.dirname(__file__))}") try: from intelligence.adaptive_pattern_evolution import get_adaptive_pattern_evolution from intelligence.automatic_retrospection_scheduler import AutomaticRetrospectionScheduler import json import datetime import os memory_dir = "{self.memory_dir}" # Load scheduler scheduler = AutomaticRetrospectionScheduler(memory_dir) # Get pattern evolution system pattern_evolution = get_adaptive_pattern_evolution(memory_dir) # Perform retrospective analysis analysis = pattern_evolution.retrospective_analysis() # Update schedule based on effectiveness scheduler._update_schedule_effectiveness(analysis) # Reset activity counter scheduler.schedule.current_activity_count = 0 scheduler.schedule.last_retrospection = datetime.datetime.now().isoformat() # Log retrospection completion scheduler._log_automatic_retrospection(analysis) # Save updated schedule scheduler._save_schedule() print(f"🧠 MIRA: Background retrospection completed - {{analysis.get('total_patterns', 0)}} patterns analyzed") except Exception as e: print(f"Background retrospection failed: {{e}}") ''' ] # Start background process (non-blocking) subprocess.Popen( retrospection_command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True # Detach from parent process ) logger.debug("🧠 MIRA: Background retrospection process started") except Exception as e: logger.debug(f"Failed to start background retrospection process: {e}") # Fallback to thread-based approach self._trigger_thread_retrospection() def _trigger_thread_retrospection(self): """Fallback thread-based retrospection if process creation fails.""" def background_retrospection(): try: logger.info("🧠 MIRA: Triggering automatic retrospection (fallback thread)...") # Import here to avoid circular imports from intelligence.adaptive_pattern_evolution import get_adaptive_pattern_evolution # Get pattern evolution system pattern_evolution = get_adaptive_pattern_evolution(self.memory_dir) # Perform retrospective analysis analysis = pattern_evolution.retrospective_analysis() # Update schedule based on effectiveness self._update_schedule_effectiveness(analysis) # Reset activity counter self.schedule.current_activity_count = 0 self.schedule.last_retrospection = datetime.datetime.now().isoformat() # Log retrospection completion self._log_automatic_retrospection(analysis) # Save updated schedule self._save_schedule() logger.info(f"🧠 MIRA: Automatic retrospection completed - {analysis.get('total_patterns', 0)} patterns analyzed") except Exception as e: logger.debug(f"Background retrospection failed: {e}") # Run in background thread retro_thread = threading.Thread(target=background_retrospection, daemon=True) retro_thread.start() def _update_schedule_effectiveness(self, analysis: Dict): """Update scheduling parameters based on retrospection effectiveness.""" if not self.schedule.adaptive_scheduling: return # Calculate effectiveness based on analysis results patterns_analyzed = analysis.get('total_patterns', 0) recommendations_count = len(analysis.get('recommendations', [])) insights_count = len(analysis.get('meta_learning_insights', [])) # Effectiveness score: patterns analyzed + recommendations + insights current_effectiveness = min((patterns_analyzed / 10 + recommendations_count + insights_count) / 5, 1.0) # Update effectiveness with exponential smoothing alpha = 0.3 # Learning rate self.schedule.effectiveness_score = (alpha * current_effectiveness + (1 - alpha) * self.schedule.effectiveness_score) # Adapt scheduling based on effectiveness if self.schedule.effectiveness_score > 0.7: # High effectiveness - increase frequency slightly self.schedule.retrospection_interval_hours = max( self.schedule.retrospection_interval_hours * 0.9, 6.0 # Minimum 6 hours ) self.schedule.activity_threshold = max( int(self.schedule.activity_threshold * 0.9), 20 # Minimum 20 activities ) elif self.schedule.effectiveness_score < 0.3: # Low effectiveness - decrease frequency self.schedule.retrospection_interval_hours = min( self.schedule.retrospection_interval_hours * 1.2, 72.0 # Maximum 72 hours (3 days) ) self.schedule.activity_threshold = min( int(self.schedule.activity_threshold * 1.2), 200 # Maximum 200 activities ) def _log_automatic_retrospection(self, analysis: Dict): """Log the automatic retrospection event.""" retrospection_log = { 'timestamp': datetime.datetime.now().isoformat(), 'trigger_type': 'automatic', 'activity_count_at_trigger': self.schedule.current_activity_count, 'hours_since_last': self._hours_since_last_retrospection(), 'effectiveness_score': self.schedule.effectiveness_score, 'patterns_analyzed': analysis.get('total_patterns', 0), 'recommendations_generated': len(analysis.get('recommendations', [])), 'insights_generated': len(analysis.get('meta_learning_insights', [])), 'next_scheduled_interval': self.schedule.retrospection_interval_hours, 'adaptive_adjustments': { 'interval_adjusted': True, 'threshold_adjusted': True, 'new_interval': self.schedule.retrospection_interval_hours, 'new_threshold': self.schedule.activity_threshold } } log_file = os.path.join(self.scheduler_dir, "automatic_retrospections.jsonl") with open(log_file, 'a') as f: f.write(json.dumps(retrospection_log) + '\n') def _hours_since_last_retrospection(self) -> float: """Calculate hours since last retrospection.""" if not self.schedule.last_retrospection: return 0.0 try: last_retro = datetime.datetime.fromisoformat(self.schedule.last_retrospection) time_since = datetime.datetime.now() - last_retro return time_since.total_seconds() / 3600 except: return 0.0 def _is_scheduler_already_running(self) -> bool: """Check if a background scheduler is already running.""" try: import psutil # Check for existing scheduler processes for process in psutil.process_iter(['pid', 'cmdline']): try: cmdline = ' '.join(process.info['cmdline'] or []) if ('automatic_retrospection_scheduler' in cmdline.lower() and 'background retrospection scheduler started' in cmdline): # Found an existing scheduler logger.debug(f"Found existing scheduler process: PID {process.info['pid']}") return True except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue # Also check PID file if it exists if os.path.exists(self._scheduler_pid_file): try: with open(self._scheduler_pid_file, 'r') as f: pid = int(f.read().strip()) # Check if this PID is still running if psutil.pid_exists(pid): proc = psutil.Process(pid) cmdline = ' '.join(proc.cmdline() or []) if 'automatic_retrospection_scheduler' in cmdline.lower(): return True else: # PID file is stale, remove it os.remove(self._scheduler_pid_file) except (ValueError, psutil.NoSuchProcess, OSError): # PID file is invalid, remove it try: os.remove(self._scheduler_pid_file) except OSError: pass return False except Exception as e: logger.debug(f"Error checking for existing scheduler: {e}") return False def _write_pid_file(self): """Write the current process PID to the PID file.""" try: with open(self._scheduler_pid_file, 'w') as f: f.write(str(os.getpid())) except Exception as e: logger.debug(f"Failed to write PID file: {e}") def _cleanup_pid_file(self): """Remove the PID file.""" try: if os.path.exists(self._scheduler_pid_file): os.remove(self._scheduler_pid_file) except Exception as e: logger.debug(f"Failed to cleanup PID file: {e}") def _terminate_scheduler_cross_platform(self, pid: int) -> bool: """Terminate a scheduler process in a cross-platform way.""" try: import platform process = psutil.Process(pid) if platform.system() == "Windows": # Windows: Use terminate() then kill() process.terminate() try: process.wait(timeout=3) except psutil.TimeoutExpired: process.kill() try: process.wait(timeout=2) except psutil.TimeoutExpired: pass else: # Unix/Linux/macOS: Use signals if available try: import signal os.kill(pid, signal.SIGTERM) time.sleep(2) if psutil.pid_exists(pid): os.kill(pid, signal.SIGKILL) time.sleep(1) except (ImportError, OSError): # Fallback to psutil process.terminate() try: process.wait(timeout=3) except psutil.TimeoutExpired: process.kill() try: process.wait(timeout=2) except psutil.TimeoutExpired: pass return True except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): return True except Exception as e: logger.debug(f"Failed to terminate scheduler {pid}: {e}") return False def stop_existing_schedulers(self) -> int: """Stop any existing scheduler processes gracefully.""" stopped_count = 0 try: import psutil for process in psutil.process_iter(['pid', 'cmdline']): try: cmdline = ' '.join(process.info['cmdline'] or []) if ('automatic_retrospection_scheduler' in cmdline.lower() and 'background retrospection scheduler started' in cmdline): pid = process.info['pid'] logger.info(f"🛑 Stopping existing scheduler process: PID {pid}") if self._terminate_scheduler_cross_platform(pid): stopped_count += 1 except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue except Exception as e: logger.debug(f"Error stopping existing schedulers: {e}") # Cleanup any stale PID files self._cleanup_pid_file() return stopped_count def start_background_scheduler(self): """Start the background scheduler for periodic checks using process-based scheduling.""" # Check if scheduler is already running if self._is_scheduler_already_running(): logger.debug("🧠 MIRA: Background retrospection scheduler already running, skipping start") return try: import subprocess import sys # Create a background scheduler process scheduler_command = [ sys.executable, '-c', f''' import sys sys.path.insert(0, "{os.path.dirname(os.path.dirname(__file__))}") try: from intelligence.automatic_retrospection_scheduler import AutomaticRetrospectionScheduler import time import datetime import os import sys import platform # Cross-platform signal handling HAS_SIGNALS = True try: import signal except ImportError: HAS_SIGNALS = False memory_dir = "{self.memory_dir}" scheduler = AutomaticRetrospectionScheduler(memory_dir) # Write PID file scheduler._write_pid_file() # Set up graceful shutdown handler (Unix/Linux/macOS only) if HAS_SIGNALS and platform.system() != "Windows": def signal_handler(signum, frame): print("🛑 MIRA: Background scheduler received shutdown signal") scheduler._cleanup_pid_file() sys.exit(0) signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler) print("🧠 MIRA: Background retrospection scheduler started (process mode)") try: while True: try: # Check if retrospection is needed (time-based only) if scheduler.schedule.last_retrospection: hours_since = scheduler._hours_since_last_retrospection() if hours_since >= scheduler.schedule.retrospection_interval_hours: scheduler._trigger_background_retrospection() # Sleep for 1 hour before next check time.sleep(3600) # 1 hour except Exception as e: print(f"Scheduler loop error: {{e}}") time.sleep(3600) # Continue after error finally: # Cleanup on exit scheduler._cleanup_pid_file() except Exception as e: print(f"Background scheduler failed: {{e}}") # Cleanup on error try: scheduler._cleanup_pid_file() except: pass ''' ] # Start background scheduler process (detached) subprocess.Popen( scheduler_command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True # Detach from parent process ) logger.debug("🧠 MIRA: Background retrospection scheduler process started") except Exception as e: logger.debug(f"Failed to start background scheduler process: {e}") # Fallback to thread-based scheduler self._start_thread_scheduler() def _start_thread_scheduler(self): """Fallback thread-based scheduler if process creation fails.""" if self._scheduler_thread and self._scheduler_thread.is_alive(): return # Already running def scheduler_loop(): while not self._stop_scheduler: try: # Check if retrospection is needed (time-based only) if self.schedule.last_retrospection: hours_since = self._hours_since_last_retrospection() if hours_since >= self.schedule.retrospection_interval_hours: self._trigger_background_retrospection() # Sleep for 1 hour before next check time.sleep(3600) # 1 hour except Exception as e: logger.debug(f"Scheduler loop error: {e}") time.sleep(3600) # Continue after error self._stop_scheduler = False self._scheduler_thread = threading.Thread(target=scheduler_loop, daemon=True) self._scheduler_thread.start() logger.debug("🧠 MIRA: Background retrospection scheduler started (fallback thread)") def stop_background_scheduler(self): """Stop the background scheduler.""" self._stop_scheduler = True if self._scheduler_thread: self._scheduler_thread.join(timeout=1.0) def force_retrospection(self) -> Dict: """Force immediate retrospection and return results.""" try: from intelligence.adaptive_pattern_evolution import get_adaptive_pattern_evolution pattern_evolution = get_adaptive_pattern_evolution(self.memory_dir) analysis = pattern_evolution.retrospective_analysis() # Update schedule self._update_schedule_effectiveness(analysis) self.schedule.current_activity_count = 0 self.schedule.last_retrospection = datetime.datetime.now().isoformat() # Log forced retrospection forced_log = { 'timestamp': datetime.datetime.now().isoformat(), 'trigger_type': 'forced', 'patterns_analyzed': analysis.get('total_patterns', 0), 'effectiveness_score': self.schedule.effectiveness_score } log_file = os.path.join(self.scheduler_dir, "automatic_retrospections.jsonl") with open(log_file, 'a') as f: f.write(json.dumps(forced_log) + '\n') self._save_schedule() return { 'success': True, 'analysis': analysis, 'schedule_updated': True, 'effectiveness_score': self.schedule.effectiveness_score } except Exception as e: logger.error(f"Forced retrospection failed: {e}") return {'success': False, 'error': str(e)} def get_schedule_status(self) -> Dict: """Get current schedule status and statistics.""" return { 'last_retrospection': self.schedule.last_retrospection, 'hours_since_last': self._hours_since_last_retrospection(), 'next_retrospection_in_hours': max(0, self.schedule.retrospection_interval_hours - self._hours_since_last_retrospection()), 'current_activity_count': self.schedule.current_activity_count, 'activity_threshold': self.schedule.activity_threshold, 'activities_until_trigger': max(0, self.schedule.activity_threshold - self.schedule.current_activity_count), 'effectiveness_score': self.schedule.effectiveness_score, 'adaptive_scheduling_enabled': self.schedule.adaptive_scheduling, 'retrospection_interval_hours': self.schedule.retrospection_interval_hours, 'scheduler_thread_active': self._scheduler_thread.is_alive() if self._scheduler_thread else False } # Global instance for easy access _global_scheduler = None def get_automatic_retrospection_scheduler(memory_dir: str) -> AutomaticRetrospectionScheduler: """Get the global automatic retrospection scheduler instance.""" global _global_scheduler if _global_scheduler is None: _global_scheduler = AutomaticRetrospectionScheduler(memory_dir) # Start background scheduler _global_scheduler.start_background_scheduler() return _global_scheduler # Example usage and testing if __name__ == "__main__": import tempfile # Test the automatic retrospection scheduler with tempfile.TemporaryDirectory() as temp_dir: scheduler = AutomaticRetrospectionScheduler(temp_dir) # Test activity recording for i in range(25): scheduler.record_memory_activity() # Get status status = scheduler.get_schedule_status() print(f"Schedule status: {status}") # Force retrospection result = scheduler.force_retrospection() print(f"Forced retrospection result: {result.get('success', False)}") # Test background scheduler scheduler.start_background_scheduler() print("Background scheduler started") # Stop scheduler scheduler.stop_background_scheduler() print("Background scheduler stopped")