limbic_engine.py

"""
Limbic State Engine — Ported from Xover-Official/LIMBIC-system-PACKGE
======================================================================
A self-contained, synchronous reimplementation of the LIMBIC system's
emotional state machine, extracted from the original async bus-based
architecture and adapted for LLM inference modulation.

Original repo: https://github.com/Xover-Official/LIMBIC-system-PACKGE

EXTRACTED FORMULAS (with exact source files):

1. AROUSAL / VALENCE (src/limbic/core/amygdala.py):
   - Threat → valence = -0.8, arousal = 0.9  (FEAR)
   - Reward → valence = +0.6, arousal = 0.4  (SEEKING)
   - Social pain (src/limbic/core/insula.py):
       valence = -0.3 × intensity
       arousal = +0.2 × intensity
   - Panic (src/limbic/core/insula.py):
       valence = -0.5, arousal = 0.8

2. RECALL TEMPERATURE (limbic_system/modules/amygdala.py):
   temp = 1.0 - (fear × 0.9) + (seeking × 2.0)
   clamped to [0.1, ∞)
   → High fear = deterministic/safety mode
   → High seeking = stochastic/creative mode

3. HORMONE DECAY (limbic_system/modules/endocrine.py):
   hormone[t+1] = hormone[t] + (baseline - hormone[t]) × 0.05

4. MODULATION FACTORS (limbic_system/modules/endocrine.py):
   fear_sensitivity  = 1.0 + cortisol
   social_bonding    = oxytocin
   fear_inhibition   = oxytocin × 0.5
   seeking_drive     = 1.0 + dopamine
   mood_stability    = serotonin

5. FEAR ENGINE (src/limbic/engines/fear.py):
   activation = max(current, arousal × hormone_modulation)
   hormone_modulation = 1.0 + cortisol - (oxytocin × 0.5)
   decay: activation *= 0.8 per tick (rapid)
   suppressed: activation *= 0.3

6. SEEKING ENGINE (src/limbic/engines/seeking.py):
   on dopamine surge: activation += magnitude
   decay: activation *= 0.95 per tick (slow)
   suppressed: increment *= 0.2

7. CARE ENGINE (src/limbic/engines/care.py):
   decay: activation *= 0.9 per tick

8. PANIC ENGINE (src/limbic/engines/panic.py):
   decay: activation *= 0.7 per tick (very rapid)
   suppressed: activation *= 0.1

9. SOMATOSENSORY → LIMBIC MAPPING (limbic_system/modules/somatosensory.py):
   pain > 0.2: fear += pain × 0.5, arousal += pain × 0.8
   |temp - 0.5| > 0.2: arousal += |temp - 0.5| × 0.5
   heart_rate > 100: arousal += (HR - 100) / 100 × 0.3

10. REWARD PREDICTION ERROR (src/limbic/core/nucleus_accumbens.py):
    RPE = actual_reward - expected_reward
    expected_reward += 0.1 × RPE  (simple learning rate)

11. OFC UTILITY (src/limbic/pfc/ofc.py):
    base_utility = mu - 0.5 × sigma  (risk penalty)
    final_utility = base_utility + (vetting_score × 0.4) - effort_cost
    Bayesian update: Normal-Normal conjugate

12. EXECUTIVE CONTROL (src/limbic/pfc/executive_control.py):
    decision_threshold = 0.3 + (effort_level × 0.3)
    deliberation_window = 0.2 + (effort_level × 0.8) seconds

13. PSYCHOLOGICAL LATTICE (src/limbic/psychology/lattice.py):
    shadow_reservoir += 0.1 × len(suppressed_drives) per override
    ego_coherence -= 0.05 per override
    shadow outburst when shadow_reservoir > 1.0 → discharge to 0.2×
    biases: loss_aversion = 1.0 + (1.0 - ego_coherence)
            optimism = 0.2 × ego_coherence

14. VAGUS NERVE (src/limbic/core/vagus_nerve.py):
    stress = (adrenaline + cortisol) / 2
    tone -= stress × 0.1
    FEAR/PANIC: tone -= level × 0.2
    CARE: tone += level × 0.1
    homeostatic: tone += (0.5 - tone) × 0.01
"""

from __future__ import annotations

import math
import random
import time
from dataclasses import dataclass, field
from typing import Optional


# ──────────────────────────────────────────────────────────────────────
# DATA TYPES
# ──────────────────────────────────────────────────────────────────────

@dataclass
class LimbicState:
    """Complete snapshot of the limbic system at a point in time."""
    # Core affect dimensions
    valence: float = 0.0       # -1.0 (negative) to +1.0 (positive)
    arousal: float = 0.5       # 0.0 (calm) to 1.0 (excited)

    # Panksepp affective engines
    fear: float = 0.0
    seeking: float = 0.2
    care: float = 0.0
    panic: float = 0.0         # separation distress

    # Hormonal state
    cortisol: float = 0.2
    oxytocin: float = 0.5
    dopamine: float = 0.4
    serotonin: float = 0.6
    adrenaline: float = 0.1

    # Homeostatic drives
    energy: float = 1.0
    sleep_pressure: float = 0.0

    # Vagal / autonomic
    vagal_tone: float = 0.5    # 0=sympathetic, 1=parasympathetic

    # Psychological layer
    shadow_reservoir: float = 0.0
    ego_coherence: float = 1.0

    # Derived LLM control signals
    temperature: float = 0.7
    top_p: float = 0.9
    dominant_engine: str = "SEEKING"

    def to_system_prompt_block(self) -> str:
        """Format limbic state as a system prompt injection."""
        autonomic = ("PARASYMPATHETIC" if self.vagal_tone > 0.6
                     else "SYMPATHETIC" if self.vagal_tone < 0.4
                     else "NEUTRAL")
        return (
            f"[LIMBIC STATE — Neuro-behavioral Context]\n"
            f"  Valence: {self.valence:+.2f}  |  Arousal: {self.arousal:.2f}\n"
            f"  Dominant Engine: {self.dominant_engine}\n"
            f"  Fear={self.fear:.2f}  Seeking={self.seeking:.2f}  "
            f"Care={self.care:.2f}  Panic={self.panic:.2f}\n"
            f"  Hormones: cortisol={self.cortisol:.2f} dopamine={self.dopamine:.2f} "
            f"oxytocin={self.oxytocin:.2f} serotonin={self.serotonin:.2f}\n"
            f"  Autonomic: {autonomic} (vagal_tone={self.vagal_tone:.2f})\n"
            f"  Ego Coherence: {self.ego_coherence:.2f}  "
            f"Shadow: {self.shadow_reservoir:.2f}\n"
            f"  → LLM Temperature: {self.temperature:.2f}  |  Top-p: {self.top_p:.2f}\n"
            f"[/LIMBIC STATE]\n"
        )

    def to_dict(self) -> dict:
        return {k: round(v, 3) if isinstance(v, float) else v
                for k, v in self.__dict__.items()}


# ──────────────────────────────────────────────────────────────────────
# LIMBIC ENGINE — The full state machine
# ──────────────────────────────────────────────────────────────────────

class LimbicEngine:
    """
    Self-contained limbic state machine that modulates LLM behavior.

    Usage:
        engine = LimbicEngine()
        state = engine.process_stimulus("I'm terrified of losing my job")
        # state.temperature is now low (deterministic/safety mode)
        # state.valence is negative
        # Use state.temperature and state.top_p for model.generate()
    """

    # Hormone baselines (from limbic_system/modules/endocrine.py)
    HORMONE_BASELINES = {
        "cortisol": 0.2,
        "oxytocin": 0.5,
        "dopamine": 0.4,
        "serotonin": 0.6,
        "adrenaline": 0.1,
    }

    # Engine decay rates per tick (from src/limbic/engines/*.py)
    ENGINE_DECAY = {
        "fear": 0.8,      # rapid decay
        "seeking": 0.95,   # slow decay
        "care": 0.9,
        "panic": 0.7,      # very rapid decay
    }

    # Keyword → (valence, arousal, engine) mapping
    # Derived from src/limbic/core/amygdala.py stimulus handling
    STIMULUS_PATTERNS = {
        # Threat / Fear triggers
        "threat": (-0.8, 0.9, "fear"),
        "danger": (-0.8, 0.9, "fear"),
        "terrified": (-0.7, 0.85, "fear"),
        "scared": (-0.6, 0.7, "fear"),
        "afraid": (-0.6, 0.7, "fear"),
        "anxious": (-0.4, 0.6, "fear"),
        "worried": (-0.3, 0.5, "fear"),
        "nervous": (-0.3, 0.5, "fear"),
        "stressed": (-0.4, 0.6, "fear"),
        "overwhelmed": (-0.5, 0.7, "fear"),

        # Panic / Separation triggers
        "alone": (-0.6, 0.7, "panic"),
        "abandoned": (-0.8, 0.8, "panic"),
        "lonely": (-0.5, 0.5, "panic"),
        "rejected": (-0.7, 0.7, "panic"),
        "loss": (-0.7, 0.6, "panic"),
        "grief": (-0.8, 0.5, "panic"),

        # Seeking / Reward triggers
        "reward": (0.6, 0.4, "seeking"),
        "excited": (0.7, 0.8, "seeking"),
        "curious": (0.4, 0.5, "seeking"),
        "interesting": (0.3, 0.4, "seeking"),
        "explore": (0.4, 0.5, "seeking"),
        "discover": (0.5, 0.6, "seeking"),
        "success": (0.7, 0.6, "seeking"),
        "achievement": (0.6, 0.5, "seeking"),
        "happy": (0.7, 0.5, "seeking"),
        "joy": (0.8, 0.6, "seeking"),

        # Care / Nurture triggers
        "help": (0.3, 0.3, "care"),
        "support": (0.4, 0.3, "care"),
        "comfort": (0.5, 0.2, "care"),
        "love": (0.8, 0.4, "care"),
        "compassion": (0.6, 0.3, "care"),
        "empathy": (0.5, 0.3, "care"),
        "kindness": (0.5, 0.3, "care"),

        # Anger / Rage
        "angry": (-0.6, 0.8, "fear"),
        "furious": (-0.8, 0.9, "fear"),
        "frustrated": (-0.4, 0.6, "fear"),
        "unfair": (-0.5, 0.7, "fear"),
        "betrayed": (-0.7, 0.8, "panic"),

        # Sadness (low arousal)
        "sad": (-0.5, 0.3, "panic"),
        "depressed": (-0.7, 0.2, "panic"),
        "hopeless": (-0.8, 0.2, "panic"),
        "miserable": (-0.7, 0.3, "panic"),
    }

    def __init__(self):
        self.state = LimbicState()
        self._tick_count = 0

    def process_stimulus(self, text: str, metadata: Optional[dict] = None) -> LimbicState:
        """
        Process a text stimulus through the full limbic pipeline.

        Pipeline (mirrors LimbicSystem.step() from limbic_system/core/system.py):
          1. Keyword → valence/arousal/engine activation (Amygdala fast-path)
          2. Hormone release based on engine activation
          3. Engine decay + hormone decay
          4. Vagal tone update
          5. Psychological lattice update
          6. Compute LLM temperature from fear/seeking balance
          7. Return complete LimbicState
        """
        self._tick_count += 1
        text_lower = text.lower()

        # ── Step 1: Amygdala fast-path — keyword stimulus evaluation ──
        # (From src/limbic/core/amygdala.py on_stimulus)
        cumulative_valence = 0.0
        cumulative_arousal = 0.0
        match_count = 0

        for keyword, (v, a, engine) in self.STIMULUS_PATTERNS.items():
            if keyword in text_lower:
                cumulative_valence += v
                cumulative_arousal += a
                match_count += 1

                # Activate the corresponding engine
                current = getattr(self.state, engine)
                # FearEngine formula: activation = max(current, arousal × hormone_modulation)
                hormone_mod = self._get_hormone_modulation(engine)
                new_activation = a * hormone_mod
                setattr(self.state, engine, max(current, min(1.0, new_activation)))

        if match_count > 0:
            self.state.valence = max(-1.0, min(1.0, cumulative_valence / match_count))
            self.state.arousal = max(0.0, min(1.0, cumulative_arousal / match_count))
        else:
            # Neutral stimulus — mild decay toward baseline
            self.state.valence *= 0.9
            self.state.arousal = self.state.arousal * 0.9 + 0.3 * 0.1

        # Apply metadata overrides if provided
        if metadata:
            if "threat_level" in metadata and metadata["threat_level"] > 0.5:
                self.state.valence = min(self.state.valence, -0.8)
                self.state.arousal = max(self.state.arousal, 0.9)
                self.state.fear = max(self.state.fear, metadata["threat_level"])

        # ── Step 2: Hormone release (from limbic_system/core/system.py) ──
        # High fear → cortisol + adrenaline
        if self.state.fear > 0.7:
            self.state.cortisol = min(1.0, self.state.cortisol + 0.1)
            self.state.adrenaline = min(1.0, self.state.adrenaline + 0.2)
        if self.state.fear > 0.5:
            self.state.cortisol = min(1.0, self.state.cortisol + 0.05 * self.state.fear)
            self.state.adrenaline = min(1.0, self.state.adrenaline + 0.1 * self.state.fear)

        # High seeking → dopamine
        if self.state.seeking > 0.7:
            self.state.dopamine = min(1.0, self.state.dopamine + 0.1)
        if self.state.seeking > 0.5:
            self.state.dopamine = min(1.0, self.state.dopamine + 0.05 * self.state.seeking)

        # Care → oxytocin
        if self.state.care > 0.5:
            self.state.oxytocin = min(1.0, self.state.oxytocin + 0.05 * self.state.care)

        # ── Step 3: Engine decay (from src/limbic/engines/*.py) ──
        for engine_name, decay_rate in self.ENGINE_DECAY.items():
            current = getattr(self.state, engine_name)
            setattr(self.state, engine_name, current * decay_rate)

        # ── Step 4: Hormone decay toward baseline ──
        # (From limbic_system/modules/endocrine.py: diff × 0.05)
        for hormone, baseline in self.HORMONE_BASELINES.items():
            current = getattr(self.state, hormone)
            diff = baseline - current
            setattr(self.state, hormone, current + diff * 0.05)

        # ── Step 5: Vagal tone update (from src/limbic/core/vagus_nerve.py) ──
        stress = (self.state.adrenaline + self.state.cortisol) / 2
        self.state.vagal_tone = max(0.0, min(1.0,
            self.state.vagal_tone - stress * 0.1))

        if self.state.fear > 0.5 or self.state.panic > 0.5:
            fear_panic_max = max(self.state.fear, self.state.panic)
            self.state.vagal_tone = max(0.0,
                self.state.vagal_tone - fear_panic_max * 0.2)

        if self.state.care > 0.5:
            self.state.vagal_tone = min(1.0,
                self.state.vagal_tone + self.state.care * 0.1)

        # Homeostatic tendency
        self.state.vagal_tone += (0.5 - self.state.vagal_tone) * 0.01

        # ── Step 6: Psychological lattice ──
        # Shadow grows when drives are suppressed (simplified: when fear overrides seeking)
        if self.state.fear > 0.5 and self.state.seeking > 0.3:
            self.state.shadow_reservoir += 0.05
            self.state.ego_coherence = max(0.0, self.state.ego_coherence - 0.02)

        # Shadow decay + ego recovery
        self.state.shadow_reservoir = max(0.0, self.state.shadow_reservoir - 0.01)
        self.state.ego_coherence = min(1.0, self.state.ego_coherence + 0.005)

        # Shadow outburst check
        if self.state.shadow_reservoir > 1.0:
            self.state.shadow_reservoir *= 0.2  # discharge
            self.state.arousal = min(1.0, self.state.arousal + 0.3)

        # ── Step 7: Compute LLM generation parameters ──
        # CORE FORMULA from limbic_system/modules/amygdala.py get_recall_temperature():
        #   temp = 1.0 - (fear × 0.9) + (seeking × 2.0)
        #   clamped to [0.1, ∞)
        #
        # We adapt this for LLM generation with reasonable bounds:
        raw_temp = 1.0 - (self.state.fear * 0.9) + (self.state.seeking * 2.0)
        # Serotonin stabilizes (reduces extremes)
        raw_temp = raw_temp * (0.5 + self.state.serotonin * 0.5)
        # Clamp to [0.1, 1.5] for safe LLM generation
        self.state.temperature = max(0.1, min(1.5, raw_temp))

        # Top-p: tighter under fear (more deterministic), wider under seeking
        self.state.top_p = max(0.5, min(0.99,
            0.85 - (self.state.fear * 0.3) + (self.state.seeking * 0.15)))

        # ── Step 8: Determine dominant engine ──
        engines = {
            "FEAR": self.state.fear,
            "SEEKING": self.state.seeking,
            "CARE": self.state.care,
            "PANIC": self.state.panic,
        }
        self.state.dominant_engine = max(engines, key=engines.get)

        return self.state

    def _get_hormone_modulation(self, engine: str) -> float:
        """
        Hormone modulation factor per engine type.
        From src/limbic/engines/fear.py:
          hormone_modulation = 1.0 + cortisol - (oxytocin × 0.5)
        """
        if engine == "fear":
            return 1.0 + self.state.cortisol - (self.state.oxytocin * 0.5)
        elif engine == "seeking":
            return 1.0 + self.state.dopamine * 0.5
        elif engine == "care":
            return 1.0 + self.state.oxytocin * 0.5
        elif engine == "panic":
            return 1.0 + self.state.cortisol * 0.3
        return 1.0

    def get_generation_params(self) -> dict:
        """Get current LLM generation parameters modulated by limbic state."""
        return {
            "temperature": self.state.temperature,
            "top_p": self.state.top_p,
            "do_sample": True,
            "repetition_penalty": 1.0 + (self.state.fear * 0.2),
            # max_new_tokens modulated: cautious under fear, verbose under seeking
            "max_new_tokens_scale": max(0.5, min(1.5,
                1.0 - (self.state.fear * 0.3) + (self.state.seeking * 0.3))),
        }

    def reset(self):
        """Reset to default resting state."""
        self.state = LimbicState()
        self._tick_count = 0

    def get_behavioral_directive(self) -> str:
        """
        Convert limbic state to a behavioral directive for the system prompt.
        This tells the LLM HOW to behave based on the simulated neuro-response.
        """
        directives = []

        if self.state.fear > 0.5:
            directives.append(
                "The user appears to be in a heightened threat-response state. "
                "Respond with calm, structured, safety-oriented language. "
                "Avoid adding new stressors. Prioritize reassurance and concrete next steps."
            )
        if self.state.panic > 0.4:
            directives.append(
                "The user shows signs of separation distress or loss. "
                "Respond with warmth and validation. Acknowledge their pain before "
                "offering solutions. Use attachment-theory-informed language."
            )
        if self.state.seeking > 0.6:
            directives.append(
                "The user is in an exploratory/curious state. "
                "Encourage exploration with novel information. Be creative and expansive. "
                "Offer multiple perspectives and interesting tangents."
            )
        if self.state.care > 0.5:
            directives.append(
                "The user's care/nurture system is active. "
                "Match their empathetic energy. Acknowledge the prosocial intent. "
                "Support their caregiving impulse with practical guidance."
            )
        if self.state.ego_coherence < 0.6:
            directives.append(
                "Psychological coherence is low — the user may be conflicted. "
                "Avoid black-and-white framing. Use gentle Socratic questioning "
                "to help them integrate conflicting feelings."
            )
        if self.state.shadow_reservoir > 0.5:
            directives.append(
                "Suppressed drives are building up. "
                "Create space for the user to express what they may be avoiding. "
                "Gently surface potential unacknowledged feelings."
            )

        if not directives:
            directives.append(
                "The user is in a balanced state. "
                "Respond naturally with a mix of warmth and intellectual engagement."
            )

        return "\n".join(f"• {d}" for d in directives)