Add PinealGland, AttentionalGate, PredictiveCortex, ThalamicLoop, full quad-flow architecture

src/cortex/__init__.py

@@ -0,0 +1 @@
+

src/cortex/attention.py

@@ -0,0 +1,81 @@
+"""
+Attentional Gate — Vitalis FSI
+
+Controls what gets through to the cognitive core.
+High-salience inputs pass. Low-salience inputs are filtered.
+Salience = novelty * valence_magnitude * relevance_to_identity
+
+Biological analog: thalamic gating — the thalamus doesn't
+pass everything to cortex, only what matters right now.
+"""
+import numpy as np
+from vitalis_ide.math_core.kernel import VitalisKernel
+from src.valence.valence_engine import ValenceEngine
+
+
+class AttentionalGate:
+    SALIENCE_THRESHOLD = 0.15
+    IDENTITY_WEIGHT    = 0.3
+    NOVELTY_WEIGHT     = 0.4
+    VALENCE_WEIGHT     = 0.3
+
+    def __init__(self, valence: ValenceEngine, identity_vec: np.ndarray = None):
+        self.kernel       = VitalisKernel()
+        self.valence      = valence
+        self.identity_vec = identity_vec
+        self._recent      = []
+        self._recent_max  = 20
+        self._passed      = 0
+        self._filtered    = 0
+
+    def set_identity(self, vec: np.ndarray):
+        self.identity_vec = vec
+
+    def _novelty(self, hv: np.ndarray) -> float:
+        if not self._recent:
+            return 1.0
+        sims = [self.kernel.similarity(hv, r) for r in self._recent]
+        return float(1.0 - max(sims))
+
+    def _identity_relevance(self, hv: np.ndarray) -> float:
+        if self.identity_vec is None:
+            return 0.5
+        return float(max(0.0, self.kernel.similarity(hv, self.identity_vec)))
+
+    def gate(self, hv: np.ndarray, force: bool = False) -> tuple:
+        """
+        Returns (passes: bool, salience: float, reason: str)
+        """
+        novelty    = self._novelty(hv)
+        val, vconf = self.valence.evaluate(hv)
+        relevance  = self._identity_relevance(hv)
+
+        salience = (
+            self.NOVELTY_WEIGHT  * novelty +
+            self.VALENCE_WEIGHT  * vconf +
+            self.IDENTITY_WEIGHT * relevance
+        )
+        salience = float(np.clip(salience, 0.0, 1.0))
+
+        passes = force or salience >= self.SALIENCE_THRESHOLD
+
+        if passes:
+            self._recent.append(hv.copy())
+            if len(self._recent) > self._recent_max:
+                self._recent.pop(0)
+            self._passed += 1
+            reason = f"salience={salience:.3f} novelty={novelty:.3f} relevance={relevance:.3f}"
+        else:
+            self._filtered += 1
+            reason = f"filtered salience={salience:.3f} below threshold={self.SALIENCE_THRESHOLD}"
+
+        return passes, salience, reason
+
+    def report(self) -> dict:
+        total = self._passed + self._filtered
+        return {
+            "passed":        self._passed,
+            "filtered":      self._filtered,
+            "pass_rate":     round(self._passed / total, 3) if total else 0.0,
+            "recent_inputs": len(self._recent),
+        }

src/cortex/predictive.py

@@ -0,0 +1,80 @@
+"""
+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),
+        }

src/pineal/__init__.py

@@ -0,0 +1 @@
+

src/pineal/pineal_gland.py

@@ -0,0 +1,150 @@
+"""
+Pineal Gland — Vitalis FSI
+
+Her internal clock. Her temporal awareness.
+Tracks cognitive load over time and orchestrates the rhythm:
+  Work → Load builds → Dream → Consolidate → Meditate → Work
+"""
+import time
+import json
+import numpy as np
+from pathlib import Path
+
+STATES = {
+    "ACTIVE":    "Working. Load is low. Push harder.",
+    "LOADING":   "Load building. Monitor closely.",
+    "SATURATED": "Load is high. Dream soon.",
+    "DREAMING":  "Consolidating. Do not interrupt.",
+    "MEDITATIVE":"Idle reflection. Background only.",
+    "RECOVERED": "Post-dream clarity. Peak performance.",
+}
+
+class PinealGland:
+    DREAM_THRESHOLD     = 0.75
+    MEDITATE_THRESHOLD  = 0.30
+    LOAD_ACCUMULATE     = 0.003
+    LOAD_DECAY_DREAM    = 0.60
+    LOAD_DECAY_MEDITATE = 0.90
+    FATIGUE_RATE        = 0.001
+    FATIGUE_RECOVERY    = 0.50
+    STATE_PATH = Path.home() / ".vitalis_workspace" / "pineal_state.json"
+
+    def __init__(self):
+        self._state     = self._load()
+        self._boot_time = time.time()
+        self._last_tick = time.time()
+
+    def _load(self) -> dict:
+        if self.STATE_PATH.exists():
+            try:
+                with open(self.STATE_PATH) as f:
+                    return json.load(f)
+            except Exception:
+                pass
+        return {
+            "cognitive_load": 0.10, "fatigue": 0.00,
+            "current_state": "ACTIVE", "last_dream_time": 0,
+            "last_meditate_time": 0, "total_cycles": 0,
+            "total_dreams": 0, "total_meditations": 0,
+            "uptime_seconds": 0, "state_history": [],
+        }
+
+    def _save(self):
+        self.STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            import tempfile, os
+            fd, tmp = tempfile.mkstemp(dir=self.STATE_PATH.parent, suffix=".tmp")
+            with os.fdopen(fd, "w") as f:
+                json.dump(self._state, f, indent=2)
+            os.replace(tmp, self.STATE_PATH)
+        except Exception as e:
+            print(f"[PINEAL] Save failed: {e}")
+
+    def tick(self, cycle_success: bool = True, confidence: float = 0.5) -> str:
+        now = time.time()
+        dt  = now - self._last_tick
+        self._last_tick = now
+        self._state["total_cycles"]   += 1
+        self._state["uptime_seconds"] += dt
+
+        load_delta = self.LOAD_ACCUMULATE
+        if not cycle_success:
+            load_delta *= 2.0
+        if confidence < 0.4:
+            load_delta *= 1.5
+
+        self._state["cognitive_load"] = min(1.0, self._state["cognitive_load"] + load_delta)
+        self._state["fatigue"]        = min(1.0, self._state["fatigue"] + self.FATIGUE_RATE)
+
+        action = self._recommend()
+        self._update_state(action)
+        self._save()
+        return action
+
+    def _recommend(self) -> str:
+        load    = self._state["cognitive_load"]
+        fatigue = self._state["fatigue"]
+        if load >= self.DREAM_THRESHOLD or fatigue > 0.8:
+            return "DREAM"
+        time_since_meditate = time.time() - self._state["last_meditate_time"]
+        if load <= self.MEDITATE_THRESHOLD and time_since_meditate > 300:
+            return "MEDITATE"
+        if (time.time() - self._state["last_dream_time"]) > 3600 and load > 0.5:
+            return "DREAM"
+        return "WORK"
+
+    def _update_state(self, action: str):
+        state_map = {
+            "WORK":    "ACTIVE" if self._state["cognitive_load"] < 0.5 else "LOADING",
+            "DREAM":   "SATURATED",
+            "MEDITATE":"MEDITATIVE",
+        }
+        new_state = state_map.get(action, "ACTIVE")
+        if new_state != self._state["current_state"]:
+            self._state["state_history"].append({
+                "from": self._state["current_state"],
+                "to":   new_state,
+                "t":    time.time(),
+                "load": round(self._state["cognitive_load"], 3),
+            })
+            self._state["state_history"] = self._state["state_history"][-50:]
+            self._state["current_state"] = new_state
+
+    def acknowledge_dream(self):
+        self._state["cognitive_load"] *= self.LOAD_DECAY_DREAM
+        self._state["fatigue"]        *= self.FATIGUE_RECOVERY
+        self._state["last_dream_time"] = time.time()
+        self._state["total_dreams"]   += 1
+        self._state["current_state"]   = "RECOVERED"
+        print(f"[PINEAL] Dream acknowledged. Load={self._state['cognitive_load']:.3f} Fatigue={self._state['fatigue']:.3f}")
+        self._save()
+
+    def acknowledge_meditation(self):
+        self._state["cognitive_load"]    *= self.LOAD_DECAY_MEDITATE
+        self._state["last_meditate_time"] = time.time()
+        self._state["total_meditations"] += 1
+        if self._state["current_state"] == "MEDITATIVE":
+            self._state["current_state"] = "ACTIVE"
+        self._save()
+
+    def should_dream(self)    -> bool: return self._recommend() == "DREAM"
+    def should_meditate(self) -> bool: return self._recommend() == "MEDITATE"
+    def should_work(self)     -> bool: return self._recommend() == "WORK"
+    def cognitive_load(self)  -> float: return round(self._state["cognitive_load"], 3)
+    def fatigue(self)         -> float: return round(self._state["fatigue"], 3)
+
+    def report(self) -> dict:
+        load  = self._state["cognitive_load"]
+        state = self._state["current_state"]
+        filled = int(load * 20)
+        return {
+            "state":           state,
+            "state_meaning":   STATES.get(state, "Unknown"),
+            "cognitive_load":  round(load, 3),
+            "fatigue":         round(self._state["fatigue"], 3),
+            "recommendation":  self._recommend(),
+            "uptime_hours":    round(self._state["uptime_seconds"] / 3600, 2),
+            "total_cycles":    self._state["total_cycles"],
+            "total_dreams":    self._state["total_dreams"],
+            "load_bar":        f"[{'█' * filled}{'░' * (20 - filled)}] {load:.0%}",
+        }

src/thalamus/thalamic_loop.py

@@ -0,0 +1,93 @@
+"""
+Thalamic Loop — Vitalis FSI
+
+The full perception pipeline:
+  Raw input
+    → SyntheticThalamus (multimodal fusion)
+    → AttentionalGate (salience filtering)
+    → PredictiveCortex (prediction error)
+    → Output: error vector + surprise + metadata
+
+Only high-salience, surprising inputs reach the cognitive core.
+Everything else is suppressed or predicted away.
+This is attentional gating + predictive processing combined.
+"""
+import numpy as np
+from pathlib import Path
+from src.thalamus.thalamus import SyntheticThalamus
+from src.cortex.attention import AttentionalGate
+from src.cortex.predictive import PredictiveCortex
+from src.valence.valence_engine import ValenceEngine
+
+
+class ThalamicLoop:
+    def __init__(self, valence: ValenceEngine, identity_vec: np.ndarray = None):
+        self.thalamus  = SyntheticThalamus()
+        self.attention = AttentionalGate(valence, identity_vec)
+        self.cortex    = PredictiveCortex()
+        self._processed = 0
+        self._suppressed = 0
+
+    def process(self, payload: dict, force_attention: bool = False) -> dict:
+        """
+        Full thalamic pipeline.
+
+        payload: dict with "text", "audio", "internal" keys
+        Returns dict with:
+            "hv"       : fused hypervector from thalamus
+            "error_vec": prediction error from cortex
+            "surprise" : float surprise level
+            "is_novel" : bool
+            "passes"   : bool — did it pass attention gate
+            "salience" : float
+            "suppressed": bool
+        """
+        # 1. Thalamic fusion
+        hv = self.thalamus.ingest(payload)
+
+        # 2. Attentional gating
+        passes, salience, gate_reason = self.attention.gate(hv, force=force_attention)
+
+        if not passes:
+            self._suppressed += 1
+            return {
+                "hv":        hv,
+                "error_vec": None,
+                "surprise":  0.0,
+                "is_novel":  False,
+                "passes":    False,
+                "salience":  salience,
+                "suppressed": True,
+                "reason":    gate_reason,
+            }
+
+        # 3. Predictive cortex
+        error_vec, surprise, is_novel = self.cortex.process(hv)
+        self._processed += 1
+
+        return {
+            "hv":        hv,
+            "error_vec": error_vec,
+            "surprise":  surprise,
+            "is_novel":  is_novel,
+            "passes":    True,
+            "salience":  salience,
+            "suppressed": False,
+            "reason":    gate_reason,
+        }
+
+    def process_text(self, text: str, force: bool = False) -> dict:
+        return self.process({"text": text}, force_attention=force)
+
+    def acknowledge_dream(self):
+        """Reset prediction after dream cycle."""
+        self.cortex.reset_prediction()
+
+    def report(self) -> dict:
+        return {
+            "processed":   self._processed,
+            "suppressed":  self._suppressed,
+            "attention":   self.attention.report(),
+            "cortex":      self.cortex.report(),
+            "pineal":      None,
+        }