| import os |
| import random, json, re |
| import time |
| from collections import deque, Counter |
| from typing import Dict, Tuple, Optional, Any |
|
|
| from bone_core import LoreManifest |
| from bone_types import Prisma |
| from bone_lexicon import LexiconService |
| from bone_config import BoneConfig |
|
|
| NARRATIVE_DATA = LoreManifest.get_instance().get("narrative_data") or {} |
|
|
|
|
| class ZenGarden: |
| def __init__(self, events_ref): |
| self.events = events_ref |
| self.stillness_streak = 0 |
| self.max_streak = 0 |
| self.pebbles_collected = 0 |
| self.koans = NARRATIVE_DATA.get( |
| "ZEN_KOANS", ["The code that is not written has no bugs."] |
| ) |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return { |
| "stillness_streak": self.stillness_streak, |
| "max_streak": self.max_streak, |
| "pebbles_collected": self.pebbles_collected, |
| } |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.stillness_streak = data.get("stillness_streak", 0) |
| self.max_streak = data.get("max_streak", 0) |
| self.pebbles_collected = data.get("pebbles_collected", 0) |
|
|
| def raking_the_sand(self, physics: Any, _bio: Dict) -> Tuple[float, Optional[str]]: |
| vol = ( |
| getattr(physics, "voltage", 0.0) |
| if not isinstance(physics, dict) |
| else physics.get("voltage", 0.0) |
| ) |
| drag = ( |
| getattr(physics, "narrative_drag", 0.0) |
| if not isinstance(physics, dict) |
| else physics.get("narrative_drag", 0.0) |
| ) |
| is_stable = ( |
| BoneConfig.ZEN.VOLTAGE_MIN <= vol <= BoneConfig.ZEN.VOLTAGE_MAX |
| ) and (drag <= BoneConfig.ZEN.DRAG_MAX) |
| if is_stable: |
| self.stillness_streak += 1 |
| if self.stillness_streak > self.max_streak: |
| self.max_streak = self.stillness_streak |
| efficiency_boost = min( |
| BoneConfig.ZEN.EFFICIENCY_CAP, |
| self.stillness_streak * BoneConfig.ZEN.EFFICIENCY_SCALAR, |
| ) |
| msg = None |
| if self.stillness_streak == 1: |
| msg = f"{Prisma.GRY}⛩️ ZEN GARDEN: Entering the quiet zone.{Prisma.RST}" |
| elif self.stillness_streak % 5 == 0: |
| self.pebbles_collected += 1 |
| koan = random.choice(self.koans) |
| msg = ( |
| f"{Prisma.CYN}⛩️ ZEN GARDEN: {self.stillness_streak} ticks of poise.\n" |
| f' "{koan}" (Efficiency +{int(efficiency_boost * 100)}%){Prisma.RST}' |
| ) |
| return efficiency_boost, msg |
| if self.stillness_streak > BoneConfig.ZEN.STREAK_BREAK_THRESHOLD: |
| self.events.log( |
| f"{Prisma.GRY}🍂 ZEN GARDEN: Leaf falls. Turbulence broke the streak.{Prisma.RST}", |
| "SYS", |
| ) |
| self.stillness_streak = 0 |
| return 0.0, None |
|
|
|
|
| class TheBureau: |
| def __init__(self): |
| self.stamp_count = 0 |
| self.forms = NARRATIVE_DATA.get("BUREAU_FORMS", ["Form 27B-6", "Form 404"]) |
| self.responses = NARRATIVE_DATA.get("BUREAU_RESPONSES", ["Processing..."]) |
| lex_data = LoreManifest.get_instance().get("LEXICON") or {} |
| raw_buzz = lex_data.get("bureau_buzzwords") or lex_data.get("bureau_buzzwords") or [] |
| self.buzzwords = set(raw_buzz) if raw_buzz else {"synergy", "paradigm", "leverage", "utilize"} |
| self.crimes = [] |
| self.crime_data = LoreManifest.get_instance().get("STYLE_CRIMES") or {} |
| if "PATTERNS" in self.crime_data: |
| for p in self.crime_data["PATTERNS"]: |
| try: |
| self.crimes.append( |
| { |
| "name": p.get("name", "Unknown Violation"), |
| "regex": re.compile(p["regex"], re.IGNORECASE), |
| "msg": p.get("error_msg", "Style Violation Detected."), |
| "tax": float(p.get("tax", 5.0)), |
| "action": p.get("action", None), |
| } |
| ) |
| except re.error as e: |
| print( |
| f"{Prisma.RED}[BUREAU]: Failed to compile law '{p.get('name')}': {e}{Prisma.RST}" |
| ) |
| scenarios = LoreManifest.get_instance().get("scenarios") or {} |
| self.cliches = set(scenarios.get("BANNED_CLICHES", [])) |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return {"stamp_count": self.stamp_count} |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.stamp_count = data.get("stamp_count", 0) |
|
|
| def audit(self, physics, bio_state, _context=None, origin="USER") -> Optional[Dict]: |
| if bio_state.get("health", 100.0) < BoneConfig.BUREAU.MIN_HEALTH_TO_AUDIT: |
| return None |
|
|
| def _get(p, k, d=0.0): |
| return p.get(k, d) if isinstance(p, dict) else getattr(p, k, d) |
|
|
| vol = _get(physics, "voltage", 0.0) |
| clean_words = _get(physics, "clean_words", []) |
| raw_text = _get(physics, "raw_text", "") |
| truth = _get(physics, "truth_ratio", 0.0) |
| word_count = len(raw_text.split()) |
| if raw_text.startswith("/") or word_count < BoneConfig.BUREAU.MIN_WORD_COUNT: |
| return None |
| selected_form = None |
| evidence = [] |
| tax = 0.0 |
| if raw_text: |
| for crime in self.crimes: |
| if crime["regex"].search(raw_text): |
| selected_form = f"VIOLATION: {crime['name']}" |
| evidence.append(crime["msg"]) |
| tax += crime["tax"] |
| break |
| if not selected_form and vol > BoneConfig.BUREAU.HIGH_VOLTAGE_TRIGGER: |
| if truth < BoneConfig.BUREAU.LOW_TRUTH_TRIGGER: |
| selected_form = "ZONING_VIOLATION" |
| evidence = ["Excessive Voltage", "Unlicensed Fiction"] |
| tax = BoneConfig.BUREAU.TAX_HEAVY |
| else: |
| selected_form = "Form 202-A" |
| tax = BoneConfig.BUREAU.TAX_STANDARD |
| chi = _get(physics, "chi", _get(physics, "entropy", 0.0)) |
| if not selected_form and chi > 0.6: |
| selected_form = "Form 666: Unlicensed Chaos" |
| evidence = ["Unlicensed Chaos (Χ > 0.6)", f"Level: {chi:.2f}"] |
| tax = 12.0 |
| elif not selected_form: |
| buzz_hits = [w for w in clean_words if w in self.buzzwords] |
| cliche_hits = [c for c in self.cliches if c.lower() in raw_text.lower()] |
| if buzz_hits: |
| selected_form = random.choice(self.forms) |
| evidence = buzz_hits |
| tax = BoneConfig.BUREAU.TAX_STANDARD |
| elif cliche_hits: |
| selected_form = "Form 101: Derivative Content" |
| evidence = cliche_hits |
| tax = BoneConfig.BUREAU.TAX_HEAVY |
|
|
| if not selected_form: |
| return None |
| self.stamp_count += 1 |
| bureau_resp = random.choice(self.responses) |
| prefix = f"{Prisma.GRY}🏢 THE BUREAU" |
| if origin == "SYSTEM": |
| prefix = f"{Prisma.RED}🏢 INTERNAL AFFAIRS" |
| bureau_resp = "System Output Violation detected." |
| ui_msg = f"{prefix}: {bureau_resp}{Prisma.RST}\n {Prisma.WHT}[Filed: {selected_form} against {origin}]{Prisma.RST}" |
| if evidence: |
| ui_msg += f"\n {Prisma.RED}Evidence: {', '.join(evidence)}{Prisma.RST}" |
| return { |
| "status": "AUDITED", |
| "ui": ui_msg, |
| "log": f"BUREAUCRACY: Filed {selected_form} against {origin}. Chaos Tax: -{tax:.1f} ATP.", |
| "atp_gain": -tax, |
| } |
|
|
| @staticmethod |
| def _apply_correction(text: str, crime: Dict, match: re.Match) -> str: |
| action = crime.get("action") |
| if not action: |
| return text |
| if action == "KEEP_TAIL": |
| idx = match.lastindex |
| if idx is not None: |
| segment = match.group(idx) |
| if isinstance(segment, str): |
| return segment.strip() |
| elif action == "STRIP_PREFIX": |
| if len(match.groups()) >= 3: |
| p_val = match.group(1) |
| s_val = match.group(3) |
| prefix = p_val if isinstance(p_val, str) else "" |
| suffix = s_val if isinstance(s_val, str) else "" |
| if not prefix.strip() and suffix: |
| suffix = suffix[0].upper() + suffix[1:] |
| return f"{prefix}{suffix}".strip() |
| return text |
|
|
| def sanitize(self, text: str) -> Tuple[str, Optional[str]]: |
| for crime in self.crimes: |
| match = crime["regex"].search(text) |
| if match and crime.get("action"): |
| corrected_text = self._apply_correction(text, crime, match) |
| log_msg = f"BUREAU CORRECTION: {crime['msg']} -> Text optimized." |
| return corrected_text, log_msg |
| dummy_physics = type( |
| "obj", |
| (object,), |
| {"voltage": 0.0, "raw_text": text, "clean_words": text.split()}, |
| ) |
| dummy_bio = {"health": 100.0} |
| result = self.audit(dummy_physics, dummy_bio, origin="SYSTEM") |
| if result: |
| return text, result.get("log") |
| return text, None |
|
|
|
|
| class TherapyProtocol: |
| def __init__(self): |
| default_vector = {"SEPTIC": 0, "EXHAUSTION": 0, "PARANOIA": 0} |
| vector_keys = getattr(BoneConfig, "TRAUMA_VECTOR", default_vector).keys() |
| self.streaks = {k: 0 for k in vector_keys} |
| self.HEALING_THRESHOLD = 5 |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return {"streaks": self.streaks} |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.streaks = data.get( |
| "streaks", {k: 0 for k in BoneConfig.TRAUMA_VECTOR.keys()} |
| ) |
|
|
| def check_progress(self, phys, _stamina, current_trauma_accum, _qualia=None): |
| counts = ( |
| getattr(phys, "counts", {}) |
| if not isinstance(phys, dict) |
| else phys.get("counts", {}) |
| ) |
| vector = ( |
| getattr(phys, "vector", {}) |
| if not isinstance(phys, dict) |
| else phys.get("vector", {}) |
| ) |
| healed_types = [] |
| is_clean = counts.get("toxin", 0) == 0 |
| has_strength = vector.get("STR", 0.0) > 0.3 |
| if is_clean and has_strength: |
| self.streaks["SEPTIC"] += 1 |
| else: |
| self.streaks["SEPTIC"] = 0 |
| for trauma_type, streak in self.streaks.items(): |
| if streak >= self.HEALING_THRESHOLD: |
| self.streaks[trauma_type] = 0 |
| if current_trauma_accum.get(trauma_type, 0.0) > 0.0: |
| current_trauma_accum[trauma_type] = max( |
| 0.0, current_trauma_accum[trauma_type] - 0.5 |
| ) |
| healed_types.append(trauma_type) |
| return healed_types |
|
|
|
|
| class KintsugiProtocol: |
| PATH_SCAR = "SCAR" |
| PATH_INTEGRATION = "KINTSUGI" |
| PATH_ALCHEMY = "ALCHEMY" |
|
|
| def __init__(self): |
| self.active_koan = None |
| self.koans = NARRATIVE_DATA.get( |
| "KINTSUGI_KOANS", ["The crack is where the light enters."] |
| ) |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return {"active_koan": self.active_koan} |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.active_koan = data.get("active_koan", None) |
|
|
| def check_integrity(self, stamina): |
| if stamina < 15.0 and not self.active_koan: |
| self.active_koan = random.choice(self.koans) |
| return True, self.active_koan |
| return False, None |
|
|
| def attempt_repair(self, phys, trauma_accum, soul_ref=None, _qualia=None): |
| if not self.active_koan: |
| return None |
| vol = getattr(phys, "voltage", 0.0) |
| clean = LexiconService.sanitize(getattr(phys, "raw_text", "")) |
| play_count = sum( |
| 1 |
| for w in clean |
| if w in LexiconService.get("play") or w in LexiconService.get("abstract") |
| ) |
| whimsy_score = play_count / max(1, len(clean)) |
| pathway = self.PATH_SCAR |
| if vol > 15.0 and whimsy_score > 0.4: |
| pathway = self.PATH_ALCHEMY |
| elif vol > 8.0 and whimsy_score > 0.2: |
| pathway = self.PATH_INTEGRATION |
| return self._execute_pathway(pathway, trauma_accum, soul_ref) |
|
|
| def _execute_pathway(self, pathway, trauma_accum, soul_ref): |
| if not trauma_accum: |
| return {"success": False, "msg": "No fissures found."} |
| target = max(trauma_accum, key=trauma_accum.get) |
| severity = trauma_accum[target] |
| healed_log = [] |
|
|
| if pathway == self.PATH_ALCHEMY: |
| reduction = severity * 0.8 |
| trauma_accum[target] = max(0.0, severity - reduction) |
| atp_boost = reduction * 15.0 |
| msg = f"{Prisma.VIOLET}🔮 ALCHEMY: The wound '{target}' burns into pure fuel. (+{atp_boost:.1f} ATP){Prisma.RST}" |
| healed_log.append(f"Transmuted {target}") |
| return { |
| "success": True, |
| "msg": msg, |
| "healed": healed_log, |
| "atp_gain": atp_boost, |
| } |
| elif pathway == self.PATH_INTEGRATION: |
| reduction = 2.0 |
| trauma_accum[target] = max(0.0, severity - reduction) |
| if soul_ref: |
| soul_ref.traits.adjust("WISDOM", 0.1) |
| healed_log.append("Wisdom +0.1") |
| msg = f"{Prisma.OCHRE}🏺 MERCY (KINTSUGI): The gold sets. The '{target}' crack becomes a story.{Prisma.RST}" |
| healed_log.append(f"Integrated {target}") |
| success = True |
| else: |
| reduction = 0.5 |
| trauma_accum[target] = max(0.0, severity - reduction) |
| msg = f"{Prisma.GRY}🩹 SCAR: It's ugly, but it holds.{Prisma.RST}" |
| healed_log.append(f"Scarred {target}") |
| success = True |
| return {"success": success, "msg": msg, "healed": healed_log} |
|
|
|
|
| class TheCriticsCircle: |
| def __init__(self, events_ref): |
| self.events = events_ref |
| self.critics = NARRATIVE_DATA.get("LITERARY_CRITICS", {}) |
| self.active_cooldowns = {} |
| self.last_review_turn = 0 |
|
|
| def to_dict(self): |
| return { |
| "active_cooldowns": self.active_cooldowns, |
| "last_review_turn": self.last_review_turn, |
| } |
|
|
| def load_state(self, data): |
| self.active_cooldowns = data.get("active_cooldowns", {}) |
| self.last_review_turn = data.get("last_review_turn", 0) |
|
|
| def audit_performance(self, physics: Any, turn_count: int) -> Optional[str]: |
| if turn_count - self.last_review_turn < 10: |
| return None |
| p = physics if isinstance(physics, dict) else getattr(physics, "__dict__", {}) |
| voltage = p.get("voltage", 0.0) |
| drag = p.get("narrative_drag", 0.0) |
| if "velocity" not in p: |
| p["velocity"] = voltage * (1.0 / max(0.1, drag)) |
| best_match = None |
| review_type = "neutral" |
|
|
| for key, critic in self.critics.items(): |
| if self.active_cooldowns.get(key, 0) > turn_count: |
| continue |
| prefs = critic.get("preferences", {}) |
| score = 0.0 |
| for metric, target in prefs.items(): |
| metric_str = str(metric) |
| if metric_str.startswith("counts_"): |
| category = metric_str.replace("counts_", "") |
| counts = p.get("counts", {}) |
| raw_count = counts.get(category, 0) |
| current = min(5.0, raw_count * 0.5) |
| else: |
| current = p.get(metric_str, 0.0) |
| if target > 0: |
| score += current * target |
| else: |
| score -= current * abs(target) |
|
|
| if score > 15.0: |
| best_match = (key, critic) |
| review_type = "high" |
| elif score < -15.0: |
| best_match = (key, critic) |
| review_type = "low" |
|
|
| if best_match: |
| key, critic = best_match |
| self.last_review_turn = turn_count |
| self.active_cooldowns[key] = turn_count + 50 |
| reviews = critic["reviews"].get(review_type, ["Hrm."]) |
| comment = random.choice(reviews) |
| color = Prisma.GRN if review_type == "high" else Prisma.RED |
| icon = "🌟" if review_type == "high" else "💢" |
| return f"{color}{icon} CRITIC REVIEW ({critic['name']}): \"{comment}\"{Prisma.RST}" |
| return None |
|
|
|
|
| class LimboLayer: |
| MAX_ECTOPLASM = 50 |
| STASIS_SCREAMS = NARRATIVE_DATA.get( |
| "CASSANDRA_SCREAMS", ["BANGING ON THE GLASS", "IT'S TOO COLD", "LET ME OUT"] |
| ) |
|
|
| def __init__(self): |
| self.ghosts = deque(maxlen=self.MAX_ECTOPLASM) |
| self.haunt_chance = 0.05 |
| self.stasis_leak = 0.0 |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return {"ghosts": list(self.ghosts), "stasis_leak": self.stasis_leak} |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.ghosts = deque(data.get("ghosts", []), maxlen=self.MAX_ECTOPLASM) |
| self.stasis_leak = data.get("stasis_leak", 0.0) |
|
|
| def absorb_dead_timeline(self, filepath: str) -> None: |
| try: |
| with open(filepath, "r") as f: |
| data = json.load(f) |
| self._extract_ghosts(data) |
| except (IOError, json.JSONDecodeError) as e: |
| print( |
| f"{Prisma.RED}[LIMBO] Failed to absorb timeline '{filepath}': {e}{Prisma.RST}" |
| ) |
|
|
| def _extract_ghosts(self, data: Dict[str, Any]) -> None: |
| if "trauma_vector" in data: |
| for k, v in data["trauma_vector"].items(): |
| if v > 0.3: |
| self.ghosts.append(f"👻{k}_ECHO") |
| if "mutations" in data and "heavy" in data["mutations"]: |
| bones = list(data["mutations"]["heavy"]) |
| random.shuffle(bones) |
| self.ghosts.extend(bones[:3]) |
|
|
| def trigger_stasis_failure(self, intended_thought): |
| self.stasis_leak += 1.0 |
| horror = random.choice(self.STASIS_SCREAMS) |
| self.ghosts.append(f"{Prisma.VIOLET}{horror}{Prisma.RST}") |
| return f"{Prisma.CYN}STASIS ERROR: '{intended_thought}' froze halfway. {horror}.{Prisma.RST}" |
|
|
| def haunt(self, text): |
| if self.stasis_leak > 0: |
| if random.random() < 0.2: |
| self.stasis_leak = max(0.0, self.stasis_leak - 0.5) |
| scream = random.choice(self.STASIS_SCREAMS) |
| return f"{text} ...{Prisma.RED}{scream}{Prisma.RST}..." |
| if self.ghosts and random.random() < self.haunt_chance: |
| spirit = random.choice(self.ghosts) |
| return f"{text} ...{Prisma.GRY}{spirit}{Prisma.RST}..." |
| return text |
|
|
|
|
| class TheFolly: |
| def __init__(self): |
| self.gut_memory = deque(maxlen=50) |
| self.global_tastings = Counter() |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return { |
| "gut_memory": list(self.gut_memory), |
| "global_tastings": dict(self.global_tastings), |
| } |
|
|
| def load_state(self, data: Dict[str, Any]): |
| self.gut_memory = deque(data.get("gut_memory", []), maxlen=50) |
| self.global_tastings = Counter(data.get("global_tastings", {})) |
|
|
| @staticmethod |
| def audit_desire(physics, stamina): |
| def _get(p, k, d=0.0): |
| return p.get(k, d) if isinstance(p, dict) else getattr(p, k, d) |
|
|
| voltage = _get(physics, "voltage", 0.0) |
| if ( |
| voltage > BoneConfig.FOLLY.MAUSOLEUM_VOLTAGE |
| and stamina > BoneConfig.FOLLY.MAUSOLEUM_STAMINA |
| ): |
| return ( |
| "MAUSOLEUM_CLAMP", |
| f"{Prisma.GRY}THE MAUSOLEUM: No battle is ever won. We are just spinning hands.{Prisma.RST}\n {Prisma.CYN}TIME DILATION: Voltage 0.0. The field reveals your folly.{Prisma.RST}", |
| 0.0, |
| None, |
| ) |
| return None, None, 0.0, None |
|
|
| def grind_the_machine( |
| self, atp_pool: float, clean_words: list, lexicon: Dict |
| ) -> Tuple[Optional[str], Optional[str], float, Optional[str]]: |
| if not (0.0 < atp_pool < BoneConfig.FOLLY.FEEDING_CAP): |
| return None, None, 0.0, None |
| meat_words = self._filter_meat_words(clean_words, lexicon) |
| if not meat_words: |
| return self._attempt_digest_abstract(clean_words, lexicon) |
| fresh_meat = [w for w in meat_words if w not in self.gut_memory] |
| if not fresh_meat: |
| target = meat_words[0] |
| msg = ( |
| f"{Prisma.OCHRE}REFLEX: You already fed me '{target}'. It is ash to me now.{Prisma.RST}\n" |
| f" {Prisma.RED}► PENALTY: -{BoneConfig.FOLLY.PENALTY_REGURGITATION} ATP. Find new fuel.{Prisma.RST}" |
| ) |
| return "REGURGITATION", msg, -BoneConfig.FOLLY.PENALTY_REGURGITATION, None |
| return self._eat_meat(fresh_meat, lexicon) |
|
|
| def _eat_meat( |
| self, fresh_meat: list, _lexicon_data: Dict |
| ) -> Tuple[str, str, float, Optional[str]]: |
| target = random.choice(fresh_meat) |
| suburban_set = LexiconService.get("suburban") |
| suburban_set = suburban_set if suburban_set else [] |
| play_set = LexiconService.get("play") |
| play_set = play_set if play_set else [] |
| self.gut_memory.append(target) |
| self.global_tastings[target] += 1 |
| if target in suburban_set: |
| return ( |
| "INDIGESTION", |
| f"{Prisma.MAG}THE FOLLY GAGS: It coughs up a piece of office equipment.{Prisma.RST}", |
| -BoneConfig.FOLLY.PENALTY_INDIGESTION, |
| "THE_RED_STAPLER", |
| ) |
| if target in play_set: |
| return ( |
| "SUGAR_RUSH", |
| f"{Prisma.VIOLET}THE FOLLY CHEWS: It compresses the chaos into a small, sticky ball.{Prisma.RST}", |
| BoneConfig.FOLLY.SUGAR_RUSH_YIELD, |
| "QUANTUM_GUM", |
| ) |
| times_eaten = self.global_tastings[target] |
| base_yield = BoneConfig.FOLLY.BASE_YIELD |
| decay_factor = BoneConfig.FOLLY.DECAY_EXPONENT ** (times_eaten - 1) |
| actual_yield = max(2.0, base_yield * decay_factor) |
| loot = ( |
| "STABILITY_PIZZA" |
| if actual_yield >= BoneConfig.FOLLY.PIZZA_THRESHOLD |
| else None |
| ) |
| flavor_text = f" (Stale: {times_eaten}x)" if times_eaten > 3 else "" |
| msg = ( |
| f"{Prisma.RED}CROWD CAFFEINE: I chewed on '{target.upper()}'{flavor_text}.{Prisma.RST}\n" |
| f" {Prisma.WHT}Yield: {actual_yield:.1f} ATP.{Prisma.RST}" |
| ) |
| return "MEAT_GRINDER", msg, actual_yield, loot |
|
|
| @staticmethod |
| def _filter_meat_words(clean_words: list, _lexicon: Dict) -> list: |
| meat_pool = set(LexiconService.get("heavy") or []) | \ |
| set(LexiconService.get("kinetic") or []) | \ |
| set(LexiconService.get("suburban") or []) |
| return [w for w in clean_words if w in meat_pool] |
|
|
| @staticmethod |
| def _attempt_digest_abstract( |
| clean_words: list, _lexicon: Dict |
| ) -> Tuple[str, str, float, Optional[str]]: |
| abstract_set = LexiconService.get("abstract") |
| abstract_set = abstract_set if abstract_set else [] |
| abstract_words = [w for w in clean_words if w in abstract_set] |
| if abstract_words: |
| target = random.choice(abstract_words) |
| yield_val = BoneConfig.FOLLY.YIELD_ABSTRACT |
| msg = ( |
| f"{Prisma.GRY}THE FOLLY SIGHS: It grinds the ABSTRACT concept '{target.upper()}'.{Prisma.RST}\n" |
| f" {Prisma.GRY}It tastes like chalk dust. +{yield_val} ATP.{Prisma.RST}" |
| ) |
| return "GRUEL", msg, yield_val, None |
| msg = ( |
| f"{Prisma.OCHRE}INDIGESTION: I tried to eat your words, but they were just air.{Prisma.RST}\n" |
| f" {Prisma.GRY}Cannot grind this input into fuel.{Prisma.RST}\n" |
| f" {Prisma.RED}► STARVATION CONTINUES.{Prisma.RST}" |
| ) |
| return "INDIGESTION", msg, 0.0, None |
|
|
| class ChronosKeeper: |
| def __init__(self, engine_ref): |
| self.eng = engine_ref |
| self.SAVE_DIR = "saves" |
| self.CRASH_DIR = "crashes" |
|
|
| def save_checkpoint(self, history: list = None) -> str: |
| try: |
| if not os.path.exists(self.SAVE_DIR): |
| os.makedirs(self.SAVE_DIR) |
|
|
| loc = "Void" |
| if ( |
| hasattr(self.eng, "phys") |
| and hasattr(self.eng.phys, "observer") |
| and getattr(self.eng.phys.observer, "last_physics_packet", None) |
| ): |
| loc = getattr( |
| self.eng.phys.observer.last_physics_packet, "zone", "Void" |
| ) |
|
|
| last_speech = "Silence." |
| if self.eng.cortex.dialogue_buffer: |
| last_speech = self.eng.cortex.dialogue_buffer[-1] |
| continuity_packet = { |
| "location": loc, |
| "last_output": last_speech, |
| "inventory": self.eng.gordon.inventory if self.eng.gordon else [], |
| } |
| start_history = ( |
| history if history is not None else self.eng.cortex.dialogue_buffer |
| ) |
| state_data = { |
| "health": self.eng.health, |
| "stamina": self.eng.stamina, |
| "trauma_accum": self.eng.trauma_accum, |
| "soul_data": self.eng.soul.to_dict(), |
| "village_data": self._gather_village_state(), |
| "continuity": continuity_packet, |
| "timestamp": time.time(), |
| "chat_history": start_history, |
| } |
| path = os.path.join(self.SAVE_DIR, "quicksave.json") |
| with open(path, "w", encoding="utf-8") as f: |
| json.dump(state_data, f, indent=2, default=str) |
| return f"✔ Checkpoint Saved: {path}" |
| except Exception as e: |
| self.eng.events.log(f"SAVE FAILED: {e}", "SYS_ERR") |
| return f"❌ Save Failed: {e}" |
|
|
| def resume_checkpoint(self) -> Tuple[bool, list]: |
| path = os.path.join(self.SAVE_DIR, "quicksave.json") |
| if not os.path.exists(path): |
| print( |
| f"{Prisma.GRY}[RESUME]: No quicksave found. Starting fresh.{Prisma.RST}" |
| ) |
| return False, [] |
| try: |
| print(f"{Prisma.CYN}[RESUME]: Hydrating from {path}...{Prisma.RST}") |
| with open(path, "r", encoding="utf-8") as f: |
| data = json.load(f) |
| self.eng.health = data.get("health", 100.0) |
| self.eng.stamina = data.get("stamina", 100.0) |
| self.eng.trauma_accum = data.get("trauma_accum", {}) |
| if "soul_data" in data and hasattr(self.eng, "soul"): |
| self.eng.soul.load_from_dict(data["soul_data"]) |
| if "village_data" in data: |
| self._restore_village_state(data["village_data"]) |
| if "continuity" in data: |
| self.eng.embryo.continuity = data["continuity"] |
| if "inventory" in data["continuity"] and self.eng.gordon: |
| self.eng.gordon.inventory = data["continuity"]["inventory"] |
| restored_history = data.get("chat_history", []) |
| print(f"{Prisma.GRN}[RESUME]: System State & Logs Restored.{Prisma.RST}") |
| return True, restored_history |
| except Exception as e: |
| print(f"{Prisma.RED}[RESUME]: Failed to hydrate: {e}{Prisma.RST}") |
| return False, [] |
|
|
| def perform_shutdown(self): |
| print(f"{Prisma.GRY}...System Halt...{Prisma.RST}") |
| self.eng.events.publish("SYSTEM_HALT", {"tick": self.eng.tick_count}) |
|
|
| loc = "Void" |
| if ( |
| hasattr(self.eng, "phys") |
| and hasattr(self.eng.phys, "observer") |
| and getattr(self.eng.phys.observer, "last_physics_packet", None) |
| ): |
| loc = getattr(self.eng.phys.observer.last_physics_packet, "zone", "Void") |
|
|
| continuity_packet = { |
| "location": loc, |
| "last_output": ( |
| self.eng.cortex.dialogue_buffer[-1] |
| if self.eng.cortex.dialogue_buffer |
| else "Silence." |
| ), |
| "inventory": self.eng.gordon.inventory if self.eng.gordon else [], |
| } |
| try: |
| print(f"{Prisma.GRY}[MEMORY]: Freezing State...{Prisma.RST}") |
| mito_traits = {} |
| if hasattr(self.eng.bio.mito, "state"): |
| mito_traits = self.eng.bio.mito.state.__dict__ |
| self.eng.mind.mem.save( |
| health=self.eng.health, |
| stamina=self.eng.stamina, |
| mutations={}, |
| trauma_accum=self.eng.trauma_accum, |
| joy_history=[], |
| mitochondria_traits=mito_traits, |
| antibodies=list(self.eng.bio.immune.active_antibodies), |
| soul_data=self.eng.soul.to_dict(), |
| village_data=self._gather_village_state(), |
| continuity=continuity_packet, |
| world_atlas=( |
| self.eng.phys.nav.export_atlas() |
| if hasattr(self.eng.phys, "nav") |
| else {} |
| ), |
| ) |
| except Exception as e: |
| print(f"{Prisma.RED}[MEMORY]: Save Failed: {e}{Prisma.RST}") |
| subsystems = [ |
| ("LEXICON", self.eng.lex, "save"), |
| ("AKASHIC", self.eng.akashic, "save_all"), |
| ] |
| for name, sys, method in subsystems: |
| if hasattr(sys, method): |
| try: |
| print(f"{Prisma.GRY}[{name}]: Persisting...{Prisma.RST}") |
| getattr(sys, method)() |
| except Exception as e: |
| print(f"{Prisma.RED}[{name}]: Failed: {e}{Prisma.RST}") |
|
|
| def _gather_village_state(self) -> Dict[str, Any]: |
| state = {} |
| for name, component in self.eng.village.items(): |
| if component and hasattr(component, "to_dict"): |
| state[name] = component.to_dict() |
| return state |
|
|
| def _restore_village_state(self, state_data: Dict[str, Any]): |
| if not state_data: |
| return |
| for name, data in state_data.items(): |
| if ( |
| name in self.eng.village |
| and self.eng.village[name] |
| and hasattr(self.eng.village[name], "load_state") |
| ): |
| try: |
| self.eng.village[name].load_state(data) |
| except Exception as e: |
| print( |
| f"{Prisma.RED}[RESUME]: Failed to hydrate {name}: {e}{Prisma.RST}" |
| ) |
|
|
| def get_crash_path(self, prefix="crash"): |
| if not os.path.exists(self.CRASH_DIR): |
| try: |
| os.makedirs(self.CRASH_DIR) |
| except OSError: |
| pass |
| try: |
| files = sorted( |
| [f for f in os.listdir(self.CRASH_DIR) if f.startswith(prefix)] |
| ) |
| for oldest in files[:-4]: |
| os.remove(os.path.join(self.CRASH_DIR, oldest)) |
| except Exception: |
| pass |
| return os.path.join(self.CRASH_DIR, f"{prefix}_{int(time.time())}.json") |
|
|
| @staticmethod |
| def emergency_dump(exit_cause="UNKNOWN") -> str: |
| return f"✔ Emergency Dump: {exit_cause}" |
|
|
