self_awareness.py

"""
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,
            }