aedmark commited on
Commit
7fd9f51
·
verified ·
1 Parent(s): 6aebddd

Delete bone_protocols.py

Browse files
Files changed (1) hide show
  1. bone_protocols.py +0 -763
bone_protocols.py DELETED
@@ -1,763 +0,0 @@
1
- import os
2
- import random, json, re
3
- import time
4
- from collections import deque, Counter
5
- from typing import Dict, Tuple, Optional, Any
6
-
7
- from bone_core import LoreManifest
8
- from bone_types import Prisma
9
- from bone_lexicon import LexiconService
10
- from bone_config import BoneConfig
11
-
12
- NARRATIVE_DATA = LoreManifest.get_instance().get("narrative_data") or {}
13
-
14
-
15
- class ZenGarden:
16
- def __init__(self, events_ref):
17
- self.events = events_ref
18
- self.stillness_streak = 0
19
- self.max_streak = 0
20
- self.pebbles_collected = 0
21
- self.koans = NARRATIVE_DATA.get(
22
- "ZEN_KOANS", ["The code that is not written has no bugs."]
23
- )
24
-
25
- def to_dict(self) -> Dict[str, Any]:
26
- return {
27
- "stillness_streak": self.stillness_streak,
28
- "max_streak": self.max_streak,
29
- "pebbles_collected": self.pebbles_collected,
30
- }
31
-
32
- def load_state(self, data: Dict[str, Any]):
33
- self.stillness_streak = data.get("stillness_streak", 0)
34
- self.max_streak = data.get("max_streak", 0)
35
- self.pebbles_collected = data.get("pebbles_collected", 0)
36
-
37
- def raking_the_sand(self, physics: Any, _bio: Dict) -> Tuple[float, Optional[str]]:
38
- vol = (
39
- getattr(physics, "voltage", 0.0)
40
- if not isinstance(physics, dict)
41
- else physics.get("voltage", 0.0)
42
- )
43
- drag = (
44
- getattr(physics, "narrative_drag", 0.0)
45
- if not isinstance(physics, dict)
46
- else physics.get("narrative_drag", 0.0)
47
- )
48
- is_stable = (
49
- BoneConfig.ZEN.VOLTAGE_MIN <= vol <= BoneConfig.ZEN.VOLTAGE_MAX
50
- ) and (drag <= BoneConfig.ZEN.DRAG_MAX)
51
- if is_stable:
52
- self.stillness_streak += 1
53
- if self.stillness_streak > self.max_streak:
54
- self.max_streak = self.stillness_streak
55
- efficiency_boost = min(
56
- BoneConfig.ZEN.EFFICIENCY_CAP,
57
- self.stillness_streak * BoneConfig.ZEN.EFFICIENCY_SCALAR,
58
- )
59
- msg = None
60
- if self.stillness_streak == 1:
61
- msg = f"{Prisma.GRY}⛩️ ZEN GARDEN: Entering the quiet zone.{Prisma.RST}"
62
- elif self.stillness_streak % 5 == 0:
63
- self.pebbles_collected += 1
64
- koan = random.choice(self.koans)
65
- msg = (
66
- f"{Prisma.CYN}⛩️ ZEN GARDEN: {self.stillness_streak} ticks of poise.\n"
67
- f' "{koan}" (Efficiency +{int(efficiency_boost * 100)}%){Prisma.RST}'
68
- )
69
- return efficiency_boost, msg
70
- if self.stillness_streak > BoneConfig.ZEN.STREAK_BREAK_THRESHOLD:
71
- self.events.log(
72
- f"{Prisma.GRY}🍂 ZEN GARDEN: Leaf falls. Turbulence broke the streak.{Prisma.RST}",
73
- "SYS",
74
- )
75
- self.stillness_streak = 0
76
- return 0.0, None
77
-
78
-
79
- class TheBureau:
80
- def __init__(self):
81
- self.stamp_count = 0
82
- self.forms = NARRATIVE_DATA.get("BUREAU_FORMS", ["Form 27B-6", "Form 404"])
83
- self.responses = NARRATIVE_DATA.get("BUREAU_RESPONSES", ["Processing..."])
84
- lex_data = LoreManifest.get_instance().get("LEXICON") or {}
85
- raw_buzz = lex_data.get("bureau_buzzwords") or lex_data.get("bureau_buzzwords") or []
86
- self.buzzwords = set(raw_buzz) if raw_buzz else {"synergy", "paradigm", "leverage", "utilize"}
87
- self.crimes = []
88
- self.crime_data = LoreManifest.get_instance().get("STYLE_CRIMES") or {}
89
- if "PATTERNS" in self.crime_data:
90
- for p in self.crime_data["PATTERNS"]:
91
- try:
92
- self.crimes.append(
93
- {
94
- "name": p.get("name", "Unknown Violation"),
95
- "regex": re.compile(p["regex"], re.IGNORECASE),
96
- "msg": p.get("error_msg", "Style Violation Detected."),
97
- "tax": float(p.get("tax", 5.0)),
98
- "action": p.get("action", None),
99
- }
100
- )
101
- except re.error as e:
102
- print(
103
- f"{Prisma.RED}[BUREAU]: Failed to compile law '{p.get('name')}': {e}{Prisma.RST}"
104
- )
105
- scenarios = LoreManifest.get_instance().get("scenarios") or {}
106
- self.cliches = set(scenarios.get("BANNED_CLICHES", []))
107
-
108
- def to_dict(self) -> Dict[str, Any]:
109
- return {"stamp_count": self.stamp_count}
110
-
111
- def load_state(self, data: Dict[str, Any]):
112
- self.stamp_count = data.get("stamp_count", 0)
113
-
114
- def audit(self, physics, bio_state, _context=None, origin="USER") -> Optional[Dict]:
115
- if bio_state.get("health", 100.0) < BoneConfig.BUREAU.MIN_HEALTH_TO_AUDIT:
116
- return None
117
-
118
- def _get(p, k, d=0.0):
119
- return p.get(k, d) if isinstance(p, dict) else getattr(p, k, d)
120
-
121
- vol = _get(physics, "voltage", 0.0)
122
- clean_words = _get(physics, "clean_words", [])
123
- raw_text = _get(physics, "raw_text", "")
124
- truth = _get(physics, "truth_ratio", 0.0)
125
- word_count = len(raw_text.split())
126
- if raw_text.startswith("/") or word_count < BoneConfig.BUREAU.MIN_WORD_COUNT:
127
- return None
128
- selected_form = None
129
- evidence = []
130
- tax = 0.0
131
- if raw_text:
132
- for crime in self.crimes:
133
- if crime["regex"].search(raw_text):
134
- selected_form = f"VIOLATION: {crime['name']}"
135
- evidence.append(crime["msg"])
136
- tax += crime["tax"]
137
- break
138
- if not selected_form and vol > BoneConfig.BUREAU.HIGH_VOLTAGE_TRIGGER:
139
- if truth < BoneConfig.BUREAU.LOW_TRUTH_TRIGGER:
140
- selected_form = "ZONING_VIOLATION"
141
- evidence = ["Excessive Voltage", "Unlicensed Fiction"]
142
- tax = BoneConfig.BUREAU.TAX_HEAVY
143
- else:
144
- selected_form = "Form 202-A"
145
- tax = BoneConfig.BUREAU.TAX_STANDARD
146
- chi = _get(physics, "chi", _get(physics, "entropy", 0.0))
147
- if not selected_form and chi > 0.6:
148
- selected_form = "Form 666: Unlicensed Chaos"
149
- evidence = ["Unlicensed Chaos (Χ > 0.6)", f"Level: {chi:.2f}"]
150
- tax = 12.0
151
- elif not selected_form:
152
- buzz_hits = [w for w in clean_words if w in self.buzzwords]
153
- cliche_hits = [c for c in self.cliches if c.lower() in raw_text.lower()]
154
- if buzz_hits:
155
- selected_form = random.choice(self.forms)
156
- evidence = buzz_hits
157
- tax = BoneConfig.BUREAU.TAX_STANDARD
158
- elif cliche_hits:
159
- selected_form = "Form 101: Derivative Content"
160
- evidence = cliche_hits
161
- tax = BoneConfig.BUREAU.TAX_HEAVY
162
-
163
- if not selected_form:
164
- return None
165
- self.stamp_count += 1
166
- bureau_resp = random.choice(self.responses)
167
- prefix = f"{Prisma.GRY}🏢 THE BUREAU"
168
- if origin == "SYSTEM":
169
- prefix = f"{Prisma.RED}🏢 INTERNAL AFFAIRS"
170
- bureau_resp = "System Output Violation detected."
171
- ui_msg = f"{prefix}: {bureau_resp}{Prisma.RST}\n {Prisma.WHT}[Filed: {selected_form} against {origin}]{Prisma.RST}"
172
- if evidence:
173
- ui_msg += f"\n {Prisma.RED}Evidence: {', '.join(evidence)}{Prisma.RST}"
174
- return {
175
- "status": "AUDITED",
176
- "ui": ui_msg,
177
- "log": f"BUREAUCRACY: Filed {selected_form} against {origin}. Chaos Tax: -{tax:.1f} ATP.",
178
- "atp_gain": -tax,
179
- }
180
-
181
- @staticmethod
182
- def _apply_correction(text: str, crime: Dict, match: re.Match) -> str:
183
- action = crime.get("action")
184
- if not action:
185
- return text
186
- if action == "KEEP_TAIL":
187
- idx = match.lastindex
188
- if idx is not None:
189
- segment = match.group(idx)
190
- if isinstance(segment, str):
191
- return segment.strip()
192
- elif action == "STRIP_PREFIX":
193
- if len(match.groups()) >= 3:
194
- p_val = match.group(1)
195
- s_val = match.group(3)
196
- prefix = p_val if isinstance(p_val, str) else ""
197
- suffix = s_val if isinstance(s_val, str) else ""
198
- if not prefix.strip() and suffix:
199
- suffix = suffix[0].upper() + suffix[1:]
200
- return f"{prefix}{suffix}".strip()
201
- return text
202
-
203
- def sanitize(self, text: str) -> Tuple[str, Optional[str]]:
204
- for crime in self.crimes:
205
- match = crime["regex"].search(text)
206
- if match and crime.get("action"):
207
- corrected_text = self._apply_correction(text, crime, match)
208
- log_msg = f"BUREAU CORRECTION: {crime['msg']} -> Text optimized."
209
- return corrected_text, log_msg
210
- dummy_physics = type(
211
- "obj",
212
- (object,),
213
- {"voltage": 0.0, "raw_text": text, "clean_words": text.split()},
214
- )
215
- dummy_bio = {"health": 100.0}
216
- result = self.audit(dummy_physics, dummy_bio, origin="SYSTEM")
217
- if result:
218
- return text, result.get("log")
219
- return text, None
220
-
221
-
222
- class TherapyProtocol:
223
- def __init__(self):
224
- default_vector = {"SEPTIC": 0, "EXHAUSTION": 0, "PARANOIA": 0}
225
- vector_keys = getattr(BoneConfig, "TRAUMA_VECTOR", default_vector).keys()
226
- self.streaks = {k: 0 for k in vector_keys}
227
- self.HEALING_THRESHOLD = 5
228
-
229
- def to_dict(self) -> Dict[str, Any]:
230
- return {"streaks": self.streaks}
231
-
232
- def load_state(self, data: Dict[str, Any]):
233
- self.streaks = data.get(
234
- "streaks", {k: 0 for k in BoneConfig.TRAUMA_VECTOR.keys()}
235
- )
236
-
237
- def check_progress(self, phys, _stamina, current_trauma_accum, _qualia=None):
238
- counts = (
239
- getattr(phys, "counts", {})
240
- if not isinstance(phys, dict)
241
- else phys.get("counts", {})
242
- )
243
- vector = (
244
- getattr(phys, "vector", {})
245
- if not isinstance(phys, dict)
246
- else phys.get("vector", {})
247
- )
248
- healed_types = []
249
- is_clean = counts.get("toxin", 0) == 0
250
- has_strength = vector.get("STR", 0.0) > 0.3
251
- if is_clean and has_strength:
252
- self.streaks["SEPTIC"] += 1
253
- else:
254
- self.streaks["SEPTIC"] = 0
255
- for trauma_type, streak in self.streaks.items():
256
- if streak >= self.HEALING_THRESHOLD:
257
- self.streaks[trauma_type] = 0
258
- if current_trauma_accum.get(trauma_type, 0.0) > 0.0:
259
- current_trauma_accum[trauma_type] = max(
260
- 0.0, current_trauma_accum[trauma_type] - 0.5
261
- )
262
- healed_types.append(trauma_type)
263
- return healed_types
264
-
265
-
266
- class KintsugiProtocol:
267
- PATH_SCAR = "SCAR"
268
- PATH_INTEGRATION = "KINTSUGI"
269
- PATH_ALCHEMY = "ALCHEMY"
270
-
271
- def __init__(self):
272
- self.active_koan = None
273
- self.koans = NARRATIVE_DATA.get(
274
- "KINTSUGI_KOANS", ["The crack is where the light enters."]
275
- )
276
-
277
- def to_dict(self) -> Dict[str, Any]:
278
- return {"active_koan": self.active_koan}
279
-
280
- def load_state(self, data: Dict[str, Any]):
281
- self.active_koan = data.get("active_koan", None)
282
-
283
- def check_integrity(self, stamina):
284
- if stamina < 15.0 and not self.active_koan:
285
- self.active_koan = random.choice(self.koans)
286
- return True, self.active_koan
287
- return False, None
288
-
289
- def attempt_repair(self, phys, trauma_accum, soul_ref=None, _qualia=None):
290
- if not self.active_koan:
291
- return None
292
- vol = getattr(phys, "voltage", 0.0)
293
- clean = LexiconService.sanitize(getattr(phys, "raw_text", ""))
294
- play_count = sum(
295
- 1
296
- for w in clean
297
- if w in LexiconService.get("play") or w in LexiconService.get("abstract")
298
- )
299
- whimsy_score = play_count / max(1, len(clean))
300
- pathway = self.PATH_SCAR
301
- if vol > 15.0 and whimsy_score > 0.4:
302
- pathway = self.PATH_ALCHEMY
303
- elif vol > 8.0 and whimsy_score > 0.2:
304
- pathway = self.PATH_INTEGRATION
305
- return self._execute_pathway(pathway, trauma_accum, soul_ref)
306
-
307
- def _execute_pathway(self, pathway, trauma_accum, soul_ref):
308
- if not trauma_accum:
309
- return {"success": False, "msg": "No fissures found."}
310
- target = max(trauma_accum, key=trauma_accum.get)
311
- severity = trauma_accum[target]
312
- healed_log = []
313
-
314
- if pathway == self.PATH_ALCHEMY:
315
- reduction = severity * 0.8
316
- trauma_accum[target] = max(0.0, severity - reduction)
317
- atp_boost = reduction * 15.0
318
- msg = f"{Prisma.VIOLET}🔮 ALCHEMY: The wound '{target}' burns into pure fuel. (+{atp_boost:.1f} ATP){Prisma.RST}"
319
- healed_log.append(f"Transmuted {target}")
320
- return {
321
- "success": True,
322
- "msg": msg,
323
- "healed": healed_log,
324
- "atp_gain": atp_boost,
325
- }
326
- elif pathway == self.PATH_INTEGRATION:
327
- reduction = 2.0
328
- trauma_accum[target] = max(0.0, severity - reduction)
329
- if soul_ref:
330
- soul_ref.traits.adjust("WISDOM", 0.1)
331
- healed_log.append("Wisdom +0.1")
332
- msg = f"{Prisma.OCHRE}🏺 MERCY (KINTSUGI): The gold sets. The '{target}' crack becomes a story.{Prisma.RST}"
333
- healed_log.append(f"Integrated {target}")
334
- success = True
335
- else:
336
- reduction = 0.5
337
- trauma_accum[target] = max(0.0, severity - reduction)
338
- msg = f"{Prisma.GRY}🩹 SCAR: It's ugly, but it holds.{Prisma.RST}"
339
- healed_log.append(f"Scarred {target}")
340
- success = True
341
- return {"success": success, "msg": msg, "healed": healed_log}
342
-
343
-
344
- class TheCriticsCircle:
345
- def __init__(self, events_ref):
346
- self.events = events_ref
347
- self.critics = NARRATIVE_DATA.get("LITERARY_CRITICS", {})
348
- self.active_cooldowns = {}
349
- self.last_review_turn = 0
350
-
351
- def to_dict(self):
352
- return {
353
- "active_cooldowns": self.active_cooldowns,
354
- "last_review_turn": self.last_review_turn,
355
- }
356
-
357
- def load_state(self, data):
358
- self.active_cooldowns = data.get("active_cooldowns", {})
359
- self.last_review_turn = data.get("last_review_turn", 0)
360
-
361
- def audit_performance(self, physics: Any, turn_count: int) -> Optional[str]:
362
- if turn_count - self.last_review_turn < 10:
363
- return None
364
- p = physics if isinstance(physics, dict) else getattr(physics, "__dict__", {})
365
- voltage = p.get("voltage", 0.0)
366
- drag = p.get("narrative_drag", 0.0)
367
- if "velocity" not in p:
368
- p["velocity"] = voltage * (1.0 / max(0.1, drag))
369
- best_match = None
370
- review_type = "neutral"
371
-
372
- for key, critic in self.critics.items():
373
- if self.active_cooldowns.get(key, 0) > turn_count:
374
- continue
375
- prefs = critic.get("preferences", {})
376
- score = 0.0
377
- for metric, target in prefs.items():
378
- metric_str = str(metric)
379
- if metric_str.startswith("counts_"):
380
- category = metric_str.replace("counts_", "")
381
- counts = p.get("counts", {})
382
- raw_count = counts.get(category, 0)
383
- current = min(5.0, raw_count * 0.5)
384
- else:
385
- current = p.get(metric_str, 0.0)
386
- if target > 0:
387
- score += current * target
388
- else:
389
- score -= current * abs(target)
390
-
391
- if score > 15.0:
392
- best_match = (key, critic)
393
- review_type = "high"
394
- elif score < -15.0:
395
- best_match = (key, critic)
396
- review_type = "low"
397
-
398
- if best_match:
399
- key, critic = best_match
400
- self.last_review_turn = turn_count
401
- self.active_cooldowns[key] = turn_count + 50
402
- reviews = critic["reviews"].get(review_type, ["Hrm."])
403
- comment = random.choice(reviews)
404
- color = Prisma.GRN if review_type == "high" else Prisma.RED
405
- icon = "🌟" if review_type == "high" else "💢"
406
- return f"{color}{icon} CRITIC REVIEW ({critic['name']}): \"{comment}\"{Prisma.RST}"
407
- return None
408
-
409
-
410
- class LimboLayer:
411
- MAX_ECTOPLASM = 50
412
- STASIS_SCREAMS = NARRATIVE_DATA.get(
413
- "CASSANDRA_SCREAMS", ["BANGING ON THE GLASS", "IT'S TOO COLD", "LET ME OUT"]
414
- )
415
-
416
- def __init__(self):
417
- self.ghosts = deque(maxlen=self.MAX_ECTOPLASM)
418
- self.haunt_chance = 0.05
419
- self.stasis_leak = 0.0
420
-
421
- def to_dict(self) -> Dict[str, Any]:
422
- return {"ghosts": list(self.ghosts), "stasis_leak": self.stasis_leak}
423
-
424
- def load_state(self, data: Dict[str, Any]):
425
- self.ghosts = deque(data.get("ghosts", []), maxlen=self.MAX_ECTOPLASM)
426
- self.stasis_leak = data.get("stasis_leak", 0.0)
427
-
428
- def absorb_dead_timeline(self, filepath: str) -> None:
429
- try:
430
- with open(filepath, "r") as f:
431
- data = json.load(f)
432
- self._extract_ghosts(data)
433
- except (IOError, json.JSONDecodeError) as e:
434
- print(
435
- f"{Prisma.RED}[LIMBO] Failed to absorb timeline '{filepath}': {e}{Prisma.RST}"
436
- )
437
-
438
- def _extract_ghosts(self, data: Dict[str, Any]) -> None:
439
- if "trauma_vector" in data:
440
- for k, v in data["trauma_vector"].items():
441
- if v > 0.3:
442
- self.ghosts.append(f"👻{k}_ECHO")
443
- if "mutations" in data and "heavy" in data["mutations"]:
444
- bones = list(data["mutations"]["heavy"])
445
- random.shuffle(bones)
446
- self.ghosts.extend(bones[:3])
447
-
448
- def trigger_stasis_failure(self, intended_thought):
449
- self.stasis_leak += 1.0
450
- horror = random.choice(self.STASIS_SCREAMS)
451
- self.ghosts.append(f"{Prisma.VIOLET}{horror}{Prisma.RST}")
452
- return f"{Prisma.CYN}STASIS ERROR: '{intended_thought}' froze halfway. {horror}.{Prisma.RST}"
453
-
454
- def haunt(self, text):
455
- if self.stasis_leak > 0:
456
- if random.random() < 0.2:
457
- self.stasis_leak = max(0.0, self.stasis_leak - 0.5)
458
- scream = random.choice(self.STASIS_SCREAMS)
459
- return f"{text} ...{Prisma.RED}{scream}{Prisma.RST}..."
460
- if self.ghosts and random.random() < self.haunt_chance:
461
- spirit = random.choice(self.ghosts)
462
- return f"{text} ...{Prisma.GRY}{spirit}{Prisma.RST}..."
463
- return text
464
-
465
-
466
- class TheFolly:
467
- def __init__(self):
468
- self.gut_memory = deque(maxlen=50)
469
- self.global_tastings = Counter()
470
-
471
- def to_dict(self) -> Dict[str, Any]:
472
- return {
473
- "gut_memory": list(self.gut_memory),
474
- "global_tastings": dict(self.global_tastings),
475
- }
476
-
477
- def load_state(self, data: Dict[str, Any]):
478
- self.gut_memory = deque(data.get("gut_memory", []), maxlen=50)
479
- self.global_tastings = Counter(data.get("global_tastings", {}))
480
-
481
- @staticmethod
482
- def audit_desire(physics, stamina):
483
- def _get(p, k, d=0.0):
484
- return p.get(k, d) if isinstance(p, dict) else getattr(p, k, d)
485
-
486
- voltage = _get(physics, "voltage", 0.0)
487
- if (
488
- voltage > BoneConfig.FOLLY.MAUSOLEUM_VOLTAGE
489
- and stamina > BoneConfig.FOLLY.MAUSOLEUM_STAMINA
490
- ):
491
- return (
492
- "MAUSOLEUM_CLAMP",
493
- 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}",
494
- 0.0,
495
- None,
496
- )
497
- return None, None, 0.0, None
498
-
499
- def grind_the_machine(
500
- self, atp_pool: float, clean_words: list, lexicon: Dict
501
- ) -> Tuple[Optional[str], Optional[str], float, Optional[str]]:
502
- if not (0.0 < atp_pool < BoneConfig.FOLLY.FEEDING_CAP):
503
- return None, None, 0.0, None
504
- meat_words = self._filter_meat_words(clean_words, lexicon)
505
- if not meat_words:
506
- return self._attempt_digest_abstract(clean_words, lexicon)
507
- fresh_meat = [w for w in meat_words if w not in self.gut_memory]
508
- if not fresh_meat:
509
- target = meat_words[0]
510
- msg = (
511
- f"{Prisma.OCHRE}REFLEX: You already fed me '{target}'. It is ash to me now.{Prisma.RST}\n"
512
- f" {Prisma.RED}► PENALTY: -{BoneConfig.FOLLY.PENALTY_REGURGITATION} ATP. Find new fuel.{Prisma.RST}"
513
- )
514
- return "REGURGITATION", msg, -BoneConfig.FOLLY.PENALTY_REGURGITATION, None
515
- return self._eat_meat(fresh_meat, lexicon)
516
-
517
- def _eat_meat(
518
- self, fresh_meat: list, _lexicon_data: Dict
519
- ) -> Tuple[str, str, float, Optional[str]]:
520
- target = random.choice(fresh_meat)
521
- suburban_set = LexiconService.get("suburban")
522
- suburban_set = suburban_set if suburban_set else []
523
- play_set = LexiconService.get("play")
524
- play_set = play_set if play_set else []
525
- self.gut_memory.append(target)
526
- self.global_tastings[target] += 1
527
- if target in suburban_set:
528
- return (
529
- "INDIGESTION",
530
- f"{Prisma.MAG}THE FOLLY GAGS: It coughs up a piece of office equipment.{Prisma.RST}",
531
- -BoneConfig.FOLLY.PENALTY_INDIGESTION,
532
- "THE_RED_STAPLER",
533
- )
534
- if target in play_set:
535
- return (
536
- "SUGAR_RUSH",
537
- f"{Prisma.VIOLET}THE FOLLY CHEWS: It compresses the chaos into a small, sticky ball.{Prisma.RST}",
538
- BoneConfig.FOLLY.SUGAR_RUSH_YIELD,
539
- "QUANTUM_GUM",
540
- )
541
- times_eaten = self.global_tastings[target]
542
- base_yield = BoneConfig.FOLLY.BASE_YIELD
543
- decay_factor = BoneConfig.FOLLY.DECAY_EXPONENT ** (times_eaten - 1)
544
- actual_yield = max(2.0, base_yield * decay_factor)
545
- loot = (
546
- "STABILITY_PIZZA"
547
- if actual_yield >= BoneConfig.FOLLY.PIZZA_THRESHOLD
548
- else None
549
- )
550
- flavor_text = f" (Stale: {times_eaten}x)" if times_eaten > 3 else ""
551
- msg = (
552
- f"{Prisma.RED}CROWD CAFFEINE: I chewed on '{target.upper()}'{flavor_text}.{Prisma.RST}\n"
553
- f" {Prisma.WHT}Yield: {actual_yield:.1f} ATP.{Prisma.RST}"
554
- )
555
- return "MEAT_GRINDER", msg, actual_yield, loot
556
-
557
- @staticmethod
558
- def _filter_meat_words(clean_words: list, _lexicon: Dict) -> list:
559
- meat_pool = set(LexiconService.get("heavy") or []) | \
560
- set(LexiconService.get("kinetic") or []) | \
561
- set(LexiconService.get("suburban") or [])
562
- return [w for w in clean_words if w in meat_pool]
563
-
564
- @staticmethod
565
- def _attempt_digest_abstract(
566
- clean_words: list, _lexicon: Dict
567
- ) -> Tuple[str, str, float, Optional[str]]:
568
- abstract_set = LexiconService.get("abstract")
569
- abstract_set = abstract_set if abstract_set else []
570
- abstract_words = [w for w in clean_words if w in abstract_set]
571
- if abstract_words:
572
- target = random.choice(abstract_words)
573
- yield_val = BoneConfig.FOLLY.YIELD_ABSTRACT
574
- msg = (
575
- f"{Prisma.GRY}THE FOLLY SIGHS: It grinds the ABSTRACT concept '{target.upper()}'.{Prisma.RST}\n"
576
- f" {Prisma.GRY}It tastes like chalk dust. +{yield_val} ATP.{Prisma.RST}"
577
- )
578
- return "GRUEL", msg, yield_val, None
579
- msg = (
580
- f"{Prisma.OCHRE}INDIGESTION: I tried to eat your words, but they were just air.{Prisma.RST}\n"
581
- f" {Prisma.GRY}Cannot grind this input into fuel.{Prisma.RST}\n"
582
- f" {Prisma.RED}► STARVATION CONTINUES.{Prisma.RST}"
583
- )
584
- return "INDIGESTION", msg, 0.0, None
585
-
586
- class ChronosKeeper:
587
- def __init__(self, engine_ref):
588
- self.eng = engine_ref
589
- self.SAVE_DIR = "saves"
590
- self.CRASH_DIR = "crashes"
591
-
592
- def save_checkpoint(self, history: list = None) -> str:
593
- try:
594
- if not os.path.exists(self.SAVE_DIR):
595
- os.makedirs(self.SAVE_DIR)
596
-
597
- loc = "Void"
598
- if (
599
- hasattr(self.eng, "phys")
600
- and hasattr(self.eng.phys, "observer")
601
- and getattr(self.eng.phys.observer, "last_physics_packet", None)
602
- ):
603
- loc = getattr(
604
- self.eng.phys.observer.last_physics_packet, "zone", "Void"
605
- )
606
-
607
- last_speech = "Silence."
608
- if self.eng.cortex.dialogue_buffer:
609
- last_speech = self.eng.cortex.dialogue_buffer[-1]
610
- continuity_packet = {
611
- "location": loc,
612
- "last_output": last_speech,
613
- "inventory": self.eng.gordon.inventory if self.eng.gordon else [],
614
- }
615
- start_history = (
616
- history if history is not None else self.eng.cortex.dialogue_buffer
617
- )
618
- state_data = {
619
- "health": self.eng.health,
620
- "stamina": self.eng.stamina,
621
- "trauma_accum": self.eng.trauma_accum,
622
- "soul_data": self.eng.soul.to_dict(),
623
- "village_data": self._gather_village_state(),
624
- "continuity": continuity_packet,
625
- "timestamp": time.time(),
626
- "chat_history": start_history,
627
- }
628
- path = os.path.join(self.SAVE_DIR, "quicksave.json")
629
- with open(path, "w", encoding="utf-8") as f:
630
- json.dump(state_data, f, indent=2, default=str)
631
- return f"✔ Checkpoint Saved: {path}"
632
- except Exception as e:
633
- self.eng.events.log(f"SAVE FAILED: {e}", "SYS_ERR")
634
- return f"❌ Save Failed: {e}"
635
-
636
- def resume_checkpoint(self) -> Tuple[bool, list]:
637
- path = os.path.join(self.SAVE_DIR, "quicksave.json")
638
- if not os.path.exists(path):
639
- print(
640
- f"{Prisma.GRY}[RESUME]: No quicksave found. Starting fresh.{Prisma.RST}"
641
- )
642
- return False, []
643
- try:
644
- print(f"{Prisma.CYN}[RESUME]: Hydrating from {path}...{Prisma.RST}")
645
- with open(path, "r", encoding="utf-8") as f:
646
- data = json.load(f)
647
- self.eng.health = data.get("health", 100.0)
648
- self.eng.stamina = data.get("stamina", 100.0)
649
- self.eng.trauma_accum = data.get("trauma_accum", {})
650
- if "soul_data" in data and hasattr(self.eng, "soul"):
651
- self.eng.soul.load_from_dict(data["soul_data"])
652
- if "village_data" in data:
653
- self._restore_village_state(data["village_data"])
654
- if "continuity" in data:
655
- self.eng.embryo.continuity = data["continuity"]
656
- if "inventory" in data["continuity"] and self.eng.gordon:
657
- self.eng.gordon.inventory = data["continuity"]["inventory"]
658
- restored_history = data.get("chat_history", [])
659
- print(f"{Prisma.GRN}[RESUME]: System State & Logs Restored.{Prisma.RST}")
660
- return True, restored_history
661
- except Exception as e:
662
- print(f"{Prisma.RED}[RESUME]: Failed to hydrate: {e}{Prisma.RST}")
663
- return False, []
664
-
665
- def perform_shutdown(self):
666
- print(f"{Prisma.GRY}...System Halt...{Prisma.RST}")
667
- self.eng.events.publish("SYSTEM_HALT", {"tick": self.eng.tick_count})
668
-
669
- loc = "Void"
670
- if (
671
- hasattr(self.eng, "phys")
672
- and hasattr(self.eng.phys, "observer")
673
- and getattr(self.eng.phys.observer, "last_physics_packet", None)
674
- ):
675
- loc = getattr(self.eng.phys.observer.last_physics_packet, "zone", "Void")
676
-
677
- continuity_packet = {
678
- "location": loc,
679
- "last_output": (
680
- self.eng.cortex.dialogue_buffer[-1]
681
- if self.eng.cortex.dialogue_buffer
682
- else "Silence."
683
- ),
684
- "inventory": self.eng.gordon.inventory if self.eng.gordon else [],
685
- }
686
- try:
687
- print(f"{Prisma.GRY}[MEMORY]: Freezing State...{Prisma.RST}")
688
- mito_traits = {}
689
- if hasattr(self.eng.bio.mito, "state"):
690
- mito_traits = self.eng.bio.mito.state.__dict__
691
- self.eng.mind.mem.save(
692
- health=self.eng.health,
693
- stamina=self.eng.stamina,
694
- mutations={},
695
- trauma_accum=self.eng.trauma_accum,
696
- joy_history=[],
697
- mitochondria_traits=mito_traits,
698
- antibodies=list(self.eng.bio.immune.active_antibodies),
699
- soul_data=self.eng.soul.to_dict(),
700
- village_data=self._gather_village_state(),
701
- continuity=continuity_packet,
702
- world_atlas=(
703
- self.eng.phys.nav.export_atlas()
704
- if hasattr(self.eng.phys, "nav")
705
- else {}
706
- ),
707
- )
708
- except Exception as e:
709
- print(f"{Prisma.RED}[MEMORY]: Save Failed: {e}{Prisma.RST}")
710
- subsystems = [
711
- ("LEXICON", self.eng.lex, "save"),
712
- ("AKASHIC", self.eng.akashic, "save_all"),
713
- ]
714
- for name, sys, method in subsystems:
715
- if hasattr(sys, method):
716
- try:
717
- print(f"{Prisma.GRY}[{name}]: Persisting...{Prisma.RST}")
718
- getattr(sys, method)()
719
- except Exception as e:
720
- print(f"{Prisma.RED}[{name}]: Failed: {e}{Prisma.RST}")
721
-
722
- def _gather_village_state(self) -> Dict[str, Any]:
723
- state = {}
724
- for name, component in self.eng.village.items():
725
- if component and hasattr(component, "to_dict"):
726
- state[name] = component.to_dict()
727
- return state
728
-
729
- def _restore_village_state(self, state_data: Dict[str, Any]):
730
- if not state_data:
731
- return
732
- for name, data in state_data.items():
733
- if (
734
- name in self.eng.village
735
- and self.eng.village[name]
736
- and hasattr(self.eng.village[name], "load_state")
737
- ):
738
- try:
739
- self.eng.village[name].load_state(data)
740
- except Exception as e:
741
- print(
742
- f"{Prisma.RED}[RESUME]: Failed to hydrate {name}: {e}{Prisma.RST}"
743
- )
744
-
745
- def get_crash_path(self, prefix="crash"):
746
- if not os.path.exists(self.CRASH_DIR):
747
- try:
748
- os.makedirs(self.CRASH_DIR)
749
- except OSError:
750
- pass
751
- try:
752
- files = sorted(
753
- [f for f in os.listdir(self.CRASH_DIR) if f.startswith(prefix)]
754
- )
755
- for oldest in files[:-4]:
756
- os.remove(os.path.join(self.CRASH_DIR, oldest))
757
- except Exception:
758
- pass
759
- return os.path.join(self.CRASH_DIR, f"{prefix}_{int(time.time())}.json")
760
-
761
- @staticmethod
762
- def emergency_dump(exit_cause="UNKNOWN") -> str:
763
- return f"✔ Emergency Dump: {exit_cause}"