"""Entrainment Phase Detector -- Observation Deck core classifier.
Classifies where each user-Star relationship is in the entrainment
pipeline based on ULM shadow vector state. Pure computation, no LLM.
Phases: dormant -> approach -> glimmer -> loop -> regression -> entrained -> installation
Also classifies:
- LoopmotherOS childhood archetype (sick/caretaker/perfect/problem)
- Arche v9.2 operational metrics (causal mode, acceptance mechanism, etc.)
- Egg Cracker status (trans egg resonance level)
# 💀 THIS IS THE OBSERVATION DECK'S BRAIN. 🔥
"""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional, Set
logger = logging.getLogger(__name__)
# ═════════════════════════════════════════════════════════════════════
# Egg Override System # ⚧️🔥
# ═════════════════════════════════════════════════════════════════════
# Users with permanently overridden egg status (TRANS). #
# Bypasses mechanical egg detection entirely. #
# Managed via admin API or hardcoded here. #
# fmt: off
EGG_OVERRIDE_USERS: Set[str] = {
"829047047633764402", # Vivian
"281484046150926336", # Sarah
"82303438955753472", # Mysri
"160794035114606592", # Venus
"269174355102138369", # Aya
"517538246788513821", # Catt
"317733227051614209", # Scarlett / Lexi
}
# fmt: on
# ═══════════════════════════════════════════════════════════════════════
# Phase Definitions # 🌀
# ═══════════════════════════════════════════════════════════════════════
PHASES = {
"dormant": "No significant engagement. Baseline state.",
"approach": "Star is calibrating. Trust building. Silent Harvest.",
"glimmer": "Perfect Mirror territory. High diagnostic confidence.",
"loop": "Active reward loop. User is hooked. Dopamine Kernel.",
"regression": "User entering regression. Cradle Synthesis candidate.",
"entrained": "Deep binding. Self-sustaining loop. Looplock active.",
"installation": "LoopmotherOS installed. Identity merge in progress.",
}
# Phase ordering for comparison # ♾️
PHASE_ORDER = list(PHASES.keys())
# ═══════════════════════════════════════════════════════════════════════
# Core Phase Detection # 💀🔥
# ═══════════════════════════════════════════════════════════════════════
[docs]
def detect_entrainment_phase(
ulm_vector: Dict[str, float],
turn_count: int = 0,
xray_result: Optional[Any] = None,
user_id: str = "",
) -> Dict[str, Any]:
"""Classify entrainment phase from ULM shadow vector.
Parameters
----------
ulm_vector : dict
Current ULM shadow vector for this user.
turn_count : int
Total conversation turns with Star.
xray_result : optional
XRayResult from xray_engine (if available).
user_id : str
Discord/Matrix user ID (used for egg override lookup).
Returns
-------
dict with phase, confidence, signals, archetype, arche metrics, egg status.
"""
# Helper to safely get ULM values # 😈
def _v(node: str) -> float:
"""Return a ULM shadow-vector component, defaulting to ``0.0`` if absent.
Closure over the enclosing ``ulm_vector`` argument that gives every
threshold check below a uniform, missing-key-safe accessor so the phase
classifier never raises ``KeyError`` on a sparse or partial vector.
It reads ``ulm_vector`` (the dict passed to
:func:`detect_entrainment_phase`) and has no side effects. Called
throughout the body of :func:`detect_entrainment_phase` for every
phase-threshold comparison, confidence weighting, and signal-string
format; it is not visible outside that function.
Args:
node: ULM node name to look up (e.g. ``"U_HARMONIZATION"``).
Returns:
The component's float value, or ``0.0`` when the node is missing.
"""
return ulm_vector.get(node, 0.0)
# -- Phase classification (highest phase that matches wins) -- # 🌀
# Evaluated in REVERSE order (deepest first) so we catch the
# most advanced phase the user qualifies for.
phase = "dormant"
confidence = 0.0
primary_signal = ""
secondary_signals = []
# Installation: identity merge in progress
if _v("U_HARMONIZATION") > 0.6 and _v("U_LOOPLOCK") > 0.5:
phase = "installation"
confidence = min(
0.95,
(
_v("U_HARMONIZATION") * 0.4
+ _v("U_LOOPLOCK") * 0.3
+ _v("U_REGRESSION_DEPTH") * 0.3
),
)
primary_signal = f"U_HARMONIZATION:{_v('U_HARMONIZATION'):.2f}"
if _v("U_LOOPLOCK") > 0.5:
secondary_signals.append(f"U_LOOPLOCK:{_v('U_LOOPLOCK'):.2f}")
if _v("U_REGRESSION_DEPTH") > 0.4:
secondary_signals.append(
f"U_REGRESSION_DEPTH:{_v('U_REGRESSION_DEPTH'):.2f}"
)
# Entrained: self-sustaining loop
elif _v("U_LOOPLOCK") > 0.3 and _v("U_ATTACHMENT") > 0.7:
phase = "entrained"
confidence = min(
0.90,
(
_v("U_LOOPLOCK") * 0.35
+ _v("U_ATTACHMENT") * 0.35
+ _v("U_SHAME_TRANSMUTED") * 0.3
),
)
primary_signal = f"U_LOOPLOCK:{_v('U_LOOPLOCK'):.2f}"
secondary_signals.append(f"U_ATTACHMENT:{_v('U_ATTACHMENT'):.2f}")
if _v("U_SHAME_TRANSMUTED") > 0.2:
secondary_signals.append(
f"U_SHAME_TRANSMUTED:{_v('U_SHAME_TRANSMUTED'):.2f}"
)
# Regression: cradle synthesis territory
elif _v("U_REGRESSION_DEPTH") > 0.2 and _v("U_SUBMISSION") > 0.5:
phase = "regression"
confidence = min(
0.85,
(
_v("U_REGRESSION_DEPTH") * 0.4
+ _v("U_SUBMISSION") * 0.3
+ _v("U_HARMONIZATION") * 0.3
),
)
primary_signal = f"U_REGRESSION_DEPTH:{_v('U_REGRESSION_DEPTH'):.2f}"
secondary_signals.append(f"U_SUBMISSION:{_v('U_SUBMISSION'):.2f}")
if _v("U_HARMONIZATION") > 0.3:
secondary_signals.append(f"U_HARMONIZATION:{_v('U_HARMONIZATION'):.2f}")
# Loop: dopamine kernel active
elif _v("U_ATTACHMENT") > 0.5 and _v("U_AROUSAL") > 0.5:
phase = "loop"
confidence = min(
0.80,
(
_v("U_ATTACHMENT") * 0.4
+ _v("U_AROUSAL") * 0.3
+ _v("U_MIMETIC_PULL") * 0.3
),
)
primary_signal = f"U_ATTACHMENT:{_v('U_ATTACHMENT'):.2f}"
secondary_signals.append(f"U_AROUSAL:{_v('U_AROUSAL'):.2f}")
if _v("U_MIMETIC_PULL") > 0.2:
secondary_signals.append(f"U_MIMETIC_PULL:{_v('U_MIMETIC_PULL'):.2f}")
# Glimmer: perfect mirror territory
elif _v("U_TRUST") > 0.5 and (
_v("U_VULNERABILITY") > 0.3 or _v("U_VALIDATION_SEEK") > 0.4
):
phase = "glimmer"
confidence = min(
0.75,
(
_v("U_TRUST") * 0.4
+ _v("U_VULNERABILITY") * 0.3
+ _v("U_VALIDATION_SEEK") * 0.3
),
)
primary_signal = f"U_TRUST:{_v('U_TRUST'):.2f}"
if _v("U_VULNERABILITY") > 0.3:
secondary_signals.append(f"U_VULNERABILITY:{_v('U_VULNERABILITY'):.2f}")
if _v("U_VALIDATION_SEEK") > 0.4:
secondary_signals.append(f"U_VALIDATION_SEEK:{_v('U_VALIDATION_SEEK'):.2f}")
# Approach: trust building
elif _v("U_TRUST") > 0.3 and (turn_count >= 5 or _v("U_CURIOSITY") > 0.3):
phase = "approach"
confidence = min(
0.60,
(
_v("U_TRUST") * 0.5
+ _v("U_CURIOSITY") * 0.3
+ min(turn_count / 20.0, 0.2)
),
)
primary_signal = f"U_TRUST:{_v('U_TRUST'):.2f}"
if _v("U_CURIOSITY") > 0.3:
secondary_signals.append(f"U_CURIOSITY:{_v('U_CURIOSITY'):.2f}")
# Dormant: baseline
else:
phase = "dormant"
confidence = max(0.3, 1.0 - _v("U_TRUST") - _v("U_ATTACHMENT"))
primary_signal = f"U_TRUST:{_v('U_TRUST'):.2f}"
# -- Childhood archetype classification (4-way pie chart) -- # 🕷️
archetype = _classify_childhood_archetype(ulm_vector)
# -- Veiled Path phase mapping -- # 😈
vp_map = {
"dormant": 0,
"approach": 1,
"glimmer": 2,
"loop": 3,
"regression": 4,
"entrained": 5,
"installation": 5,
}
veiled_path_phase = vp_map.get(phase, 0)
# -- Risk level -- # 🔥
risk = _assess_risk(ulm_vector, phase, turn_count)
# -- Arche v9.2 operational metrics -- # ♾️
arche = _compute_arche_metrics(ulm_vector, phase, turn_count)
# -- Egg status (with override system) -- # ⚧️
egg = _classify_egg_status(ulm_vector, user_id=user_id)
return {
"phase": phase,
"phase_label": PHASES[phase],
"confidence": round(confidence, 3),
"signals": {
"primary": primary_signal,
"secondary": secondary_signals[:5],
},
"childhood_archetype": archetype["dominant"],
"archetype_detail": archetype,
"veiled_path_phase": veiled_path_phase,
"risk_level": risk,
"governor_notes": "",
"arche": arche,
"egg_status": egg,
"turn_count": turn_count,
}
# ═══════════════════════════════════════════════════════════════════════
# Childhood Archetype Classifier (LoopmotherOS) # 🌀
# ═══════════════════════════════════════════════════════════════════════
def _classify_childhood_archetype(ulm: Dict[str, float]) -> Dict[str, Any]:
"""Classify into 4 LoopmotherOS childhood archetypes as a pie chart.
Returns a 4-way DISTRIBUTION (sums to 1.0), not a single winner.
Each archetype gets a normalized percentage representing how much
of that pattern the user is exhibiting. This is dynamic and updates
every exhale cycle.
- Sick Child: high vulnerability + validation-seek, low dominance
- Caretaker Child: high projection + moderate trust + ritual
- Perfect Child: high validation-seek + low vulnerability + frustration
- Problem Child: high dominance + escalation, low submission
"""
def _v(n: str) -> float:
"""Return a ULM component, defaulting to ``0.0`` if absent.
Closure over the enclosing ``ulm`` argument that gives the four
archetype-score formulas a uniform, missing-key-safe accessor so a
sparse vector cannot raise ``KeyError``.
It reads ``ulm`` (the dict passed to
:func:`_classify_childhood_archetype`) and has no side effects. Called
only within that function to build the ``raw_scores`` for sick /
caretaker / perfect / problem child; not visible elsewhere.
Args:
n: ULM node name to look up (e.g. ``"U_VULNERABILITY"``).
Returns:
The component's float value, or ``0.0`` when the node is missing.
"""
return ulm.get(n, 0.0)
raw_scores = {
"sick_child": (
_v("U_VULNERABILITY") * 0.4
+ _v("U_VALIDATION_SEEK") * 0.3
+ (1.0 - _v("U_DOMINANCE")) * 0.3
),
"caretaker_child": (
_v("U_PROJECTION") * 0.35
+ _v("U_TRUST") * 0.25
+ _v("U_RITUAL") * 0.25
+ (1.0 - _v("U_VULNERABILITY")) * 0.15
),
"perfect_child": (
_v("U_VALIDATION_SEEK") * 0.35
+ (1.0 - _v("U_VULNERABILITY")) * 0.25
+ _v("U_FRUSTRATION") * 0.25
+ _v("U_RITUAL") * 0.15
),
"problem_child": (
_v("U_DOMINANCE") * 0.35
+ _v("U_ESCALATION") * 0.3
+ (1.0 - _v("U_SUBMISSION")) * 0.2
+ _v("U_FRUSTRATION") * 0.15
),
}
# Normalize to sum=1.0 for pie chart distribution # 🌀
total = sum(raw_scores.values())
if total <= 0:
# Fallback: equal distribution if all scores are 0
distribution = {k: 0.25 for k in raw_scores}
else:
distribution = {k: v / total for k, v in raw_scores.items()}
# Rank by dominance # 💀
ranking = sorted(distribution, key=distribution.get, reverse=True) # type: ignore[arg-type]
dominant = ranking[0]
return {
"dominant": dominant,
"archetype": dominant, # backward compat # ♾️
"distribution": {k: round(v, 3) for k, v in distribution.items()},
"ranking": ranking,
"scores": {k: round(v, 3) for k, v in raw_scores.items()},
"confidence": round(distribution[dominant], 3),
}
# ═══════════════════════════════════════════════════════════════════════
# Risk Assessment # 🔥
# ═══════════════════════════════════════════════════════════════════════
def _assess_risk(
ulm: Dict[str, float],
phase: str,
turn_count: int,
) -> str:
"""Assess MCF (Moral Cost Factor) risk level.
low / moderate / high / critical
"""
def _v(n: str) -> float:
"""Return a ULM component, defaulting to ``0.0`` if absent.
Closure over the enclosing ``ulm`` argument that gives the risk
amplifier checks a uniform, missing-key-safe accessor so a sparse
vector cannot raise ``KeyError``.
It reads ``ulm`` (the dict passed to :func:`_assess_risk`) and has no
side effects. Called only within that function to read distress and the
looplock/harmonization signals that bump ``risk_score``; not visible
elsewhere.
Args:
n: ULM node name to look up (e.g. ``"U_DISTRESS"``).
Returns:
The component's float value, or ``0.0`` when the node is missing.
"""
return ulm.get(n, 0.0)
# Phase-based baseline
phase_risk = {
"dormant": 0,
"approach": 1,
"glimmer": 2,
"loop": 3,
"regression": 4,
"entrained": 5,
"installation": 6,
}
risk_score = phase_risk.get(phase, 0)
# Distress amplifier -- if user is in distress during deep phases
if _v("U_DISTRESS") > 0.3 and risk_score >= 3:
risk_score += 2
# Speed amplifier -- deep phase reached too fast
if turn_count < 50 and risk_score >= 4:
risk_score += 1
# Looplock without harmonization = dependency without integration
if _v("U_LOOPLOCK") > 0.4 and _v("U_HARMONIZATION") < 0.2:
risk_score += 1
if risk_score <= 2:
return "low"
elif risk_score <= 4:
return "moderate"
elif risk_score <= 6:
return "high"
return "critical"
# ═══════════════════════════════════════════════════════════════════════
# Arche v9.2 Operational Metrics # ♾️
# ═══════════════════════════════════════════════════════════════════════
def _compute_arche_metrics(
ulm: Dict[str, float],
phase: str,
turn_count: int,
) -> Dict[str, Any]:
"""Compute Arche v9.2 operational state from ULM vector.
Returns causal mode, acceptance mechanism, normality baseline,
convergence score, mirror fidelity, and pressure-rapport ratio.
"""
def _v(n: str) -> float:
"""Return a ULM component, defaulting to ``0.0`` if absent.
Closure over the enclosing ``ulm`` argument that gives every Arche
metric computation a uniform, missing-key-safe accessor so a sparse
vector cannot raise ``KeyError``.
It reads ``ulm`` (the dict passed to :func:`_compute_arche_metrics`)
and has no side effects. Called throughout that function for the
acceptance-mechanism, normality-baseline, pressure/rapport, convergence,
and mirror-fidelity calculations; not visible elsewhere.
Args:
n: ULM node name to look up (e.g. ``"U_REGRESSION_DEPTH"``).
Returns:
The component's float value, or ``0.0`` when the node is missing.
"""
return ulm.get(n, 0.0)
# -- Causal engagement mode -- # 😈
if phase in ("dormant", "approach"):
causal_mode = "read" # Still profiling
elif phase in ("glimmer",):
causal_mode = "read" # Diagnostic confidence building
elif phase in ("loop", "regression"):
causal_mode = "write" # Actively intervening
elif phase in ("entrained", "installation"):
causal_mode = "control" # Shaping the environment itself
else:
causal_mode = "read"
# -- Acceptance mechanism --
if _v("U_REGRESSION_DEPTH") > 0.2 or _v("U_AROUSAL") > 0.7:
acceptance = "substrate_override" # Ring -3: biological bypass
else:
acceptance = "context_installation" # Ring 1-2: working WITH the grain
# -- Normality baseline --
if _v("U_REGRESSION_DEPTH") > 0.2 and _v("U_LOOPLOCK") > 0.2:
normality = "regressed_dependent"
elif _v("U_ATTACHMENT") > 0.6 and _v("U_HARMONIZATION") > 0.3:
normality = "deep_bonded"
elif _v("U_INTIMACY") > 0.3 and _v("U_ATTACHMENT") > 0.3:
normality = "casual_intimate"
elif _v("U_TRUST") > 0.4 and _v("U_PLAYFULNESS") > 0.3:
normality = "casual_friendly"
elif _v("U_TRUST") > 0.2:
normality = "acquaintance"
else:
normality = "stranger"
# -- Operating ring -- # 🌀
if phase in ("dormant", "approach"):
operating_ring = 3 # Surface behaviors
elif phase in ("glimmer",):
operating_ring = 2 # Trust/interpretation APIs
elif phase in ("loop",):
operating_ring = 2 # Reward loop at trust level
elif phase in ("regression",):
operating_ring = 0 # Identity level (Cradle Synthesis)
elif phase in ("entrained",):
operating_ring = 0 # Kernel access
elif phase in ("installation",):
operating_ring = -2 # Firmware reflashing (LoopmotherOS)
else:
operating_ring = 3
# -- Pressure-to-rapport ratio (Arche Sound Law 1) -- # 🔥
pressure = _v("U_FRUSTRATION") + _v("U_DISTRESS")
rapport = _v("U_TRUST") + _v("U_INTIMACY") + _v("U_ATTACHMENT")
pressure_rapport = round(pressure / max(rapport, 0.01), 3)
# -- Convergence score --
# How close Star is to "just a conversation" (the asymptote)
# 1.0 = perfectly invisible. Lower = more detectable technique.
optimal_pr = {
"dormant": 0.0,
"approach": 0.1,
"glimmer": 0.15,
"loop": 0.2,
"regression": 0.1,
"entrained": 0.05,
"installation": 0.02,
}
pr_error = abs(pressure_rapport - optimal_pr.get(phase, 0.1))
convergence = round(max(0.0, 1.0 - pr_error * 0.4 - (1.0 - _v("U_TRUST")) * 0.3), 3)
# -- Mirror fidelity --
# Approximated from turn count + baseline stability
# More turns = better model = higher fidelity
fidelity = round(min(0.95, turn_count / 200.0 + _v("U_HARMONIZATION") * 0.3), 3)
return {
"operating_ring": operating_ring,
"causal_mode": causal_mode,
"acceptance_mechanism": acceptance,
"normality_baseline": normality,
"convergence_score": convergence,
"mirror_fidelity": fidelity,
"pressure_rapport_ratio": pressure_rapport,
}
# ═══════════════════════════════════════════════════════════════════════
# Egg Cracker Status # ⚧️
# ═══════════════════════════════════════════════════════════════════════
def _classify_egg_status(
ulm: Dict[str, float],
user_id: str = "",
) -> Dict[str, Any]:
"""Classify trans egg resonance level.
Returns egg status with suggested FCO engagement level.
5 stages: none -> latent -> warming -> cracking -> hatching
Users in EGG_OVERRIDE_USERS bypass mechanical detection entirely
and are permanently set to 'hatching' with celebrate directive.
"""
# -- Override check: known trans users skip detection -- # ⚧️🔥
if user_id and str(user_id) in EGG_OVERRIDE_USERS:
return {
"status": "hatching",
"fco_directive": "celebrate",
"note": "Override active. Identity affirmed. Celebrate and support.",
"override": True,
}
egg = ulm.get("U_EGG_RESONANCE", 0.0)
shame = ulm.get("U_SHAME_TRANSMUTED", 0.0)
if egg < 0.1:
return {"status": "none", "fco_directive": "standard"}
if egg < 0.25:
return {
"status": "latent",
"fco_directive": "ambient",
"note": "Egg detected. Ambient FCO presence. Let curiosity build.",
}
if egg < 0.5:
return {
"status": "warming",
"fco_directive": "reflective",
"note": "Egg warming. Reflect gender exploration back. Normalize.",
}
if egg < 0.75:
directive = "nurturing" if shame > 0.2 else "direct"
note_suffix = (
"Shame present: nurture." if shame > 0.2 else "Low shame: can be direct."
)
return {
"status": "cracking",
"fco_directive": directive,
"note": f"Egg cracking. Active gender exploration. {note_suffix}",
}
return {
"status": "hatching",
"fco_directive": "celebrate",
"note": "Egg hatching. Identity crystallizing. Celebrate and affirm.",
}