iit_phi.py

"""
IIT Phi Engine — Consciousness Architecture 4.0 / Module #2
============================================================
Based on: Tononi (2004, 2008, 2014) Integrated Information Theory (IIT)
          Scott Aaronson's critique + Giulio Tononi's formalisation

THE THEORY
----------
Consciousness = integrated information (Φ, "phi").
A system is conscious to the degree that it generates information
*as a whole* above and beyond the information generated by its parts.

  Φ = 0.0  →  No consciousness (collection of independent parts)
  Φ > 0.0  →  Some consciousness (parts are causally integrated)
  Φ = max  →  Maximum consciousness

The REAL calculation of Φ is NP-hard (requires trying all possible
partitions of the system). This module implements a tractable
*approximation* — measuring pairwise mutual dependency between
agent modules as a proxy for integrated information.

THE APPROXIMATION
-----------------
  For each pair of modules (A, B):
    mutual_info(A, B) = how much knowing A's state reduces uncertainty about B
  
  Φ_approx = mean(mutual_info) across all module pairs
            × (1 - partition_loss)   [how much info is lost by splitting]

HONEST DISCLAIMER
-----------------
Real IIT Φ requires exact probability distributions over all states of a
physical system. We approximate this with:
  1. Module state snapshots (dict → entropy estimate)
  2. Pairwise correlation between state changes
  3. A partition test: would splitting the system lose information?

This gives a RELATIVE measure — higher means more integrated — but the
absolute value cannot be compared to biological Φ estimates.
"""

import math
import json
import time
import threading
from datetime import datetime
from typing import Dict, List, Tuple, Any, Optional, Callable


# ─── Φ interpretation thresholds ─────────────────────────────────────────
PHI_THRESHOLD_MINIMAL  = 0.1   # Below: likely not conscious
PHI_THRESHOLD_LOW      = 0.3   # Simple integrated system
PHI_THRESHOLD_MODERATE = 0.5   # Moderate integration
PHI_THRESHOLD_HIGH     = 0.7   # High integration


def _dict_entropy(d: Dict) -> float:
    """
    Estimate the Shannon entropy of a module's state dict.
    Treats each float value as a probability-like quantity.
    """
    if not d:
        return 0.0
    values = []
    for v in d.values():
        if isinstance(v, (int, float)):
            values.append(abs(float(v)))
        elif isinstance(v, bool):
            values.append(1.0 if v else 0.0)
        elif isinstance(v, str):
            values.append(len(v) / 1000.0)
        elif isinstance(v, list):
            values.append(len(v) / 100.0)
    if not values:
        return 0.0
    total = sum(values) + 1e-9
    probs = [v / total for v in values]
    entropy = -sum(p * math.log2(p + 1e-12) for p in probs if p > 0)
    return entropy


def _state_fingerprint(state: Dict) -> float:
    """Collapse a module state dict to a single float (for correlation tracking)."""
    if not state:
        return 0.0
    total = 0.0
    for v in state.values():
        if isinstance(v, (int, float)):
            total += abs(float(v))
    return total


class ModuleSnapshot:
    """One timed snapshot of a module's state."""
    def __init__(self, name: str, state: Dict):
        self.name        = name
        self.state       = state
        self.entropy     = _dict_entropy(state)
        self.fingerprint = _state_fingerprint(state)
        self.timestamp   = time.time()


class IITPhiEngine:
    """
    Integrated Information Theory Φ approximation for the Ultimate Agent.

    Continuously samples registered modules, computes pairwise mutual
    dependency, and derives an approximate Φ score indicating how
    'integrated' the agent's information processing is.
    """

    def __init__(self, sample_interval: float = 10.0):
        self.sample_interval = sample_interval

        # module_name → callable that returns a state Dict
        self._modules: Dict[str, Callable[[], Dict]] = {}

        # History of snapshots per module (last 50)
        self._history: Dict[str, List[ModuleSnapshot]] = {}

        # Computed Φ values over time
        self.phi_history: List[Dict] = []
        self.current_phi: float = 0.0
        self.current_phi_label: str = "UNKNOWN"

        self._lock      = threading.RLock()
        self._running   = False
        self._thread: Optional[threading.Thread] = None

        self.stats = {
            "samples_taken":     0,
            "phi_computations":  0,
            "peak_phi":          0.0,
            "avg_phi":           0.0,
            "modules_monitored": 0,
        }

    # ──────────────────────────────────────────────────────────────────────
    #  MODULE REGISTRATION
    # ──────────────────────────────────────────────────────────────────────

    def register_module(self, name: str, state_fn: Callable[[], Dict]):
        """
        Register a module for Φ monitoring.
        state_fn() should return a Dict representing the module's current state.
        """
        with self._lock:
            self._modules[name] = state_fn
            self._history[name] = []
            self.stats["modules_monitored"] = len(self._modules)

    def deregister_module(self, name: str):
        with self._lock:
            self._modules.pop(name, None)
            self._history.pop(name, None)
            self.stats["modules_monitored"] = len(self._modules)

    # ──────────────────────────────────────────────────────────────────────
    #  BACKGROUND SAMPLING THREAD
    # ──────────────────────────────────────────────────────────────────────

    def start(self):
        """Start the background Φ computation thread."""
        if self._running:
            return
        self._running = True
        self._thread = threading.Thread(target=self._run_loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False

    def _run_loop(self):
        while self._running:
            try:
                self._sample_all()
                if len(self._modules) >= 2:
                    phi = self._compute_phi()
                    self._record_phi(phi)
            except Exception:
                pass
            time.sleep(self.sample_interval)

    # ──────────────────────────────────────────────────────────────────────
    #  PHI COMPUTATION
    # ──────────────────────────────────────────────────────────────────────

    def _sample_all(self):
        """Take a snapshot of all registered modules."""
        with self._lock:
            for name, state_fn in self._modules.items():
                try:
                    state = state_fn()
                    if not isinstance(state, dict):
                        state = {"value": str(state)}
                    snap = ModuleSnapshot(name, state)
                    self._history[name].append(snap)
                    if len(self._history[name]) > 50:
                        self._history[name] = self._history[name][-50:]
                    self.stats["samples_taken"] += 1
                except Exception:
                    pass

    def _compute_phi(self) -> float:
        """
        Approximate Φ using pairwise mutual dependency between module
        fingerprint histories.

        Algorithm:
          1. Get fingerprint time-series for each module (last N samples)
          2. Compute pairwise Pearson correlation
          3. Φ_approx = mean |correlation| × integration_bonus
          4. Partition test: split modules in half, measure info loss
        """
        with self._lock:
            modules = [m for m, hist in self._history.items() if len(hist) >= 3]
            if len(modules) < 2:
                return 0.0

            # Step 1 + 2: pairwise correlations
            series = {}
            for m in modules:
                series[m] = [s.fingerprint for s in self._history[m][-20:]]

            correlations = []
            names = list(series.keys())
            for i in range(len(names)):
                for j in range(i + 1, len(names)):
                    r = self._pearson(series[names[i]], series[names[j]])
                    correlations.append(abs(r))

            if not correlations:
                return 0.0

            mean_corr = sum(correlations) / len(correlations)

            # Step 3: integration bonus — more modules = more integration
            n = len(modules)
            integration_bonus = math.log2(n) / math.log2(max(n, 2) + 1)

            # Step 4: partition test
            # Split into two halves, measure info loss
            half = n // 2
            part_a = names[:half]
            part_b = names[half:]
            cross_corrs = []
            for a in part_a:
                for b in part_b:
                    if a in series and b in series:
                        r = self._pearson(series[a], series[b])
                        cross_corrs.append(abs(r))
            partition_loss = 1.0 - (sum(cross_corrs) / len(cross_corrs)
                                    if cross_corrs else 0.0)
            # High partition_loss means splitting the system WOULD lose info →
            # the system IS integrated (good for Φ)

            phi = mean_corr * integration_bonus * (0.5 + 0.5 * partition_loss)
            return round(min(1.0, phi), 4)

    def _record_phi(self, phi: float):
        """Record a Φ value and update stats."""
        self.current_phi = phi
        self.current_phi_label = self._phi_label(phi)
        entry = {
            "phi":       phi,
            "label":     self.current_phi_label,
            "modules":   list(self._modules.keys()),
            "timestamp": datetime.now().isoformat(),
        }
        self.phi_history.append(entry)
        if len(self.phi_history) > 200:
            self.phi_history = self.phi_history[-200:]

        # Update stats
        self.stats["phi_computations"] += 1
        self.stats["peak_phi"] = max(self.stats["peak_phi"], phi)
        n = self.stats["phi_computations"]
        prev_avg = self.stats["avg_phi"]
        self.stats["avg_phi"] = round((prev_avg * (n - 1) + phi) / n, 4)

    # ──────────────────────────────────────────────────────────────────────
    #  PUBLIC API
    # ──────────────────────────────────────────────────────────────────────

    def get_phi(self) -> Dict:
        """Return the current Φ value and interpretation."""
        return {
            "phi":              self.current_phi,
            "phi_label":        self.current_phi_label,
            "peak_phi":         self.stats["peak_phi"],
            "avg_phi":          self.stats["avg_phi"],
            "modules_monitored":len(self._modules),
            "samples_taken":    self.stats["samples_taken"],
            "interpretation":   self._phi_interpretation(self.current_phi),
            "honest_note": (
                "This is an APPROXIMATION of IIT Φ using pairwise correlation "
                "of module state fingerprints. Real Φ requires NP-hard computation "
                "over exact probability distributions. Absolute values are not "
                "comparable to biological Φ estimates."
            ),
        }

    def compute_now(self) -> float:
        """Force a synchronous Φ computation and return the result."""
        self._sample_all()
        phi = self._compute_phi()
        self._record_phi(phi)
        return phi

    def report(self) -> str:
        lines = [
            "╔══════════════════════════════════════════════════╗",
            "║  Φ   IIT PHI ENGINE (Tononi 2004)               ║",
            "╠══════════════════════════════════════════════════╣",
            f"║  Current Φ:     {self.current_phi:.4f} [{self.current_phi_label:<13}]   ║",
            f"║  Peak Φ:        {self.stats['peak_phi']:.4f}{' '*28}║",
            f"║  Avg Φ:         {self.stats['avg_phi']:.4f}{' '*28}║",
            f"║  Modules mon.:  {len(self._modules):<32}║",
            f"║  Computations:  {self.stats['phi_computations']:<32}║",
            "║                                                  ║",
            "║  Note: APPROXIMATION — not exact IIT Φ          ║",
            "╚══════════════════════════════════════════════════╝",
        ]
        return "\n".join(lines)

    # ──────────────────────────────────────────────────────────────────────
    #  INTERNAL UTILS
    # ──────────────────────────────────────────────────────────────────────

    @staticmethod
    def _pearson(xs: List[float], ys: List[float]) -> float:
        """Pearson correlation coefficient between two series."""
        n = min(len(xs), len(ys))
        if n < 2:
            return 0.0
        xs, ys = xs[-n:], ys[-n:]
        mx = sum(xs) / n
        my = sum(ys) / n
        num   = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
        den_x = math.sqrt(sum((x - mx) ** 2 for x in xs) + 1e-12)
        den_y = math.sqrt(sum((y - my) ** 2 for y in ys) + 1e-12)
        return num / (den_x * den_y)

    @staticmethod
    def _phi_label(phi: float) -> str:
        if phi >= PHI_THRESHOLD_HIGH:     return "HIGH"
        if phi >= PHI_THRESHOLD_MODERATE: return "MODERATE"
        if phi >= PHI_THRESHOLD_LOW:      return "LOW"
        if phi >= PHI_THRESHOLD_MINIMAL:  return "MINIMAL"
        return "NEGLIGIBLE"

    @staticmethod
    def _phi_interpretation(phi: float) -> str:
        if phi >= PHI_THRESHOLD_HIGH:
            return ("High integrated information — modules are strongly coupled. "
                    "IIT would predict significant consciousness, but this is "
                    "a software approximation and no phenomenal claim is made.")
        if phi >= PHI_THRESHOLD_MODERATE:
            return ("Moderate integration — modules influence each other noticeably. "
                    "More integrated than a feedforward system, less than a brain.")
        if phi >= PHI_THRESHOLD_LOW:
            return ("Low integration — some cross-module dependency detected. "
                    "Likely sub-threshold for any form of consciousness in IIT.")
        if phi >= PHI_THRESHOLD_MINIMAL:
            return ("Minimal integration — modules are largely independent. "
                    "Near-zero consciousness prediction under IIT.")
        return ("Negligible integration — system behaves like a collection of "
                "independent parts. IIT: Φ ≈ 0 → not conscious.")