"""GameGirl Color -- Two-tier game memory system.
Manages per-game memories in Redis with two tiers:
- Basic: Important, long-lived (characters, plot, bosses, items)
- Channel: Ephemeral, recent context (recent events, minor NPCs)
Includes hot-swap memory bleed logic for carrying context
between cartridge swaps.
# 🧠💀 CORRUPTED MEMORY BANKS
"""
from __future__ import annotations
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from typing import Any
logger = logging.getLogger(__name__)
# Redis key patterns # 🕷️
_BASIC_KEY = "game:mem:basic:{game_id}"
_CHANNEL_KEY = "game:mem:channel:{game_id}"
_MAX_BASIC = 50
_MAX_CHANNEL = 30
_CHANNEL_TTL = 0 # No TTL -- 5ever
_BASIC_TTL = 0 # No TTL -- 5ever
_PROMOTION_THRESHOLD = 3 # References needed to promote
[docs]
@dataclass
class GameMemory:
"""A single memory entry in the game's memory bank."""
label: str
content: str
importance: float = 0.5 # 0.0 - 1.0 scale
turn_created: int = 0
turn_last_referenced: int = 0
reference_count: int = 1
category: str = "general" # general, character, plot, boss, item
glitched: bool = False # True if bleeded from another game
source_game: str = "" # Original game name if glitched
created_at: float = field(default_factory=time.time)
[docs]
def to_dict(self) -> dict[str, Any]:
return asdict(self)
[docs]
@classmethod
def from_dict(cls, d: dict[str, Any]) -> GameMemory:
valid_fields = cls.__dataclass_fields__
return cls(**{k: v for k, v in d.items() if k in valid_fields})
# =====================================================================
# Storage operations # 🌀
# =====================================================================
[docs]
async def store_basic(
game_id: str,
label: str,
content: str,
importance: float = 0.7,
category: str = "general",
turn: int = 0,
redis: Any = None,
) -> bool:
"""Store or update a basic (important) memory.
Basic memories persist for the life of the game. Used for
characters, plot points, bosses, key items, relationships.
"""
if redis is None:
return False
key = _BASIC_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
memories: list[dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
memories = []
# Check for existing memory with same label # 💀
found = False
for mem in memories:
if mem.get("label", "").lower() == label.lower():
mem["content"] = content
mem["importance"] = max(mem.get("importance", 0), importance)
mem["reference_count"] = mem.get("reference_count", 1) + 1
mem["turn_last_referenced"] = turn
found = True
break
if not found:
entry = GameMemory(
label=label,
content=content,
importance=importance,
category=category,
turn_created=turn,
turn_last_referenced=turn,
)
memories.append(entry.to_dict())
# Trim to max (keep highest importance) # 🔥
if len(memories) > _MAX_BASIC:
memories.sort(key=lambda m: m.get("importance", 0), reverse=True)
memories = memories[:_MAX_BASIC]
try:
await redis.set(key, json.dumps(memories))
return True
except Exception as exc:
logger.error("Failed to store basic game memory: %s", exc)
return False
[docs]
async def store_channel(
game_id: str,
label: str,
content: str,
turn: int = 0,
redis: Any = None,
) -> bool:
"""Store a channel (ephemeral) memory.
Channel memories are recent context — scene descriptions,
minor NPCs, dialogue snippets. FIFO eviction.
Auto-promoted to basic if referenced 3+ times.
"""
if redis is None:
return False
key = _CHANNEL_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
memories: list[dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
memories = []
# Check for existing # 🕷️
found = False
for mem in memories:
if mem.get("label", "").lower() == label.lower():
mem["content"] = content
mem["reference_count"] = mem.get("reference_count", 1) + 1
mem["turn_last_referenced"] = turn
# Auto-promote to basic if referenced enough # 😈
if mem["reference_count"] >= _PROMOTION_THRESHOLD:
await store_basic(
game_id, label, content,
importance=0.6,
category="promoted",
turn=turn,
redis=redis,
)
# Remove from channel after promotion
memories = [
m for m in memories
if m.get("label", "").lower() != label.lower()
]
logger.info(
"Memory '%s' promoted to basic (ref count: %d)",
label, mem["reference_count"],
)
found = True
break
if not found:
entry = GameMemory(
label=label,
content=content,
importance=0.3,
category="channel",
turn_created=turn,
turn_last_referenced=turn,
)
memories.append(entry.to_dict())
# FIFO eviction # 💦
if len(memories) > _MAX_CHANNEL:
memories = memories[-_MAX_CHANNEL:]
try:
await redis.set(key, json.dumps(memories))
return True
except Exception as exc:
logger.error("Failed to store channel game memory: %s", exc)
return False
# =====================================================================
# Retrieval # 📼
# =====================================================================
[docs]
async def get_basic_memories(
game_id: str,
redis: Any = None,
) -> list[GameMemory]:
"""Get all basic memories for a game."""
if redis is None:
return []
key = _BASIC_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return []
return [GameMemory.from_dict(d) for d in json.loads(raw)]
except Exception as exc:
logger.error("Failed to read basic memories: %s", exc)
return []
[docs]
async def get_channel_memories(
game_id: str,
redis: Any = None,
) -> list[GameMemory]:
"""Get all channel memories for a game."""
if redis is None:
return []
key = _CHANNEL_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return []
return [GameMemory.from_dict(d) for d in json.loads(raw)]
except Exception as exc:
logger.error("Failed to read channel memories: %s", exc)
return []
[docs]
async def get_context_summary(
game_id: str,
redis: Any = None,
) -> str:
"""Build a formatted memory context block for the system prompt.
Returns a string suitable for injection into the LLM context,
organized by importance and category.
"""
basic = await get_basic_memories(game_id, redis=redis)
channel = await get_channel_memories(game_id, redis=redis)
if not basic and not channel:
return "[GAME MEMORY: Empty — no memories recorded yet]"
lines: list[str] = ["[GAME MEMORY BANK]"]
if basic:
# Sort by importance # 💀
basic.sort(key=lambda m: m.importance, reverse=True)
lines.append("=== CORE MEMORIES (important) ===")
for mem in basic:
glitch_tag = " [BLEED-THROUGH]" if mem.glitched else ""
lines.append(
f"* [{mem.category.upper()}] {mem.label}: "
f"{mem.content}{glitch_tag}"
)
if channel:
lines.append("=== RECENT CONTEXT (ephemeral) ===")
for mem in channel[-10:]: # Last 10 only
lines.append(f"- {mem.label}: {mem.content}")
return "\n".join(lines)
# =====================================================================
# Hot-swap memory bleed # 🌀🔥
# =====================================================================
[docs]
async def bleed_memories(
source_game_id: str,
target_game_id: str | None,
redis: Any = None,
max_bleed: int = 7,
) -> str:
"""Transfer important memories from source to target game.
Selects top memories by importance, with priority for:
- Antagonists/bosses (category == 'boss')
- Active combat state
- High-importance plot points
Bleeded memories get the `glitched` flag and `source_game` tag.
Channel memories are discarded (ephemeral by design).
Returns a summary of what bled through.
"""
if redis is None:
return "[BLEED-THROUGH] No Redis connection — memories lost in static."
basic = await get_basic_memories(source_game_id, redis=redis)
if not basic:
return "[BLEED-THROUGH] Source game had no memories to bleed."
# Prioritize bosses/antagonists, then by importance # 😈
boss_memories = [m for m in basic if m.category in ("boss", "antagonist")]
other_memories = [m for m in basic if m.category not in ("boss", "antagonist")]
other_memories.sort(key=lambda m: m.importance, reverse=True)
# Bosses always bleed through, plus top N others
to_bleed = boss_memories + other_memories
to_bleed = to_bleed[:max_bleed]
if not to_bleed:
return "[BLEED-THROUGH] No memories strong enough to survive the swap."
if target_game_id is None:
# Just return what would bleed (for preview)
names = [m.label for m in to_bleed]
return (
f"[BLEED-THROUGH] {len(to_bleed)} memories ready to bleed: "
f"{', '.join(names)}"
)
# Glitch-tag and inject into target # 🌀
bleed_labels: list[str] = []
for mem in to_bleed:
glitch_label = f"[BLEED] {mem.label}"
glitch_content = (
f"[Glitched from '{mem.source_game or 'unknown cartridge'}'] "
f"{mem.content}"
)
await store_basic(
target_game_id,
label=glitch_label,
content=glitch_content,
importance=min(1.0, mem.importance + 0.1), # Slightly boosted
category=mem.category,
redis=redis,
)
# Set glitch flags on the stored entry # 💀
key = _BASIC_KEY.format(game_id=target_game_id)
try:
raw = await redis.get(key)
if raw:
memories = json.loads(raw)
for stored in memories:
if stored.get("label") == glitch_label:
stored["glitched"] = True
stored["source_game"] = mem.source_game or source_game_id
break
await redis.set(key, json.dumps(memories))
except Exception:
pass
bleed_labels.append(mem.label)
boss_count = len(boss_memories)
return (
f"[BLEED-THROUGH] {len(bleed_labels)} memories bled through "
f"({boss_count} boss/antagonist, {len(bleed_labels) - boss_count} other): "
f"{', '.join(bleed_labels)}. "
f"They carry the [BLEED-THROUGH] tag and will glitch into the "
f"new narrative as corrupted echoes."
)
[docs]
async def clear_game_memories(
game_id: str,
redis: Any = None,
) -> None:
"""Delete all memories for a game."""
if redis is None:
return
try:
await redis.delete(
_BASIC_KEY.format(game_id=game_id),
_CHANNEL_KEY.format(game_id=game_id),
)
except Exception as exc:
logger.error("Failed to clear game memories: %s", exc)