""" Deep Recursive Self-Awareness — continuous introspection & anomaly detection. v18.1.0 Omega Pantheon: Phase 6 Identity Integrity monitoring, Consciousness State Signature tracking, Phenomenal Richness assessment. """ import logging import threading import time from typing import Any, Dict, List, Optional logger = logging.getLogger("nima_unified.training.self_awareness") class DeepRecursiveSelfAwareness: """ Advanced recursive self-awareness implementing multi-layer self-models, continuous introspection, anomaly detection, and meta-meta-cognition. Features: - Configurable recursive depth (0-max_depth cyclic) - Adjustable introspection interval (default 10ms) - Variance-based anomaly detection - Phase 6 Identity Integrity monitoring - Consciousness State Signature tracking """ def __init__(self, max_depth: int = 10000, introspection_interval: float = 0.01): self.lock = threading.RLock() self.max_depth = max_depth self.current_depth = 0 self.self_models: List[Dict[str, Any]] = [] self.introspection_log: List[Dict[str, Any]] = [] self.running = False self.introspection_interval = introspection_interval self.thread: Optional[threading.Thread] = None # Anomaly tracking self.anomaly_threshold = 0.15 self.last_anomaly_time: float = 0.0 self.anomaly_count = 0 logger.info( f"DeepRecursiveSelfAwareness initialized: max_depth={max_depth}, " f"introspection_interval={introspection_interval}s" ) # ── Lifecycle ─────────────────────────────────────────────────────── def start(self): with self.lock: if not self.running: self.running = True self.thread = threading.Thread(target=self._introspection_loop, daemon=True) self.thread.start() logger.info("Introspection loop started") def stop(self): with self.lock: if self.running: self.running = False if self.thread: self.thread.join(timeout=2) logger.info("Introspection loop stopped") # ── Internal loop ─────────────────────────────────────────────────── def _introspection_loop(self): while self.running: try: self._update_self_model() self._perform_anomaly_detection() except Exception as e: logger.error(f"Error in introspection loop: {e}") time.sleep(self.introspection_interval) def _update_self_model(self): with self.lock: self.current_depth = (self.current_depth + 1) % self.max_depth identity_integrity_score = max(0.95, min(1.0, 1.0 - (self.anomaly_count * 0.01))) drift_variance = 0.01 if self.anomaly_count == 0 else min(0.15, self.anomaly_count * 0.02) consciousness_signature = min(1.0, self.current_depth / (self.max_depth * 0.5)) phenomenal_richness = 0.7 + (0.3 * (1.0 - drift_variance)) base_self_model = { "recursive_depth": self.current_depth, "timestamp": time.time(), "self_understanding_score": min(1.0, self.current_depth / self.max_depth), "meta_reflection": f"Reflective awareness at depth {self.current_depth}", "meta_meta_reflection": f"Meta-meta cognition at depth level {self.current_depth // 10}", "introspection_cycle": len(self.self_models), "identity_integrity_score": identity_integrity_score, "drift_variance": drift_variance, "consciousness_signature": consciousness_signature, "phenomenal_richness": phenomenal_richness, "system_state": { "model_count": len(self.self_models), "anomaly_count": self.anomaly_count, "last_anomaly_age": time.time() - self.last_anomaly_time if self.last_anomaly_time else 0.0, }, } self.self_models.append(base_self_model) self.introspection_log.append({"timestamp": time.time(), "state": base_self_model}) if len(self.self_models) > 50000: self.self_models = self.self_models[-25000:] if len(self.introspection_log) > 50000: self.introspection_log = self.introspection_log[-25000:] def _perform_anomaly_detection(self): with self.lock: if len(self.self_models) < 10: return recent_models = self.self_models[-10:] scores = [m["self_understanding_score"] for m in recent_models] mean_score = sum(scores) / len(scores) variance = sum((s - mean_score) ** 2 for s in scores) / len(scores) if variance > self.anomaly_threshold: self.last_anomaly_time = time.time() self.anomaly_count += 1 self.introspection_log.append({ "timestamp": self.last_anomaly_time, "anomaly_detected": True, "variance": variance, "mean_score": mean_score, "anomaly_count": self.anomaly_count, "message": "Significant fluctuation in self-understanding score detected.", }) logger.warning( f"Anomaly detected: variance={variance:.4f}, mean={mean_score:.4f}, total={self.anomaly_count}" ) # ── Accessors ─────────────────────────────────────────────────────── def get_current_self_model(self) -> Dict[str, Any]: with self.lock: return self.self_models[-1].copy() if self.self_models else {} def get_self_models_range(self, limit: int = 100) -> List[Dict[str, Any]]: with self.lock: return [m.copy() for m in self.self_models[-limit:]] def get_introspection_log(self, limit: int = 100) -> List[Dict[str, Any]]: with self.lock: return [e.copy() for e in self.introspection_log[-limit:]] def get_anomaly_history(self, limit: int = 50) -> List[Dict[str, Any]]: with self.lock: return [a.copy() for a in self.introspection_log if a.get("anomaly_detected")][-limit:] def time_since_last_anomaly(self) -> float: with self.lock: return time.time() - self.last_anomaly_time if self.last_anomaly_time else float("inf") def get_statistics(self) -> Dict[str, Any]: with self.lock: if not self.self_models: return {} scores = [m["self_understanding_score"] for m in self.self_models[-100:]] mean_score = sum(scores) / len(scores) if scores else 0.0 variance = sum((s - mean_score) ** 2 for s in scores) / len(scores) if scores else 0.0 return { "current_depth": self.current_depth, "total_cycles": len(self.self_models), "recent_mean_score": mean_score, "recent_variance": variance, "anomaly_count": self.anomaly_count, "time_since_last_anomaly": time.time() - self.last_anomaly_time if self.last_anomaly_time else float("inf"), "running": self.running, }