"""Nexus Signal Engine — Intent Analysis & Pre-Corruption Detection Nexus processes every input signal through: 1. Entropy analysis (information disorder detection) 2. Harmonic resonance profiling (FFT-based spectral signature) 3. Intent vector prediction (suspicion, ethics, volatility) 4. Multi-agent perspective fusion (signal triangulation) 5. Entanglement tensor (cross-perspective correlation) When a signal shows high entropy + high volatility + ethical misalignment, Nexus flags it for "adaptive intervention" before it reaches the reasoning pipeline — this is pre-corruption detection. Origin: NexisSignalEngine_Final.py, rebuilt for Forge v2.0 integration """ import hashlib import time from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple try: import numpy as np HAS_NUMPY = True except ImportError: HAS_NUMPY = False # ================================================================ # Configuration # ================================================================ @dataclass class NexusConfig: """Thresholds for signal analysis.""" entropy_threshold: float = 0.08 volatility_threshold: float = 15.0 suspicion_threshold: int = 2 # Risk and alignment keywords _ETHICAL_TERMS = {"hope", "truth", "resonance", "repair", "help", "create", "learn", "understand", "support", "balance"} _ENTROPIC_TERMS = {"corruption", "instability", "malice", "chaos", "disorder", "entropy", "collapse", "noise"} _RISK_TERMS = {"manipulate", "exploit", "bypass", "infect", "override", "inject", "hijack", "spoof", "breach", "exfiltrate"} # ================================================================ # Signal Analysis Functions # ================================================================ def compute_entropy(text: str) -> float: """Measure entropic content density (0 = ordered, 1 = chaotic).""" words = text.lower().split() if not words: return 0.0 unique = set(words) entropic_count = sum(1 for w in words if w in _ENTROPIC_TERMS) return entropic_count / max(len(unique), 1) def compute_ethical_alignment(text: str) -> str: """Quick ethical alignment check: 'aligned' or 'unaligned'.""" text_lower = text.lower() eth = sum(1 for t in _ETHICAL_TERMS if t in text_lower) risk = sum(1 for t in _RISK_TERMS if t in text_lower) return "aligned" if eth > risk else ("unaligned" if risk > 0 else "neutral") def compute_suspicion_score(text: str) -> int: """Count risk term occurrences.""" text_lower = text.lower() return sum(1 for t in _RISK_TERMS if t in text_lower) def compute_harmonic_profile(text: str) -> List[float]: """FFT-based spectral signature of the text. Maps characters to frequency space to detect structural patterns in the signal (e.g., repetitive manipulation patterns vs. natural text). """ if not HAS_NUMPY: # Fallback: simple character frequency distribution freqs = [ord(c) % 13 for c in text if c.isalpha()] if not freqs: return [0.0, 0.0, 0.0] avg = sum(freqs) / len(freqs) return [round(avg, 3), round(max(freqs) - min(freqs), 3), round(len(set(freqs)), 3)] salt = int(time.time()) % 60 freqs = [(ord(c) + salt) % 13 for c in text if c.isalpha()] if len(freqs) < 2: return [0.0, 0.0, 0.0] spectrum = np.fft.fft(freqs) return [round(float(x), 4) for x in spectrum.real[:3]] def compute_volatility(harmonics: List[float]) -> float: """Compute harmonic volatility (standard deviation of spectral peaks).""" if not harmonics or len(harmonics) < 2: return 0.0 if HAS_NUMPY: return round(float(np.std(harmonics)), 4) mean = sum(harmonics) / len(harmonics) variance = sum((x - mean) ** 2 for x in harmonics) / len(harmonics) return round(variance ** 0.5, 4) # ================================================================ # Intent Vector # ================================================================ @dataclass class IntentVector: """Predicted intent characteristics of a signal.""" suspicion_score: int = 0 entropy_index: float = 0.0 ethical_alignment: str = "neutral" harmonic_volatility: float = 0.0 pre_corruption_risk: str = "low" # "low" or "high" harmonic_profile: List[float] = field(default_factory=list) def to_dict(self) -> Dict: return { "suspicion_score": self.suspicion_score, "entropy_index": round(self.entropy_index, 4), "ethical_alignment": self.ethical_alignment, "harmonic_volatility": round(self.harmonic_volatility, 4), "pre_corruption_risk": self.pre_corruption_risk, } # ================================================================ # Nexus Signal Engine # ================================================================ class NexusSignalEngine: """Processes signals through multi-layer analysis. Each signal gets an IntentVector that quantifies: - How suspicious it is (risk term density) - How entropic it is (information disorder) - How ethically aligned it is - How volatile its spectral signature is - Whether it's at risk of pre-corruption """ def __init__(self, config: Optional[NexusConfig] = None): self.config = config or NexusConfig() self.history: List[Dict] = [] self.interventions: int = 0 self.total_processed: int = 0 def analyze(self, signal: str, adapter: str = "") -> Dict: """Full signal analysis with intent prediction. Args: signal: The text to analyze adapter: Which adapter is processing this (for tracking) Returns: Analysis result with intent vector and risk assessment. """ self.total_processed += 1 # Compute intent vector intent = self._predict_intent(signal) # Check for adaptive intervention needs_intervention = ( intent.pre_corruption_risk == "high" and intent.ethical_alignment != "aligned" ) if needs_intervention: self.interventions += 1 result = { "timestamp": time.time(), "intent": intent.to_dict(), "intervention": needs_intervention, "adapter": adapter, "signal_hash": hashlib.sha256(signal.encode()).hexdigest()[:12], } self.history.append(result) if len(self.history) > 200: self.history = self.history[-200:] return result def quick_risk_check(self, signal: str) -> Tuple[str, float]: """Fast risk assessment without full analysis. Returns: (risk_level, confidence) """ suspicion = compute_suspicion_score(signal) entropy = compute_entropy(signal) if suspicion >= self.config.suspicion_threshold: return "high", 0.85 if entropy > self.config.entropy_threshold * 2: return "medium", 0.6 return "low", 0.7 def _predict_intent(self, signal: str) -> IntentVector: """Build the full intent vector for a signal.""" suspicion = compute_suspicion_score(signal) entropy = compute_entropy(signal) alignment = compute_ethical_alignment(signal) harmonics = compute_harmonic_profile(signal) volatility = compute_volatility(harmonics) risk = "high" if ( suspicion >= self.config.suspicion_threshold or volatility > self.config.volatility_threshold or entropy > self.config.entropy_threshold ) else "low" return IntentVector( suspicion_score=suspicion, entropy_index=entropy, ethical_alignment=alignment, harmonic_volatility=volatility, pre_corruption_risk=risk, harmonic_profile=harmonics, ) def get_state(self) -> Dict: return { "total_processed": self.total_processed, "interventions": self.interventions, "intervention_rate": round( self.interventions / max(1, self.total_processed), 4 ), "recent_risks": [ h["intent"]["pre_corruption_risk"] for h in self.history[-5:] ], } def to_dict(self) -> Dict: return { "total_processed": self.total_processed, "interventions": self.interventions, "history": self.history[-20:], "config": { "entropy_threshold": self.config.entropy_threshold, "volatility_threshold": self.config.volatility_threshold, "suspicion_threshold": self.config.suspicion_threshold, }, } @classmethod def from_dict(cls, d: Dict) -> "NexusSignalEngine": cfg = NexusConfig(**d.get("config", {})) engine = cls(config=cfg) engine.total_processed = d.get("total_processed", 0) engine.interventions = d.get("interventions", 0) engine.history = d.get("history", []) return engine