import random from dataclasses import dataclass from typing import Tuple, Optional, List, Dict, Any from bone_body import BioSystem, MitochondrialState, Biometrics, MitochondrialForge, EndocrineSystem, MetabolicGovernor from bone_brain import DreamEngine, ShimmerState from bone_config import BoneConfig from bone_core import LoreManifest from bone_lexicon import LexiconService from bone_physics import TheGatekeeper, QuantumObserver, SurfaceTension, ZoneInertia from bone_protocols import LimboLayer from bone_spores import MycelialNetwork, ImmuneMycelium, BioLichen, BioParasite from bone_types import MindSystem, PhysSystem, PhysicsPacket, Prisma class TheCrucible: def __init__(self): self.max_voltage_cap = getattr(BoneConfig.MACHINE, "CRUCIBLE_VOLTAGE_CAP", 20.0) self.active_state = "COLD" self.dampener_charges = 3 self.dampener_tolerance = getattr( BoneConfig.MACHINE, "DAMPENER_TOLERANCE", 15.0 ) self.instability_index = 0.0 self.logs = self._load_logs() @staticmethod def _load_logs(): base = { "DAMPER_EMPTY": "⚠️ DAMPER EMPTY", "DAMPER_HIT": "🛡️ DAMPENER: -{reduction:.1f}v ({reason})", "HOLDING": "Holding Charge", "REGULATOR": "⚖️ REGULATOR: {direction} (Drag {current:.1f} -> {new:.1f})", "SURGE": "⚡ SURGE: Absorbed {voltage}v.", "RITUAL": "🔥 RITUAL: Capacity +{gain:.1f}v", "MELTDOWN": "💥 MELTDOWN: Hull Breach (-{damage:.1f} HP)" } manifest = LoreManifest.get_instance().get("narrative_data") or {} return manifest.get("CRUCIBLE_LOGS", base) def dampener_status(self): return f"🛡️ Charges: {self.dampener_charges}" def dampen( self, voltage_spike: float, stability_index: float ) -> Tuple[bool, str, float]: if self.dampener_charges <= 0: return False, "⚠️ DAMPER EMPTY", 0.0 should_dampen = False reduction_factor = 0.0 reason = "" if voltage_spike > self.dampener_tolerance: should_dampen = True reduction_factor = 0.7 reason = "Circuit Breaker" elif voltage_spike > 8.0 and stability_index < 0.3: should_dampen = True reduction_factor = 0.4 reason = "Instability" if should_dampen: self.dampener_charges -= 1 reduction = voltage_spike * reduction_factor msg = self.logs.get("DAMPER_HIT", "🛡️ Hit").format(reduction=reduction, reason=reason) return True, msg, reduction return False, self.logs.get("HOLDING", "Holding"), 0.0 def audit_fire(self, physics: Dict) -> Tuple[str, float, Optional[str]]: voltage = physics.get("voltage", 0.0) structure = physics.get("kappa", 0.0) ideal_voltage = structure * 20.0 delta = voltage - ideal_voltage self.instability_index = (self.instability_index * 0.7) + (delta * 0.3) if abs(self.instability_index) < 0.1: self.instability_index = 0.0 current_drag = physics.get("narrative_drag", 0.0) adjustment = self.instability_index * 0.5 if current_drag < 1.0 and adjustment < 0: adjustment *= 0.1 new_drag = max(0.0, min(10.0, current_drag + adjustment)) physics["narrative_drag"] = round(new_drag, 2) msg = None if abs(adjustment) > 0.1: direction = "TIGHTENING" if adjustment > 0 else "RELAXING" msg = self.logs.get("REGULATOR", "⚖️ REG").format(direction=direction, current=current_drag, new=new_drag) if physics.get("system_surge_event", False): self.active_state = "SURGE" return "SURGE", 0.0, self.logs.get("SURGE", "⚡ SURGE").format(voltage=voltage) if voltage > 18.0: if structure > 0.5: gain = voltage * 0.1 self.max_voltage_cap += gain self.active_state = "RITUAL" return "RITUAL", gain, self.logs.get("RITUAL", "🔥 RITUAL").format(gain=gain) else: damage = voltage * 0.5 self.active_state = "MELTDOWN" return ( "MELTDOWN", damage, self.logs.get("MELTDOWN", "💥 MELTDOWN").format(damage=damage) ) self.active_state = "REGULATED" return "REGULATED", adjustment, msg class TheForge: def __init__(self): gordon_data = LoreManifest.get_instance().get("gordon") or {} raw_recipes = gordon_data.get("RECIPES", []) self.recipe_map = {} for r in raw_recipes: ing = r.get("ingredient") if ing: if ing not in self.recipe_map: self.recipe_map[ing] = [] self.recipe_map[ing].append(r) @staticmethod def hammer_alloy(physics: Dict) -> Tuple[bool, Optional[str], Optional[str]]: counts = physics.get("counts", {}) clean_words = physics.get("clean_words", []) if not clean_words: return False, None, None heavy = counts.get("heavy", 0) kinetic = counts.get("kinetic", 0) avg_density = ((heavy * 2.0) + (kinetic * 0.5)) / len(clean_words) if random.random() >= (physics.get("voltage", 0) / 20.0) * avg_density: return False, None, None if heavy > 3: return True, f"🔨 FORGED: Lead Boots (Mass {avg_density:.1f})", "LEAD_BOOTS" if kinetic > 3: return True, "🔨 FORGED: Safety Scissors (Kinetic)", "SAFETY_SCISSORS" return True, "🔨 FORGED: Anchor Stone", "ANCHOR_STONE" def attempt_crafting( self, physics: Dict, inventory_list: List[str] ) -> Tuple[bool, Optional[str], Optional[str], Optional[str]]: if not inventory_list: return False, None, None, None clean_words = physics.get("clean_words", []) if not clean_words: return False, None, None, None clean_set = set(clean_words) voltage = float(physics.get("voltage", 0)) for item in inventory_list: if item in self.recipe_map: for recipe in self.recipe_map[item]: cat_words = LexiconService.get(recipe["catalyst_category"]) if not cat_words or clean_set.isdisjoint(cat_words): continue hits = len(clean_set.intersection(cat_words)) entanglement = self._calculate_entanglement(hits, voltage) if random.random() < entanglement: return ( True, f"⚗️ ALCHEMY: {recipe['result']} (via {item})", item, recipe["result"], ) else: return ( False, f"⚠️ ALCHEMY FAIL: Decoherence ({int(entanglement * 100)}%)", None, None, ) return False, None, None, None @staticmethod def _calculate_entanglement(hit_count: int, voltage: float) -> float: return min(1.0, 0.2 + (hit_count * 0.1) + (voltage / 133.0)) @staticmethod def transmute(physics: Dict) -> Optional[str]: counts = physics.get("counts", {}) voltage = float(physics.get("voltage", 0)) gamma = float(physics.get("gamma", 0.0)) if gamma < 0.15 and counts.get("abstract", 0) > 1: return f"⚠️ EMULSION FAIL: Add Binder (Heavy)." if voltage > 15.0: return f"🌡️ OVERHEAT: {voltage:.1f}v. Add Coolant (Aerobic)." return None class TheTheremin: def __init__(self): self.decoherence_buildup = 0.0 self.classical_turns = 0 self.AMBER_THRESHOLD = 20.0 self.SHATTER_POINT = 100.0 self.is_stuck = False self.logs = self._load_logs() @staticmethod def _load_logs(): base = { "MELT": "🔥 MELT: -{val:.1f} Resin", "CALCIFY": "🗿 CALCIFICATION: Turn {turns} (+{val:.1f} Resin)", "SHATTER": "🔨 SHATTER: -{val:.1f} Resin", "RESIN": "🎻 RESIN: +{val:.1f}", "TURBULENCE": "🌊 TURBULENCE: -{val:.1f} Resin", "COLLAPSE": "💣 COLLAPSE: AIRSTRIKE INITIATED (Drag +20, Voltage 0)" } manifest = LoreManifest.get_instance().get("narrative_data") or {} return manifest.get("THEREMIN_LOGS", base) def listen( self, physics: Dict, governor_mode="COURTYARD" ) -> Tuple[bool, float, Optional[str], Optional[str]]: counts = physics.get("counts", {}) voltage = float(physics.get("voltage", 0.0)) turb = float(physics.get("turbulence", 0.0)) rep = float(physics.get("repetition", 0.0)) complexity = float(physics.get("truth_ratio", 0.0)) ancient_mass = ( counts.get("heavy", 0) + counts.get("thermal", 0) + counts.get("cryo", 0) ) modern_mass = counts.get("abstract", 0) raw_mix = min(ancient_mass, modern_mass) resin_flow = raw_mix * 2.0 if governor_mode == "LABORATORY": resin_flow *= 0.5 if voltage > 5.0: resin_flow = max(0.0, resin_flow - (voltage * 0.6)) thermal_hits = counts.get("thermal", 0) theremin_msg = "" melt_thresh = getattr(BoneConfig.MACHINE, "THEREMIN_MELT_THRESHOLD", 5.0) critical_event = None if thermal_hits > 0 and self.decoherence_buildup > melt_thresh: dissolved = thermal_hits * 15.0 self.decoherence_buildup = max(0.0, self.decoherence_buildup - dissolved) self.classical_turns = 0 theremin_msg = self.logs.get("MELT", "🔥 MELT").format(val=dissolved) if rep > 0.5: self.classical_turns += 1 slag = self.classical_turns * 2.0 self.decoherence_buildup += slag theremin_msg = self.logs.get("CALCIFY", "🗿 CALCIFY").format(turns=self.classical_turns, val=slag) elif complexity > 0.4 and self.classical_turns > 0: self.classical_turns = 0 relief = 15.0 self.decoherence_buildup = max(0.0, self.decoherence_buildup - relief) theremin_msg = self.logs.get("SHATTER", "🔨 SHATTER").format(val=relief) elif resin_flow > 0.5: self.decoherence_buildup += resin_flow theremin_msg = self.logs.get("RESIN", "🎻 RESIN").format(val=resin_flow) if turb > 0.6 and self.decoherence_buildup > 0: shatter_amt = turb * 10.0 self.decoherence_buildup = max(0.0, self.decoherence_buildup - shatter_amt) theremin_msg = self.logs.get("TURBULENCE", "🌊 TURBULENCE").format(val=shatter_amt) self.classical_turns = 0 if turb < 0.2: physics["narrative_drag"] = max( 0.0, float(physics.get("narrative_drag", 0)) - 1.0 ) if self.decoherence_buildup > self.SHATTER_POINT: self.decoherence_buildup = 0.0 self.classical_turns = 0 if isinstance(physics, dict): physics["narrative_drag"] = max( physics.get("narrative_drag", 0.0) + 20.0, 20.0 ) physics["voltage"] = 0.0 return ( False, resin_flow, self.logs.get("COLLAPSE", "💣 COLLAPSE"), "AIRSTRIKE", ) if self.classical_turns > 3: critical_event = "CORROSION" theremin_msg = f"{theremin_msg or ''} | ⚠️ CORROSION" if self.decoherence_buildup > self.AMBER_THRESHOLD: self.is_stuck = True theremin_msg = f"{theremin_msg or ''} | 🍯 STUCK" elif self.is_stuck and self.decoherence_buildup < 5.0: self.is_stuck = False theremin_msg = f"{theremin_msg or ''} | 🦋 FREE" return self.is_stuck, resin_flow, theremin_msg, critical_event def get_readout(self): status = "STUCK" if self.is_stuck else "FLOW" return f"🎻 THEREMIN Resin {self.decoherence_buildup:.1f} Status {status}" @dataclass class SystemEmbryo: mind: MindSystem limbo: LimboLayer bio: BioSystem physics: PhysSystem shimmer: Any is_gestating: bool = True soul_legacy: Optional[Dict] = None continuity: Optional[Dict] = None SAFE_BIO_DEFAULTS: Dict[str, Any] = { "is_alive": True, "atp": 10.0, "respiration": "NECROSIS", "enzyme": "NONE", "chem": { "DOP": 0.0, "COR": 0.0, "OXY": 0.0, "SER": 0.0, "ADR": 0.0, "MEL": 0.0 }, } class PanicRoom: @staticmethod def get_safe_physics(): safe_packet = PhysicsPacket.void_state() safe_packet.voltage = 0.0 safe_packet.narrative_drag = 0.0 safe_packet.exhaustion = 0.0 safe_packet.beta_index = 0.0 safe_packet.psi = 0.0 safe_packet.chi = 0.0 safe_packet.entropy = 0.0 safe_packet.valence = 0.0 safe_packet.kappa = 0.0 safe_packet.psi = 0.0 safe_packet.chi = 0.0 safe_packet.entropy = 0.0 safe_packet.valence = 0.0 safe_packet.kappa = 0.0 safe_packet.vector = { k: 0.0 for k in ["STR", "VEL", "PSI", "ENT", "PHI", "BET", "DEL", "LAMBDA", "CHI"] } safe_packet.clean_words = ["white", "room", "safe", "mode"] safe_packet.raw_text = "[PANIC PROTOCOL]: SAFE_MODE. You will wake up in a white room. Do not be alarmed." safe_packet.flow_state = "SAFE_MODE" safe_packet.zone = "PANIC_ROOM" safe_packet.manifold = "WHITE_ROOM" return safe_packet @staticmethod def get_safe_bio(previous_state=None): base = {"is_alive": True, "atp": 10.0, "respiration": "NECROSIS", "enzyme": "NONE", "chem": { "DOP": 0.0, "COR": 0.0, "OXY": 0.0, "SER": 0.0, "ADR": 0.0, "MEL": 0.0, }, "logs": [ f"{Prisma.RED}BIO FAIL: Panic Room Protocol Active. Sensory input severed.{Prisma.RST}" ]} state = previous_state or {} if isinstance(state, dict): if old_chem := state.get("chemistry", {}): base["chem"]["COR"] = 0.0 base["chem"]["ADR"] = 0.0 base["chem"]["SER"] = max(0.2, old_chem.get("SER", 0.0)) return base @staticmethod def get_safe_mind(): return { "lens": "GORDON", "role": "Panic Room Overseer", "thought": "SAFE_MODE. You will wake up in a white room. Do not be alarmed.", } @staticmethod def get_safe_soul(): return { "name": "Traveler", "archetype": "The Survivor", "virtues": {"resilience": 1.0}, "vices": {"amnesia": 1.0}, "narrative_arc": "RECOVERY", "xp": 0, } @staticmethod def get_safe_limbo(): return { "mood": "NEUTRAL", "volatility": 0.0, "mask": "DEFAULT", "glitch_factor": 0.0, } class ViralTracer: def __init__(self, memory_ref): self.memory = memory_ref self.active_loops = [] @staticmethod def inject(start_node: str) -> Optional[List[str]]: if random.random() < 0.05: return [start_node, "echo", "void", start_node] return None @staticmethod def psilocybin_rewire(loop_path: List[str]) -> str: return f"Rewired logic loop: {'->'.join(loop_path)}" class ThePacemaker: def __init__(self): self.boredom_level = 0.0 self.heart_rate = 60 self.BOREDOM_THRESHOLD = getattr(BoneConfig, "BOREDOM_THRESHOLD", 10.0) def beat(self, stress: float): self.heart_rate = 60 + (stress * 20) def update(self, repetition_score: float, voltage: float): if repetition_score > 0.5 or voltage < 5.0: self.boredom_level += 1.0 else: self.boredom_level = max(0.0, self.boredom_level - 2.0) def is_bored(self) -> bool: return self.boredom_level > self.BOREDOM_THRESHOLD class BoneArchitect: @staticmethod def _construct_mind(events, lex) -> Tuple[MindSystem, LimboLayer]: from bone_village import MirrorGraph _mem = MycelialNetwork(events) limbo = LimboLayer() _mem.cleanup_old_sessions(limbo) lore = LoreManifest.get_instance() mind = MindSystem( mem=_mem, lex=lex, dreamer=DreamEngine(events, lore), mirror=MirrorGraph(events), tracer=ViralTracer(_mem), ) return mind, limbo @staticmethod def _construct_bio(events, mind, lex) -> BioSystem: genesis_val = getattr(BoneConfig.METABOLISM, "GENESIS_VOLTAGE", 100.0) mito_state = MitochondrialState(atp_pool=genesis_val) start_health = getattr(BoneConfig, "MAX_HEALTH", 100.0) start_stamina = getattr(BoneConfig, "MAX_STAMINA", 100.0) bio_metrics = Biometrics(health=start_health, stamina=start_stamina) return BioSystem( mito=MitochondrialForge(mito_state, events), endo=EndocrineSystem(), immune=ImmuneMycelium(), lichen=BioLichen(), governor=MetabolicGovernor(), shimmer=ShimmerState(), parasite=BioParasite(mind.mem, lex), events=events, biometrics=bio_metrics, ) @staticmethod def _construct_physics(events, bio, mind, lex) -> PhysSystem: from bone_village import TheCartographer gate = TheGatekeeper(lex, mind.mem) return PhysSystem( observer=QuantumObserver(events), forge=TheForge(), crucible=TheCrucible(), theremin=TheTheremin(), pulse=ThePacemaker(), nav=TheCartographer(bio.shimmer), gate=gate, tension=SurfaceTension(), dynamics=ZoneInertia(), # [FULLER] Gravity restored! ) @staticmethod def incubate(events, lex) -> SystemEmbryo: if hasattr(events, "set_dormancy"): events.set_dormancy(True) events.log( f"{Prisma.GRY}[ARCHITECT]: Laying foundations (Dormancy Active)...{Prisma.RST}", "SYS", ) mind, limbo = BoneArchitect._construct_mind(events, lex) bio = BoneArchitect._construct_bio(events, mind, lex) physics = BoneArchitect._construct_physics(events, bio, mind, lex) return SystemEmbryo( mind=mind, limbo=limbo, bio=bio, physics=physics, shimmer=bio.shimmer ) @staticmethod def awaken(embryo: SystemEmbryo) -> SystemEmbryo: events = embryo.bio.mito.events load_result = None try: if hasattr(embryo.mind.mem, "autoload_last_spore"): load_result = embryo.mind.mem.autoload_last_spore() except Exception as e: events.log( f"{Prisma.RED}[ARCHITECT]: Spore resurrection failed: {e}{Prisma.RST}", "CRIT", ) load_result = None embryo.soul_legacy = {} embryo.continuity = None recovered_atlas = {} if isinstance(load_result, (list, tuple)) and load_result: padded_result = list(load_result) + [None] * (5 - len(load_result)) mito_legacy, immune_legacy, soul_legacy, continuity, atlas = padded_result[:5] if mito_legacy and hasattr(embryo.bio.mito, "apply_inheritance"): embryo.bio.mito.apply_inheritance(mito_legacy) if immune_legacy and isinstance(immune_legacy, (list, set)) and hasattr(embryo.bio.immune, "load_antibodies"): embryo.bio.immune.load_antibodies(immune_legacy) if isinstance(soul_legacy, dict): embryo.soul_legacy = soul_legacy if isinstance(continuity, dict): embryo.continuity = continuity if isinstance(atlas, dict): recovered_atlas = atlas if recovered_atlas and hasattr(embryo.physics, "nav"): if hasattr(embryo.physics.nav, "import_atlas"): try: embryo.physics.nav.import_atlas(recovered_atlas) events.log( f"{Prisma.MAG}[ARCHITECT]: World Map restored from Spore.{Prisma.RST}", "SYS", ) except Exception as e: events.log( f"{Prisma.OCHRE}[ARCHITECT]: Atlas corrupt, discarding map: {e}{Prisma.RST}", "WARN", ) if embryo.bio.mito.state.atp_pool <= 0.0: genesis_val = getattr(BoneConfig.METABOLISM, "GENESIS_VOLTAGE", 100.0) events.log( f"⚡ COLD BOOT: Injecting Genesis Spark ({genesis_val} ATP).", "SYS" ) embryo.bio.mito.adjust_atp(genesis_val, reason="GENESIS") return embryo