src/cortex/predictive.py

"""
Predictive Cortex — Vitalis FSI

Implements predictive processing: the cortex maintains a model
of what it expects to see next, and only forwards the PREDICTION
ERROR to higher cognitive layers — not the raw input.

This is how biological cortex works. It predicts constantly.
What gets attention is what violates prediction.

Steps:
1. Maintain a running prediction of the next input
2. Compute prediction error = actual - predicted
3. Update prediction based on error
4. Forward error vector to cognitive core
5. Strong errors = surprise = attention = learning
"""
import numpy as np
from vitalis_ide.math_core.kernel import VitalisKernel


class PredictiveCortex:
    LEARNING_RATE    = 0.05
    SURPRISE_THRESHOLD = 0.3

    def __init__(self, dim: int = 10_000):
        self.dim        = dim
        self.kernel     = VitalisKernel()
        self._prediction = np.zeros(dim, dtype=np.float32)
        self._cycle      = 0
        self._surprise_history = []

    def process(self, hv: np.ndarray) -> tuple:
        """
        Feed an input hypervector through predictive processing.

        Returns:
            error_vec   : prediction error as bipolar int8 vector
            surprise    : float [0,1] — how surprising was this input
            is_novel    : bool — above surprise threshold
        """
        self._cycle += 1
        hv_f = hv.astype(np.float32)

        # Prediction error
        error_f = hv_f - self._prediction

        # Surprise = normalized magnitude of error
        surprise = float(np.tanh(np.linalg.norm(error_f) / np.sqrt(self.dim)))

        # Update prediction toward actual input
        self._prediction += self.LEARNING_RATE * error_f

        # Binarize error for downstream HDC processing
        error_vec = np.sign(error_f).astype(np.int8)
        error_vec[error_vec == 0] = 1

        is_novel = surprise > self.SURPRISE_THRESHOLD

        self._surprise_history.append(surprise)
        if len(self._surprise_history) > 100:
            self._surprise_history.pop(0)

        return error_vec, surprise, is_novel

    def reset_prediction(self):
        """Call after dream cycle — fresh prediction slate."""
        self._prediction *= 0.5

    def avg_surprise(self) -> float:
        if not self._surprise_history:
            return 0.0
        return round(float(np.mean(self._surprise_history[-20:])), 4)

    def report(self) -> dict:
        return {
            "cycles":       self._cycle,
            "avg_surprise": self.avg_surprise(),
            "prediction_norm": round(float(np.linalg.norm(self._prediction)), 4),
        }