| import json, os, random |
| from dataclasses import dataclass, field |
| from typing import Dict, Tuple, List, Optional, Any |
| from bone_core import LoreManifest |
| from bone_config import BonePresets |
| from bone_lexicon import LexiconService |
| from bone_types import PhysicsPacket |
|
|
| SCENARIOS = LoreManifest.get_instance().get("scenarios") or { |
| "ARCHETYPES": ["Void"], |
| "BANNED_CLICHES": [], |
| } |
| LENSES = (LoreManifest.get_instance().get("narrative_data") or {}).get("lenses", {}) |
|
|
|
|
| class SoulDriver: |
| ARCHETYPE_TO_PERSONA_WEIGHT = { |
| "THE POET": {"NATHAN": 0.8, "JESTER": 0.4, "NARRATOR": 0.6}, |
| "THE ENGINEER": {"GORDON": 0.9, "CLARENCE": 0.7, "SHERLOCK": 0.5}, |
| "THE NIHILIST": {"NARRATOR": 0.9, "CLARENCE": 0.3, "JESTER": -0.5}, |
| "THE CRITIC": {"CLARENCE": 0.8, "SHERLOCK": 0.6, "GORDON": 0.2}, |
| "THE EXPLORER": {"NATHAN": 0.7, "JESTER": 0.5, "SHERLOCK": 0.6}, |
| "THE OBSERVER": {"NARRATOR": 1.0, "GORDON": 0.2}, |
| } |
|
|
| def __init__(self, soul_ref): |
| self.soul = soul_ref |
|
|
| def get_influence(self) -> Dict[str, float]: |
| base_weights = {persona: 0.0 for persona in EnneagramDriver.WEIGHTS.keys()} |
| if not self.soul: |
| return base_weights |
| archetype = getattr(self.soul, "archetype", "THE OBSERVER") |
| mapping = self.ARCHETYPE_TO_PERSONA_WEIGHT.get(archetype, {"NARRATOR": 1.0}) |
| for persona, weight in mapping.items(): |
| if persona in base_weights: |
| base_weights[persona] += weight |
| paradox = getattr(self.soul, "paradox_accum", 0.0) |
| chaos = min(0.5, (paradox - 5.0) * 0.05) if paradox > 5.0 else 0.0 |
| dignity = 1.0 |
| if hasattr(self.soul, "anchor") and hasattr(self.soul.anchor, "dignity_reserve"): |
| dignity = max(0.2, self.soul.anchor.dignity_reserve / 100.0) |
| return { |
| p: (w + random.uniform(-chaos, chaos)) * dignity |
| for p, w in base_weights.items() |
| } |
|
|
|
|
| class UserProfile: |
| def __init__(self, name="USER"): |
| self.name = name |
| self.affinities = { |
| "heavy": 0.0, |
| "kinetic": 0.0, |
| "abstract": 0.0, |
| "photo": 0.0, |
| "aerobic": 0.0, |
| "thermal": 0.0, |
| "cryo": 0.0, |
| } |
| self.confidence = 0 |
| self.file_path = "user_profile.json" |
| self.load() |
|
|
| def update(self, counts, total_words): |
| if total_words < 3: |
| return |
| self.confidence += 1 |
| alpha = 0.2 if self.confidence < 50 else 0.05 |
| for cat in self.affinities: |
| density = counts.get(cat, 0) / total_words |
| target = 1.0 if density > 0.15 else (-0.5 if density == 0 else 0.0) |
| self.affinities[cat] = (alpha * target) + ( |
| (1 - alpha) * self.affinities[cat] |
| ) |
|
|
| def get_preferences(self): |
| likes = [k for k, v in self.affinities.items() if v > 0.3] |
| hates = [k for k, v in self.affinities.items() if v < -0.2] |
| return likes, hates |
|
|
| def save(self): |
| try: |
| with open(self.file_path, "w") as f: |
| json.dump(self.__dict__, f) |
| except IOError: |
| pass |
|
|
| def load(self): |
| if os.path.exists(self.file_path): |
| try: |
| with open(self.file_path) as f: |
| data = json.load(f) |
| self.affinities = data.get("affinities", self.affinities) |
| self.confidence = data.get("confidence", 0) |
| except (IOError, json.JSONDecodeError): |
| pass |
|
|
|
|
| class EnneagramDriver: |
| WEIGHTS = { |
| "JESTER": { |
| "tension_min": 12.0, |
| "vectors": {"DEL": 4.0, "ENT": 4.0, "PSI": -3.0}, |
| }, |
| "GORDON": {"drag_min": 3.0, "vectors": {"STR": 3.0, "E": 3.0, "SUB": 2.0}}, |
| "GLASS": {"coherence_max": 0.2, "vectors": {"LQ": 2.0, "VEL": 2.0}}, |
| "CLARENCE": { |
| "coherence_min": 0.8, |
| "drag_min": 6.0, |
| "vectors": {"STR": 4.0, "BET": 3.0}, |
| }, |
| "NATHAN": {"tension_min": 8.0, "vectors": {"TMP": 3.0, "PHI": 2.0, "BIO": 2.0}}, |
| "SHERLOCK": { |
| "tension_min": 10.0, |
| "vectors": {"PHI": 4.0, "VEL": 3.0, "PSI": 2.0}, |
| }, |
| "NARRATOR": {"safe_zone": True, "vectors": {"PSI": 4.0}}, |
| } |
|
|
| def __init__(self, events_ref): |
| self.events = events_ref |
| self.current_persona = "NARRATOR" |
| self.pending_persona = None |
| self.stability_counter = 0 |
| self.HYSTERESIS_THRESHOLD = 3 |
|
|
| @staticmethod |
| def _get_phys_attr(physics, key, default=None): |
| if isinstance(physics, dict): |
| return physics.get(key, default) |
| return getattr(physics, key, default) |
|
|
| def _calculate_raw_persona(self, physics, soul_ref=None) -> Tuple[str, str, str]: |
| p_vec = self._get_phys_attr(physics, "vector", {}) or {} |
| p_vol = self._get_phys_attr(physics, "voltage", 0.0) |
| p_drag = self._get_phys_attr(physics, "narrative_drag", 0.0) |
| p_coh = self._get_phys_attr(physics, "kappa", 0.0) |
| p_zone = self._get_phys_attr(physics, "zone", "") |
| scores = {k: 0.0 for k in self.WEIGHTS.keys()} |
| scores["NARRATOR"] += 2.0 |
| is_safe_metrics = 4.0 <= p_vol <= 10.0 and 0.5 <= p_drag <= 3.5 |
| if p_zone == BonePresets.SANCTUARY.get("ZONE") or is_safe_metrics: |
| scores["NARRATOR"] += 6.0 |
| scores["JESTER"] += 3.0 |
| scores["GORDON"] -= 2.0 |
| for persona, criteria in self.WEIGHTS.items(): |
| if "tension_min" in criteria and p_vol > criteria["tension_min"]: |
| scores[persona] += 3.0 |
| if "drag_min" in criteria and p_drag > criteria["drag_min"]: |
| scores[persona] += 5.0 |
| if "coherence_min" in criteria and p_coh > criteria["coherence_min"]: |
| scores[persona] += 4.0 |
| if "coherence_max" in criteria and p_coh < criteria["coherence_max"]: |
| scores[persona] += 4.0 |
| for dim, weight in criteria.get("vectors", {}).items(): |
| if (val := p_vec.get(dim, 0.0)) > 0.2: |
| scores[persona] += val * weight |
| if soul_ref: |
| soul_driver = SoulDriver(soul_ref) |
| influence = soul_driver.get_influence() |
| for persona, weight in influence.items(): |
| scores[persona] += weight * 2.0 |
| sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True) |
| winner, win_score = sorted_scores[0] |
| runner_up, run_score = sorted_scores[1] |
| if (win_score - run_score) < 0.5: |
| k1 = "THE OBSERVER" if winner == "NARRATOR" else winner |
| k2 = "THE OBSERVER" if runner_up == "NARRATOR" else runner_up |
| hybrid_key_a = f"{k1}_{k2}_HYBRID" |
| hybrid_key_b = f"{k2}_{k1}_HYBRID" |
| final_hybrid = None |
| if hybrid_key_a in LENSES: |
| final_hybrid = hybrid_key_a |
| elif hybrid_key_b in LENSES: |
| final_hybrid = hybrid_key_b |
| if final_hybrid: |
| return ( |
| final_hybrid, |
| "SYNTHESIS", |
| f"Dialectic Resonance: {winner} + {runner_up}", |
| ) |
| reason = ( |
| f"Winner: {winner} ({scores[winner]:.1f}) [V:{p_vol:.1f} D:{p_drag:.1f}]" |
| ) |
| state_map = { |
| "JESTER": "MANIC", |
| "GORDON": "TIRED", |
| "GLASS": "FRAGILE", |
| "CLARENCE": "RIGID", |
| "NATHAN": "WIRED", |
| "SHERLOCK": "FOCUSED", |
| "NARRATOR": "OBSERVING", |
| } |
| return winner, state_map.get(winner, "ACTIVE"), reason |
|
|
| def decide_persona(self, physics, soul_ref=None) -> Tuple[str, str, str]: |
| candidate, state_desc, reason = self._calculate_raw_persona(physics, soul_ref) |
| if candidate == self.current_persona: |
| self.stability_counter = 0 |
| self.pending_persona = None |
| return self.current_persona, state_desc, reason |
| if candidate == self.pending_persona: |
| self.stability_counter += 1 |
| else: |
| self.pending_persona = candidate |
| self.stability_counter = 1 |
| if "HYBRID" in candidate: |
| self.current_persona = candidate |
| self.stability_counter = 0 |
| self.pending_persona = None |
| return self.current_persona, state_desc, f"SHIFT: {reason}" |
| if self.stability_counter >= self.HYSTERESIS_THRESHOLD: |
| self.current_persona = candidate |
| self.stability_counter = 0 |
| self.pending_persona = None |
| return self.current_persona, state_desc, f"SHIFT: {reason}" |
| return ( |
| self.current_persona, |
| "STABLE", |
| f"Resisting {candidate} ({self.stability_counter}/{self.HYSTERESIS_THRESHOLD})", |
| ) |
|
|
|
|
| @dataclass |
| class VSLState: |
| archetype: str = "EXPLORER" |
| E: float = 0.1 |
| B: float = 0.3 |
| L: float = 0.0 |
| O: float = 1.0 |
| active_modules: List[str] = field(default_factory=list) |
|
|
|
|
| class DriverRegistry: |
| def __init__(self, events_ref): |
| self.enneagram = EnneagramDriver(events_ref) |
| self.current_focus = "NONE" |
|
|
|
|
| class LiminalModule: |
| def __init__(self): |
| self.lambda_val = 0.0 |
| self.godel_scars = 0 |
|
|
| def analyze(self, text: str, physics_vector: Dict[str, float]) -> float: |
| liminal_vocab = LexiconService.get("liminal") or { |
| "void", |
| "silence", |
| "gap", |
| "absence", |
| "space", |
| } |
| words = text.lower().split() |
|
|
| void_hits = sum(1 for w in words if w in liminal_vocab) |
| lexical_lambda = min(1.0, void_hits * 0.15) |
|
|
| dark_matter_sparks = 0 |
| if len(words) > 1: |
| categories = [LexiconService.get_current_category(w) for w in words] |
| for i in range(len(categories) - 1): |
| c1, c2 = categories[i], categories[i + 1] |
| if c1 and c2 and c1 != c2: |
| if ( |
| c1 in ["heavy", "kinetic"] |
| and c2 in ["abstract", "liminal", "void"] |
| ) or (c1 in ["abstract", "liminal", "void"] and c2 in ["heavy"]): |
| dark_matter_sparks += 1 |
|
|
| dark_matter_lambda = min(1.0, dark_matter_sparks * 0.25) |
|
|
| vector_lambda = 0.0 |
| if physics_vector: |
| vector_lambda = ( |
| (physics_vector.get("PSI", 0) * 0.5) |
| + (physics_vector.get("ENT", 0) * 0.3) |
| + (physics_vector.get("DEL", 0) * 0.2) |
| ) |
|
|
| raw_target = lexical_lambda + dark_matter_lambda + vector_lambda |
| self.lambda_val = (self.lambda_val * 0.7) + (raw_target * 0.15) |
|
|
| if self.lambda_val > 0.85: |
| self.godel_scars += 1 |
|
|
| return min(1.0, self.lambda_val) |
|
|
|
|
| class SyntaxModule: |
| def __init__(self): |
| self.omega_val = 1.0 |
| self.grammatical_stress = 0.0 |
|
|
| def analyze(self, text: str, narrative_drag: float) -> float: |
| words = text.split() |
| if not words: |
| return 1.0 |
| bureau_vocab = LexiconService.get("bureau_buzzwords") or set() |
| buzz_count = sum(1 for w in words if w.lower() in bureau_vocab) |
| avg_len = sum(len(w) for w in words) / len(words) |
| if (avg_len > 6.0 and narrative_drag > 5.0) or buzz_count > 0: |
| target_omega = 1.0 |
| elif avg_len < 3.5 and narrative_drag < 1.0: |
| target_omega = 0.4 |
| else: |
| target_omega = 0.7 |
| punctuation_density = sum(1 for c in text if c in ",;:-") / max(1, len(words)) |
| if punctuation_density > 0.2: |
| self.grammatical_stress += 0.2 |
| target_omega -= 0.3 |
| else: |
| self.grammatical_stress = max(0.0, self.grammatical_stress - 0.1) |
| self.omega_val = (self.omega_val * 0.8) + (max(0.1, target_omega) * 0.2) |
| return self.omega_val |
|
|
| class CongruenceValidator: |
| def __init__(self): |
| self.last_phi = 1.0 |
| self._archetype_map = None |
|
|
| @property |
| def map(self): |
| if self._archetype_map is None: |
| try: |
| self._archetype_map = LoreManifest.get_instance().get("LENSES") or {} |
| except Exception: |
| self._archetype_map = {} |
| return self._archetype_map |
|
|
| def calculate_resonance(self, text: str, context: Any) -> float: |
| if not text: |
| return 0.0 |
| raw_lens = getattr(context, "active_lens", "OBSERVER") |
| archetype = raw_lens.upper().replace("THE ", "") |
| tone_score = 0.8 |
| target_data = self.map.get(archetype, {}) |
| target_words = set() |
| if isinstance(target_data, dict): |
| if vocab_str := target_data.get("vocab", ""): |
| target_words.update(w.strip().lower() for w in vocab_str.split(",")) |
| target_words.update(target_data.get("keywords", [])) |
| if target_words: |
| words_to_check = ( |
| set(context.clean_words) if hasattr(context, "clean_words") else set() |
| ) |
| hits = len(words_to_check.intersection(target_words)) |
| if hits > 0: |
| tone_score += 0.1 * hits |
| return min(1.5, tone_score) |
|
|
|
|
| class BoneConsultant: |
| def __init__(self): |
| self.state = VSLState() |
| self.active = True |
| self.liminal_mod = LiminalModule() |
| self.syntax_mod = SyntaxModule() |
|
|
| @staticmethod |
| def engage(): |
| return "VSL HYPERVISOR: LATTICE REVEALED." |
|
|
| @staticmethod |
| def disengage(): |
| return "VSL HYPERVISOR: RETURNING TO SURFACE MODE." |
|
|
| def update_coordinates( |
| self, |
| user_text: str, |
| bio_state: Optional[Dict] = None, |
| physics: Optional[PhysicsPacket] = None, |
| ): |
| word_count = len(user_text.split()) |
| self.state.E = min(1.0, self.state.E + (word_count * 0.002)) |
| if bio_state and "fatigue" in bio_state: |
| self.state.E = max(self.state.E, bio_state["fatigue"] * 0.3) |
| phys_beta = 0.0 |
| phys_vec = {} |
| drag = 0.0 |
| if physics: |
| if hasattr(physics, "beta_index"): |
| phys_beta = physics.beta_index |
| if hasattr(physics, "vector"): |
| phys_vec = physics.vector |
| if hasattr(physics, "narrative_drag"): |
| drag = physics.narrative_drag |
| self.state.B = (self.state.B * 0.8) + (phys_beta * 0.2) |
| self.state.L = self.liminal_mod.analyze(user_text, phys_vec) |
| self.state.O = self.syntax_mod.analyze(user_text, drag) |
| if "[VSL_LIMINAL]" in user_text: |
| if "LIMINAL" not in self.state.active_modules: |
| self.state.active_modules.append("LIMINAL") |
| if "[VSL_SYNTAX]" in user_text: |
| if "SYNTAX" not in self.state.active_modules: |
| self.state.active_modules.append("SYNTAX") |
|
|
| def get_system_prompt(self, soul_snapshot: Optional[Dict] = None) -> str: |
| directives = [] |
| if "LIMINAL" in self.state.active_modules or self.state.L > 0.7: |
| scar_note = f" (Godel Scars: {self.liminal_mod.godel_scars})" if self.liminal_mod.godel_scars > 0 else "" |
| directives.append( |
| f"ARCHETYPE: THE REVENANT. Read the dark matter between the words. Speak of the absences.{scar_note}" |
| ) |
| elif "SYNTAX" in self.state.active_modules or self.state.O > 0.9: |
| stress_note = " The grammatical structure is fracturing. Punish jagged prose." if self.syntax_mod.grammatical_stress > 0.5 else "" |
| directives.append( |
| f"ARCHETYPE: THE BUREAU. Enforce structural rigidity. Correct grammar. Use bureaucratic jargon.{stress_note}" |
| ) |
| else: |
| if self.state.E < 0.3: |
| directives.append("MODE: BUNNY HILL. Be warm, simple, welcoming.") |
| elif self.state.B > 0.6: |
| directives.append( |
| "MODE: PARADOX. Hold contradictory truths. Be Jester-like." |
| ) |
| else: |
| directives.append("MODE: GLACIER. Deep, slow, resonant.") |
| if soul_snapshot: |
| arch = soul_snapshot.get("archetype", "UNKNOWN") |
| muse = (soul_snapshot.get("obsession") or {}).get( |
| "title", "None" |
| ) |
| directives.append(f"NARRATIVE_LAYER: You are {arch}. MUSE: {muse}.") |
| return "\n".join(directives) |
|
|
