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