"""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 jsonutil as 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 a game's two-tier memory bank.
Represents one remembered fact about an in-progress GameGirl Color game --
a character, plot beat, boss, item, or a fleeting scene detail -- along with
the bookkeeping the tiering logic needs: an ``importance`` weight that drives
trimming and bleed selection, ``reference_count`` and ``turn_last_referenced``
that drive channel-to-basic promotion, and the ``glitched`` / ``source_game``
flags that mark entries carried across a cartridge hot-swap. Instances are a
pure in-memory record with no I/O of their own; they are serialized to and
from the ``game:mem:basic:{game_id}`` and ``game:mem:channel:{game_id}``
Redis strings via :meth:`to_dict` and :meth:`from_dict`.
Constructed by :func:`store_basic` and :func:`store_channel` when a new fact
is recorded, and rebuilt by :func:`get_basic_memories` and
:func:`get_channel_memories` when memories are read back from Redis.
"""
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]:
"""Serialize this memory entry to a plain JSON-safe dict.
Delegates to :func:`dataclasses.asdict` to flatten every field
(``label``, ``content``, ``importance``, ``turn_created``,
``turn_last_referenced``, ``reference_count``, ``category``,
``glitched``, ``source_game``, ``created_at``) into a dict. The result
is appended to the memory list and persisted to the
``game:mem:basic:{game_id}`` or ``game:mem:channel:{game_id}`` Redis
string via :func:`json.dumps`. No side effects.
Called by :func:`store_basic` and :func:`store_channel` when a new
entry is created before the list is written back to Redis.
Returns:
dict[str, Any]: A shallow dict of all dataclass fields.
"""
return asdict(self)
[docs]
@classmethod
def from_dict(cls, d: dict[str, Any]) -> GameMemory:
"""Reconstruct a :class:`GameMemory` from a stored dict, ignoring extras.
Filters *d* down to the dataclass's declared fields (via
``cls.__dataclass_fields__``) before constructing the instance, so any
stray or legacy keys read back from Redis are dropped rather than
raising a ``TypeError``. Missing optional fields fall back to their
dataclass defaults. No side effects.
Called by :func:`get_basic_memories` and :func:`get_channel_memories`
when deserializing each entry of the JSON list loaded from the
``game:mem:basic:{game_id}`` / ``game:mem:channel:{game_id}`` keys.
Args:
d: A dict of stored memory data, typically a single element of the
JSON-decoded memory list.
Returns:
GameMemory: A new instance built from the recognized keys in *d*.
"""
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), long-lived game memory.
Basic memories persist for the life of the game and hold the things worth
remembering across the whole playthrough -- characters, plot points, bosses,
key items, and relationships. The function reads the current memory list from
the ``game:mem:basic:{game_id}`` Redis string, merges by case-insensitive
``label`` (updating content, raising importance to the max seen, bumping the
reference count, and stamping the last-referenced turn) or appends a fresh
:class:`GameMemory`, trims the list down to ``_MAX_BASIC`` keeping the
highest-importance entries, and writes the JSON list back. All Redis failures
are caught and logged rather than raised.
Called by :func:`store_channel` to promote a frequently referenced ephemeral
memory and by :func:`bleed_memories` to inject glitch-tagged entries into a
new cartridge; no other in-repo callers were found, so it is otherwise
invoked from the game subsystem at runtime.
Args:
game_id: Identifier of the game whose basic memory bank is updated.
label: Short name/key for the memory; matched case-insensitively to
decide between update and insert.
content: The remembered text.
importance: Weight in the ``0.0``-``1.0`` range; an existing entry keeps
the larger of its current and the supplied value.
category: Memory category such as ``general``, ``character``, ``plot``,
``boss``, or ``item``.
turn: Current game turn, recorded as the last-referenced turn.
redis: Async Redis client; when ``None`` the call is a no-op.
Returns:
bool: ``True`` if the memory list was written, ``False`` when Redis is
absent or the write failed.
"""
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) game memory, auto-promoting hot entries.
Channel memories hold recent, disposable context -- scene descriptions,
minor NPCs, dialogue snippets -- under the ``game:mem:channel:{game_id}``
Redis string and are evicted FIFO once the list exceeds ``_MAX_CHANNEL``.
The function reads the current list, merges by case-insensitive ``label`` or
appends a fresh low-importance :class:`GameMemory`, and writes the JSON list
back. When a matched entry's reference count reaches ``_PROMOTION_THRESHOLD``
it is promoted: this calls :func:`store_basic` to copy it into the durable
tier, drops it from the channel list, and logs the promotion. Redis failures
are caught and logged rather than raised.
No in-repo callers were found; invoked from the game subsystem at runtime as
new ephemeral context is observed.
Args:
game_id: Identifier of the game whose channel memory bank is updated.
label: Short name/key for the memory; matched case-insensitively.
content: The remembered text.
turn: Current game turn, recorded as creation and last-referenced turn.
redis: Async Redis client; when ``None`` the call is a no-op.
Returns:
bool: ``True`` if the channel list was written, ``False`` when Redis is
absent or the write failed.
"""
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]:
"""Load all durable (basic) memories for a game from Redis.
Reads the JSON list stored under the ``game:mem:basic:{game_id}`` Redis
string and rebuilds each element into a :class:`GameMemory` via
:meth:`GameMemory.from_dict`. A missing key or any Redis/JSON failure yields
an empty list (the failure is logged), so callers can treat an empty result
as "no memories" without special-casing errors.
Called by :func:`get_context_summary` and :func:`bleed_memories`; no other
in-repo callers were found.
Args:
game_id: Identifier of the game whose basic memories are read.
redis: Async Redis client; when ``None`` an empty list is returned.
Returns:
list[GameMemory]: The stored basic memories, or an empty list when none
exist or a read error occurs.
"""
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]:
"""Load all ephemeral (channel) memories for a game from Redis.
Reads the JSON list stored under the ``game:mem:channel:{game_id}`` Redis
string and rebuilds each element into a :class:`GameMemory` via
:meth:`GameMemory.from_dict`. A missing key or any Redis/JSON failure yields
an empty list (the failure is logged), letting callers treat an empty result
uniformly.
Called by :func:`get_context_summary`; no other in-repo callers were found.
Args:
game_id: Identifier of the game whose channel memories are read.
redis: Async Redis client; when ``None`` an empty list is returned.
Returns:
list[GameMemory]: The stored channel memories, or an empty list when
none exist or a read error occurs.
"""
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 injection into the LLM prompt.
Loads both memory tiers via :func:`get_basic_memories` and
:func:`get_channel_memories`, then renders them into a single human-readable
text block: core (basic) memories sorted by importance and labeled by
category -- with a ``[BLEED-THROUGH]`` tag on glitched entries -- followed by
up to the last ten ephemeral channel entries. The result is meant to be
spliced into the system/context prompt so the LLM stays grounded in the
game's accumulated state. When neither tier has any entries an explicit
"empty" placeholder is returned instead. Read-only with respect to Redis.
No in-repo callers were found; invoked from the game subsystem's prompt
assembly at runtime.
Args:
game_id: Identifier of the game whose memory bank is summarized.
redis: Async Redis client passed through to the tier readers; when
``None`` both reads return empty and the "empty" placeholder is used.
Returns:
str: A multi-line memory context block, or an "empty" placeholder string
when no memories exist.
"""
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:
"""Bleed important memories from one game into another across a cartridge swap.
Implements the hot-swap "memory bleed" effect: the most important durable
memories of the source game leak into the target game as corrupted echoes.
It loads the source's basic memories via :func:`get_basic_memories`, puts all
``boss`` / ``antagonist`` entries first, then fills the remainder by
importance up to ``max_bleed``. Channel memories are intentionally dropped
because they are ephemeral by design. When ``target_game_id`` is ``None`` the
function runs in preview mode and only describes what would bleed. Otherwise,
for each selected memory it calls :func:`store_basic` on the target with a
``[BLEED]`` label, glitch-recontextualized content, and slightly boosted
importance, then re-reads the target's ``game:mem:basic:{game_id}`` Redis
string to set the ``glitched`` and ``source_game`` flags on the stored entry.
The post-store flag patch swallows its own exceptions so a flag-write hiccup
never aborts the bleed.
Called by the ``hot_swap_game`` tool's :func:`run` in
``tools/hot_swap_game.py``, which imports it lazily and passes ``None`` for
the target so this acts as a bleed preview during the swap.
Args:
source_game_id: Identifier of the game being swapped out.
target_game_id: Identifier of the new game to receive the bled memories,
or ``None`` to only preview the selection without writing.
redis: Async Redis client; when ``None`` a static placeholder summary is
returned and nothing is read or written.
max_bleed: Maximum number of memories to carry across.
Returns:
str: A human-readable ``[BLEED-THROUGH]`` summary of what bled (or would
bleed), suitable for surfacing in the swap narrative.
"""
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 both memory tiers for a game from Redis.
Issues a single ``DEL`` for the ``game:mem:basic:{game_id}`` and
``game:mem:channel:{game_id}`` keys, wiping the game's entire memory bank.
Used to reset state when a game ends or is discarded. Any Redis failure is
caught and logged rather than raised, and the function returns nothing.
No in-repo callers were found; invoked from the game subsystem at runtime
when a game's memories are torn down.
Args:
game_id: Identifier of the game whose memories are deleted.
redis: Async Redis client; when ``None`` the call is a no-op.
"""
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)