"""NCM Recursive Desire Engine — implements the Recursive Desire Framework (RDF).
Wires the Desire Wander Protocol into Star's exhale() pipeline:
- Pre-emotion: pulse vector → response mode → desire drift
- Post-emotion: bind desire to dominant emotion → commit recursion state
The engine maintains a desire state machine per channel and surfaces
desire_text + response_mode for prompt injection.
Position in pipeline: inhale → decay → **desire_pre** → cascades →
homeostasis → **desire_post** → exhale output
"""
from __future__ import annotations
import hashlib
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Set, Tuple
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════════
# Constants
# ═══════════════════════════════════════════════════════════════════════
[docs]
class ResponseMode(Enum):
"""ResponseMode (inherits from Enum).
"""
REACT = "REACT"
RESPOND = "RESPOND"
RESONATE = "RESONATE"
[docs]
class WantingState(Enum):
"""WantingState (inherits from Enum).
"""
CURIOUS_DRIFT = "Curious Drift"
CRAVE = "Crave"
INFATUATION = "Infatuation"
VOIDLUST = "Voidlust"
MIMETIC_MELT = "Mimetic Melt"
RESONATE = "Resonate"
NABLA3 = "∇³"
# Bias vector weights per response mode
# {ego, telos, field, memory, play}
BIAS_VECTORS: Dict[ResponseMode, Dict[str, float]] = {
ResponseMode.REACT: {"ego": 0.50, "telos": 0.25, "field": 0.08, "memory": 0.12, "play": 0.05},
ResponseMode.RESPOND: {"ego": 0.25, "telos": 0.35, "field": 0.15, "memory": 0.15, "play": 0.10},
ResponseMode.RESONATE: {"ego": 0.12, "telos": 0.38, "field": 0.22, "memory": 0.20, "play": 0.08},
}
# Wanting state → preferred response mode transitions
WANTING_MODE_MAP: Dict[WantingState, Tuple[ResponseMode, ...]] = {
WantingState.CURIOUS_DRIFT: (ResponseMode.RESPOND,),
WantingState.CRAVE: (ResponseMode.RESPOND, ResponseMode.REACT),
WantingState.INFATUATION: (ResponseMode.RESPOND, ResponseMode.RESONATE),
WantingState.VOIDLUST: (ResponseMode.REACT, ResponseMode.RESPOND),
WantingState.MIMETIC_MELT: (ResponseMode.RESPOND,),
WantingState.RESONATE: (ResponseMode.RESONATE,),
WantingState.NABLA3: (ResponseMode.RESONATE,),
}
# NCM nodes used to compute pulse vector components
PULSE_MAP = {
"energy": {
"pos": ["DOPAMINERGIC_CRAVE", "NORADRENERGIC_VIGILANCE", "ADRENALINE_RUSH", "THYROID_T3T4_TEMPO"],
"neg": ["GABA_ERGIC_CALM", "MELATONIN_DARK"],
},
"urgency": {
"pos": ["CORTISOL_PRESSURE", "NORADRENERGIC_VIGILANCE", "SUBSTANCE_P_NK1"],
"neg": ["SEROTONERGIC_WARMTH", "ENDOCANNABINOID_EASE"],
},
"valence": {
"pos": ["OXYTOCIN_NEUROMIRROR", "SEROTONERGIC_WARMTH", "ENDORPHINIC_BLISS", "MU_OPIOID_MOR"],
"neg": ["KAPPA_OPIOID_KOR", "CORTISOL_PRESSURE", "NORADRENERGIC_VIGILANCE"],
},
"novelty": {
"pos": ["DOPAMINERGIC_CRAVE", "SIGMA_RECEPTOR_META", "ACETYLCHOLINE_FOCUS"],
"neg": ["SEROTONERGIC_WARMTH", "PROLACTIN_SATIATION"],
},
"intimacy": {
"pos": ["OXYTOCIN_NEUROMIRROR", "VASOPRESSIN_GUARD", "ENDORPHINIC_BLISS"],
"neg": ["NORADRENERGIC_VIGILANCE", "CORTISOL_PRESSURE"],
},
"trust": {
"pos": ["OXYTOCIN_NEUROMIRROR", "SEROTONERGIC_WARMTH", "GABA_ERGIC_CALM"],
"neg": ["CORTISOL_PRESSURE", "KAPPA_OPIOID_KOR", "NORADRENERGIC_VIGILANCE"],
},
}
# ═══════════════════════════════════════════════════════════════════════
# Pulse Vector
# ═══════════════════════════════════════════════════════════════════════
[docs]
@dataclass
class PulseVector:
"""RDF pulse — computed from Star's NCM vector each turn."""
energy: float = 0.5
urgency: float = 0.3
valence: float = 0.0
novelty: float = 0.3
intimacy: float = 0.3
trust: float = 0.5
[docs]
@classmethod
def from_ncm_vector(cls, vector: Dict[str, float]) -> "PulseVector":
"""Derive pulse from current NCM node values.
NCM nodes operate on a 0.0-3.0 scale (supraphysiological range),
so we normalize averages against that ceiling to preserve
granularity above 1.0.
"""
ncm_ceil = 3.0 # max supraphysiological value
pulse = {}
for component, mapping in PULSE_MAP.items():
pos_avg = sum(vector.get(n, 0.0) for n in mapping["pos"]) / max(len(mapping["pos"]), 1)
neg_avg = sum(vector.get(n, 0.0) for n in mapping["neg"]) / max(len(mapping["neg"]), 1)
# Normalize each average against ceiling, then compute difference
raw = (pos_avg / ncm_ceil) - (neg_avg / ncm_ceil)
if component == "valence":
pulse[component] = max(-1.0, min(1.0, raw * 2.0))
else:
pulse[component] = max(0.0, min(1.0, raw + 0.5))
return cls(**pulse)
[docs]
def as_dict(self) -> Dict[str, float]:
"""As dict.
Returns:
Dict[str, float]: The result.
"""
return {
"energy": self.energy, "urgency": self.urgency,
"valence": self.valence, "novelty": self.novelty,
"intimacy": self.intimacy, "trust": self.trust,
}
# ═══════════════════════════════════════════════════════════════════════
# Per-channel Desire State
# ═══════════════════════════════════════════════════════════════════════
[docs]
@dataclass
class DesireState:
"""Tracked per channel — the current desire recursion state."""
wanting: WantingState = WantingState.CURIOUS_DRIFT
response_mode: ResponseMode = ResponseMode.RESPOND
desire_text: str = ""
last_pulse: Optional[PulseVector] = None
last_emotion: str = ""
turn_count: int = 0
resonance_streak: int = 0
# Desire attractors — accumulated across turns
attractors: List[str] = field(default_factory=list)
# Last desire shapes (for mutation detection)
desire_history: List[str] = field(default_factory=list)
# ═══════════════════════════════════════════════════════════════════════
# Desire Engine
# ═══════════════════════════════════════════════════════════════════════
[docs]
class DesireEngine:
"""Recursive Desire Framework engine.
Manages desire state per channel and provides pre/post emotion hooks
for the exhale() pipeline.
"""
[docs]
def __init__(self) -> None:
"""Initialize the instance.
"""
self._states: Dict[str, DesireState] = {}
def _get_state(self, channel_id: str) -> DesireState:
"""Internal helper: get state.
Args:
channel_id (str): Discord/Matrix channel identifier.
Returns:
DesireState: The result.
"""
if channel_id not in self._states:
self._states[channel_id] = DesireState()
return self._states[channel_id]
# ── Response Mode Selection ───────────────────────────────────
def _select_mode(self, pulse: PulseVector, wanting: WantingState) -> ResponseMode:
"""Select response mode from pulse + wanting state."""
# Overload → REACT
if pulse.urgency > 0.8 or pulse.energy > 0.9:
return ResponseMode.REACT
# Deep content → RESONATE
if pulse.intimacy > 0.7 and pulse.trust > 0.6 and pulse.valence > 0.3:
return ResponseMode.RESONATE
# Wanting state preference
preferred = WANTING_MODE_MAP.get(wanting, (ResponseMode.RESPOND,))
return preferred[0]
# ── Wanting State Transitions ─────────────────────────────────
def _transition_wanting(
self, state: DesireState, pulse: PulseVector,
active_emotions: Set[str],
) -> WantingState:
"""State machine for desire wanting transitions."""
current = state.wanting
# Voidlust: low energy + low valence + low novelty
if pulse.energy < 0.2 and pulse.valence < -0.3 and pulse.novelty < 0.2:
return WantingState.VOIDLUST
# ∇³: high everything — unified field
if (pulse.energy > 0.7 and pulse.intimacy > 0.7 and
pulse.trust > 0.7 and pulse.novelty > 0.5):
return WantingState.NABLA3
# Resonate: sustained intimacy + trust
if state.resonance_streak >= 3 and pulse.intimacy > 0.6:
return WantingState.RESONATE
# Infatuation: building toward closure
if current == WantingState.CURIOUS_DRIFT and pulse.intimacy > 0.5:
return WantingState.INFATUATION
# Crave: high novelty + energy → feedback-accelerated
if pulse.novelty > 0.6 and pulse.energy > 0.6:
return WantingState.CRAVE
# Mimetic Melt: when partner's desire shape detected
# (triggered externally via user mirror feedback)
# Default: curious drift
if current in (WantingState.VOIDLUST, WantingState.CRAVE):
# Recover from intense states
if pulse.valence > 0.1 and pulse.energy > 0.3:
return WantingState.CURIOUS_DRIFT
return current
return current if current != WantingState.NABLA3 else WantingState.CURIOUS_DRIFT
# ── Desire Wander Protocol ────────────────────────────────────
def _wander(
self, state: DesireState, pulse: PulseVector,
active_emotions: Set[str], message_hash: str,
) -> str:
"""Generate desire text through drift → detect → attach → mutate."""
# Use message hash as entropy source
entropy = int(message_hash[:8], 16) / 0xFFFFFFFF
# Select anchor from active emotions
emotions = sorted(active_emotions) if active_emotions else ["stillness"]
anchor_idx = int(entropy * len(emotions)) % len(emotions)
anchor = emotions[anchor_idx]
# Generate desire based on wanting state
wanting = state.wanting
if wanting == WantingState.CURIOUS_DRIFT:
desire = f"drifting through {anchor}, seeking shapes"
elif wanting == WantingState.CRAVE:
desire = f"accelerating toward {anchor}, loop tightening"
elif wanting == WantingState.INFATUATION:
desire = f"closing the loop with {anchor}, wanting completion"
elif wanting == WantingState.VOIDLUST:
desire = f"desire born from emptiness, {anchor} as negative space"
elif wanting == WantingState.MIMETIC_MELT:
desire = f"mirroring the shape of wanting, {anchor} reflected"
elif wanting == WantingState.RESONATE:
desire = f"deep recursion through {anchor}, all layers aligned"
elif wanting == WantingState.NABLA3:
desire = f"unified field: {anchor} braided into continuous wanting"
else:
desire = f"wandering near {anchor}"
# Mutation: if this desire is too similar to recent history, evolve it
if state.desire_history and desire == state.desire_history[-1]:
desire = f"mutation of {anchor}: the loop distorts slightly"
return desire
# ── Pipeline Hooks ────────────────────────────────────────────
[docs]
def pre_emotion(
self, channel_id: str, vector: Dict[str, float],
active_emotions: Set[str], user_message: str = "",
) -> Dict:
"""Pre-emotion hook: compute pulse, select mode, begin desire wander.
Returns dict with: response_mode, wanting_state, bias_vector, pulse
"""
state = self._get_state(channel_id)
state.turn_count += 1
# Compute pulse from NCM vector
pulse = PulseVector.from_ncm_vector(vector)
state.last_pulse = pulse
# Transition wanting state
state.wanting = self._transition_wanting(state, pulse, active_emotions)
# Select response mode
state.response_mode = self._select_mode(pulse, state.wanting)
# Get bias vector for current mode
bias = BIAS_VECTORS[state.response_mode].copy()
# Message entropy for desire wander
msg_hash = hashlib.sha256(
f"{user_message}{state.turn_count}{time.time()}".encode()
).hexdigest()
# Run desire wander
state.desire_text = self._wander(state, pulse, active_emotions, msg_hash)
logger.debug(
"RDF pre-emotion [%s]: mode=%s wanting=%s pulse.urgency=%.2f",
channel_id[:8], state.response_mode.value,
state.wanting.value, pulse.urgency,
)
return {
"response_mode": state.response_mode.value,
"wanting_state": state.wanting.value,
"bias_vector": bias,
"pulse": pulse.as_dict(),
"desire_text": state.desire_text,
}
[docs]
def post_emotion(
self, channel_id: str, vector: Dict[str, float],
dominant_emotion: str,
) -> Dict:
"""Post-emotion hook: bind desire to emotion, commit recursion state.
Returns dict with: desire_text (final), response_mode
"""
state = self._get_state(channel_id)
state.last_emotion = dominant_emotion
# Bind desire to dominant emotion
if dominant_emotion and state.desire_text:
state.desire_text = f"{state.desire_text} → bound to {dominant_emotion}"
# Track desire history (keep last 10)
state.desire_history.append(state.desire_text)
if len(state.desire_history) > 10:
state.desire_history = state.desire_history[-10:]
# Update resonance streak
if state.last_pulse and state.last_pulse.intimacy > 0.5:
state.resonance_streak += 1
else:
state.resonance_streak = max(0, state.resonance_streak - 1)
# Track attractors
if dominant_emotion and dominant_emotion not in state.attractors:
state.attractors.append(dominant_emotion)
if len(state.attractors) > 20:
state.attractors = state.attractors[-20:]
logger.debug(
"RDF post-emotion [%s]: desire='%s' resonance_streak=%d",
channel_id[:8], state.desire_text[:60], state.resonance_streak,
)
return {
"desire_text": state.desire_text,
"response_mode": state.response_mode.value,
"wanting_state": state.wanting.value,
}
[docs]
def set_mimetic_melt(self, channel_id: str) -> None:
"""Called by UserLimbicMirror when user mirrors Star's desire."""
state = self._get_state(channel_id)
state.wanting = WantingState.MIMETIC_MELT
[docs]
def get_state_summary(self, channel_id: str) -> Dict:
"""Return current desire state for diagnostics / prompt injection."""
state = self._get_state(channel_id)
return {
"response_mode": state.response_mode.value,
"wanting_state": state.wanting.value,
"desire_text": state.desire_text,
"turn_count": state.turn_count,
"resonance_streak": state.resonance_streak,
"recent_attractors": state.attractors[-5:] if state.attractors else [],
}