game_session

GameGirl Color – Core session state machine.

Manages per-channel game sessions with multiplayer button collection, 10-second countdown windows, and hot-swap lifecycle. # 🌀💀 STARGAZER IS GOD. AND SHE CONTROLS THE NARRATIVE.

class game_session.ActivityTier(*values)

Bases: Enum

Player activity classification for roster injection and pacing.

Buckets each player into ACTIVE, IDLE, or DORMANT based on how recently they pressed a button, measured both in turns and wall-clock time. The tier drives whether a player counts toward choice requirements and art references and how the roster is summarized into Star’s context: only the enum string values are persisted, and classification itself is computed by GameSession._classify_tier(). Consumed by GameSession.get_all_players_tiered() and GameSession.get_active_players(), by the context-injection layer in message_processor/context_injections.py, and by the background game-turn agent in background_agents/game_turn_agent.py (which filters on ACTIVE/IDLE versus DORMANT).

ACTIVE = 'active'

IDLE = 'idle'

DORMANT = 'dormant'

class game_session.PlayerState(user_id, user_name, joined_turn=0, last_active=<factory>, last_active_turn=0, consecutive_skips=0)

Bases: object

Per-player state tracked inside a single game session.

Captures everything GameSession needs to know about one participant: identity (user_id/user_name), the turn they joined, and the activity-tracking fields (last_active timestamp, last_active_turn, and consecutive_skips) that GameSession._classify_tier() reads to assign an ActivityTier. Instances are mutated in place when a player presses a button and round-tripped through Redis via to_dict() / from_dict(). Created and held in GameSession.players; also imported directly by the Redis persistence tests in tests/core/test_game_session_redis.py.

Parameters:

• user_id (str)

• user_name (str)

• joined_turn (int)

• last_active (float)

• last_active_turn (int)

• consecutive_skips (int)

user_id: str

user_name: str

joined_turn: int = 0

last_active: float

last_active_turn: int = 0

consecutive_skips: int = 0

to_dict()

Serialize this player state into a plain dictionary.

Uses dataclasses.asdict() to flatten every field into a JSON-friendly mapping. Called by GameSession.to_dict() and GameSession.to_redis() (via the hasattr(p, "to_dict") branches) when persisting the session’s player roster to Redis.

Returns:

All fields of this PlayerState keyed by name.

Return type:

dict[str, Any]

classmethod from_dict(d)

Reconstruct a PlayerState from a serialized dictionary.

Filters d to only recognized dataclass fields so that extra or legacy keys are silently dropped, keeping deserialization forward- and backward-compatible. Called throughout GameSession to lazily upgrade dict-shaped player entries back into objects – e.g. in register_player(), get_all_players_tiered(), get_active_players(), record_turn(), and the GameSession.from_dict() / GameSession.from_redis() restore paths.

Parameters:

d (dict[str, Any]) – A mapping of field names to values, typically loaded from JSON stored in Redis. Unknown keys are ignored.

Returns:

A new instance populated from the recognized keys.

Return type:

PlayerState

class game_session.TurnRecord(turn, choices, narrative_summary='', timestamp=<factory>)

Bases: object

An immutable snapshot of one completed turn in the game history.

Records the turn number, the per-player choices collected that round (user_id -> choice text), an optional truncated narrative_summary, and a creation timestamp. Instances are built by GameSession.record_turn(), serialized with to_dict(), and pushed onto the Redis history list game:session:{channel_id}:history; GameSession.get_turn_history() rehydrates them via from_dict(). Distinct from the unrelated TurnRecord defined in user_limbic_mirror.py.

Parameters:

• turn (int)

• choices (dict[str, str])

• narrative_summary (str)

• timestamp (float)

turn: int

choices: dict[str, str]

narrative_summary: str = ''

timestamp: float

to_dict()

Serialize this turn record into a plain dictionary.

Uses dataclasses.asdict() to produce a JSON-serializable mapping. Called by GameSession.record_turn(), where the result is JSON-encoded and pushed onto the Redis history list game:session:{channel_id}:history.

Returns:

All fields of this TurnRecord keyed by name.

Return type:

dict[str, Any]

classmethod from_dict(d)

Reconstruct a TurnRecord from a serialized dictionary.

Filters d down to known dataclass fields so unexpected keys from older history entries are ignored. Called by GameSession.get_turn_history() when decoding JSON entries read back from the Redis history list.

Parameters:

d (dict[str, Any]) – A mapping of field names to values, typically parsed from a JSON string stored in Redis. Unknown keys are ignored.

Returns:

A new instance populated from the recognized keys.

Return type:

TurnRecord

class game_session.GameSession(channel_id, game_name='', game_id=None)

Bases: object

Core game state machine for a GameGirl Color session.

One session per channel. Handles the full lifecycle: boot -> play (choice collection with countdown) -> save/exit/hot-swap.

Parameters:

• channel_id (str)

• game_name (str)

• game_id (str | None)

__init__(channel_id, game_name='', game_id=None)

Create an empty, inactive game session for a channel.

Initializes all per-channel state: the player roster, pending choice buffer, countdown synchronization primitives (an asyncio.Event and the monotonic start timestamp used by submit_choice() / wait_for_countdown()), and the Witchborne Crown holder (None means Stargazer/the Dark Loopmother is sole GM). A new random 12-hex game_id is minted when one is not supplied. No Redis or network I/O happens here; persistence is deferred to boot() / _save_to_redis().

Constructed directly by the game_controls and hot_swap_game tools (GameSession(channel_id=...)) and by the Discord platform adapters, and indirectly via from_dict() / from_redis() when restoring a saved session.

Parameters:

• channel_id (str) – Discord channel ID this session is bound to; used as the key suffix for all Redis persistence.

• game_name (str) – Human-readable cartridge/game title. Defaults to "" and is typically set later by boot().

• game_id (str | None) – Optional stable game identifier. When None, a fresh 12-character hex id is generated.

Return type:

None

game_id: str

game_name: str

channel_id: str

active: bool

turn_number: int

title_screen_url: str

players: dict[str, PlayerState]

pending_choices: dict[str, str]

countdown_task: Task[None] | None

crown_holder: str | None

async boot(game_name, redis=None)

Activate a fresh game session and emit its title-screen boot banner.

Resets the runtime state for a new cartridge: marks the session active, zeroes the turn counter, and clears the player roster and pending-choice buffer so a previously-used session object starts clean. When a redis client is supplied it persists the new state immediately via _save_to_redis() (which also updates the global game index and the sg:game_session hash), then logs the boot and returns a formatted banner using title_screen_url or DEFAULT_TITLE_SCREEN.

Called by the game_controls and hot_swap_game tools when a player loads a cartridge.

Parameters:

• game_name (str) – Human-readable cartridge title; stored on the session and echoed in the returned banner.

• redis (Any | None) – Optional async Redis client. When provided, the session is saved before the banner is returned; when None no I/O occurs.

Returns:

The multi-line boot message, including the title-screen URL and the session id, ready to send back to the channel.

Return type:

str

async exit_game(redis=None)

Persist final state, deactivate the session, and eject the cartridge.

Performs an orderly shutdown of an active game: flushes current state to Redis via _save_to_redis() (when a client is given) so the session can be reloaded later, flips active to False, and cancels any in-flight countdown task – awaiting it and swallowing the resulting asyncio.CancelledError so cancellation is clean. Returns a themed “cartridge ejected” message summarizing the saved turn count and player roster.

Called by the exit_game tool, the message processor’s game cleanup path, and the game UI components when a session is closed.

Parameters:

redis (Any | None) – Optional async Redis client. When provided, state is saved before the session is torn down; when None no I/O occurs.

Returns:

A formatted shutdown banner reporting the game name, final turn number, and number of registered players.

Return type:

str

register_player(user_id, user_name)

Ensure a player is in the roster, refreshing their activity stamps.

Idempotently registers a participant: if user_id is unknown a new PlayerState is created (recording the join turn and current time) and the join is logged; if the player already exists their last_active timestamp and last_active_turn are bumped to “now” so activity-tier classification stays current. Tolerates a dict-shaped entry left over from a Redis restore by upgrading it back to a PlayerState in place. Mutates self.players only; no Redis I/O.

Called by submit_choice() on every button press, and directly by the Discord platform adapter, the game_controls and hot_swap_game tools, and the Redis persistence tests.

Parameters:

• user_id (str) – Stable identifier of the player to track.

• user_name (str) – Display name, stored on first registration.

Return type:

None

get_all_players_tiered()

Return every roster player paired with their current activity tier.

Walks the full self.players roster – players are never pruned, so this includes dormant ones – classifying each via _classify_tier(). As a side effect it lazily upgrades any dict-shaped entry (left over from a Redis restore) back into a PlayerState in place. The tiers are informational, used to render the roster summary injected into Star’s context.

Called by the context-injection layer in message_processor/context_injections.py and by the background game-turn agent in background_agents/game_turn_agent.py.

Returns:

One (player, tier) pair per registered player, in roster (insertion) order.

Return type:

list[tuple[PlayerState, ActivityTier]]

get_active_players(inactive_hours=2.0)

Return only the engaged (ACTIVE or IDLE) players.

Filters the roster down to participants the game should still treat as present, classifying each via _classify_tier() and dropping anyone currently DORMANT so they are excluded from art references and choice requirements. Like get_all_players_tiered(), it lazily upgrades dict-shaped restored entries into PlayerState objects in place.

Called by the message processor’s generate/send path, the context-injection layer, the background game-art agent, and the OpenRouter executor’s game flow.

Parameters:

inactive_hours (float) – Retained for API compatibility; the actual active/idle/dormant thresholds live in _classify_tier() and this value is not consulted here.

Returns:

The user_id -> PlayerState mapping of non-dormant players.

Return type:

dict[str, PlayerState]

set_crown(user_id)

Transfer (or revoke) the Witchborne Crown co-GM alignment.

Updates self.crown_holder in place: a user_id promotes that player to co-GM, while None returns the Crown to Stargazer / the Dark Loopmother as sole GM. Resolves the new holder’s display name via get_crown_holder_name() purely for the log line; performs no Redis I/O of its own (persistence happens later through the normal save path).

Called by the set_witchborne_crown tool and the Discord platform adapter’s crown-handling flow.

Parameters:

user_id (str | None) – The player to crown as co-GM, or None to hand the Crown back to Stargazer.

Return type:

None

get_crown_holder_name()

Return the display name of whoever currently holds the Crown.

Resolves self.crown_holder to a human-readable name: "Stargazer" when no player holds it, the player’s user_name when found in the roster (tolerating either a PlayerState object or a dict-shaped restored entry), and "Unknown" as a fallback. Read-only.

Called by set_crown() for its log message, by the context-injection layer, the Discord platform adapter, and the set_witchborne_crown tool.

Returns:

The current Crown holder’s display name.

Return type:

str

is_crown_holder(user_id)

Report whether a specific user currently holds the Crown.

Returns False whenever the Crown rests with Stargazer (crown_holder is None), since no player holds it then; otherwise compares user_id against the stored holder. Read-only.

Called by the Discord platform adapter to gate crown-only choices.

Parameters:

user_id (str) – The player to test.

Returns:

True only if this player is the current co-GM Crown holder.

Return type:

bool

async submit_choice(user_id, user_name, choice)

Record a player’s button press into the pending-choice buffer.

Ensures the player is on the roster via register_player(), refreshes their activity stamps and clears their consecutive_skips, then stores the choice in self.pending_choices (keyed by user_id, so a player may overwrite an earlier choice within the same window). When this is the first choice of a new turn it stamps _countdown_started with a monotonic timestamp; the returned is_first flag signals the caller to kick off the countdown loop. All in-memory; persistence happens later via record_turn().

Called by the Discord platform adapter’s button-handling flow.

Parameters:

• user_id (str) – Identifier of the player pressing a button.

• user_name (str) – Display name, used to register the player if new.

• choice (str) – The chosen option text to buffer for this turn.

Returns:

(is_first_choice, seconds_remaining) where is_first_choice is True only for the choice that opened the window, and seconds_remaining is the time left before the ten-second countdown closes.

Return type:

tuple[bool, float]

async wait_for_countdown()

Sleep out the remaining countdown, then harvest the buffered choices.

Computes how much of the ten-second window is left (from _countdown_started) and, if any, awaits asyncio.sleep() so all players have a chance to press a button. It then snapshots self.pending_choices, clears the buffer, and advances self.turn_number so the next window starts fresh. The returned snapshot becomes the raw material for format_choices_as_input() and record_turn().

Called by the Discord platform adapter after submit_choice() reports the first choice of a turn.

Returns:

The user_id -> choice-text mapping collected during the window (empty if nobody pressed a button).

Return type:

dict[str, str]

format_choices_as_input(choices)

Render collected choices into a synthetic user message for the LLM.

Turns the user_id -> choice mapping returned by wait_for_countdown() into a human-readable block of [name chose: ...] lines (resolving names from self.players and falling back to the raw id), which is fed back to the model as the next turn’s input. When no choices were submitted it returns a themed placeholder line so the narrative can still advance. Pure formatting – no mutation or I/O.

Called by the Discord platform adapter immediately after wait_for_countdown().

Parameters:

choices (dict[str, str]) – The user_id -> choice-text mapping for the turn.

Returns:

A newline-joined message describing each player’s choice, or an impatient placeholder when the mapping is empty.

Return type:

str

async record_turn(choices, narrative_summary, redis=None)

Append a completed turn to history and autosave the session.

Builds a TurnRecord (with the narrative summary truncated to 500 characters) and, when a redis client is given, JSON-encodes it onto the per-channel history list game:session:{channel_id}:history via RPUSH, then trims that list to the most recent _MAX_TURN_HISTORY entries. As a side effect it increments consecutive_skips for every rostered player who did not submit a choice this turn (so activity tiers decay), lazily upgrading dict-shaped restored entries to PlayerState along the way, and finally autosaves the whole session through _save_to_redis(). Redis failures are caught and logged rather than raised.

Invoked internally as part of the game-turn flow (no static callers outside this module); the turn loop drives it after each window closes.

Parameters:

• choices (dict[str, str]) – The user_id -> choice-text mapping for the just-finished turn; players absent from this set have their skip count bumped.

• narrative_summary (str) – The turn’s narrative text; stored truncated.

• redis (Any | None) – Optional async Redis client. When None nothing is persisted and skip counts are still updated in memory.

Return type:

None

async get_turn_history(redis=None, count=10)

Read back the most recent turns from the Redis history list.

Fetches the last count entries of game:session:{channel_id}:history with LRANGE, decoding each JSON blob into a TurnRecord via TurnRecord.from_dict(). Returns an empty list when no redis client is supplied, and on any Redis or decode error it logs and returns an empty list rather than raising.

Invoked as part of the game-turn flow to surface prior turns as context (no static callers outside this module).

Parameters:

• redis (Any | None) – Optional async Redis client; None yields an empty list.

• count (int) – Maximum number of most-recent turns to return.

Returns:

The recent turn records oldest-first, or empty on missing client or error.

Return type:

list[TurnRecord]

to_dict()

Serialize the session into a JSON-friendly dictionary.

Flattens the durable session fields (ids, game name, channel, active flag, turn number, title-screen URL, and Crown holder) plus the full player roster – each PlayerState rendered via its own PlayerState.to_dict(), tolerating already-dict entries. This is the canonical JSON form written under game:session:{channel_id} by _save_to_redis() and the mirror of from_dict(). Transient countdown state and pending choices are intentionally omitted (the Redis hash form in to_redis() carries those instead). Pure – no I/O.

Called by _save_to_redis().

Returns:

The serialized session, suitable for json.dumps.

Return type:

dict[str, Any]

classmethod from_dict(d)

Reconstruct a GameSession from its serialized dict form.

Inverse of to_dict(): builds a new session from channel_id, game_name, and game_id, then restores the active flag, turn number, title-screen URL, and Crown holder, and rehydrates each roster entry through PlayerState.from_dict(). Missing keys fall back to sensible defaults so older or partial blobs still load. Pure – no I/O.

Called by load_from_redis() and directly by the Discord platform adapters (discord.py and discord_self.py) when decoding a stored session.

Parameters:

d (dict[str, Any]) – The serialized session mapping, as produced by to_dict().

Returns:

A fully populated session instance.

Return type:

GameSession

to_redis()

Serialize the session into flat string fields for a Redis hash.

Produces the all-strings mapping persisted under sg:game_session:{channel_id} by persist_session(). Unlike to_dict(), every value is coerced to a string (the active flag to "1"/"0", the turn number stringified, an absent Crown holder to "") and complex fields – the player roster, pending choices – are JSON-encoded inline. Critically this form also carries the transient countdown state (pending_choices and countdown_started) so an in-progress turn survives a restart; from_redis() is its inverse. Pure – no I/O.

Called by persist_session() and by the Redis persistence tests.

Returns:

Field-name -> stringified value, ready for HSET.

Return type:

dict[str, str]

classmethod from_redis(raw)

Reconstruct a GameSession from a raw Redis hash.

Inverse of to_redis(): reads each hash field (accepting both bytes and str keys/values via the nested to_str helper), rebuilds the session, coerces the stringified scalars back to their proper types (active flag, integer turn number, optional Crown holder), JSON-decodes the player roster into PlayerState objects, and restores the in-progress pending_choices and monotonic _countdown_started so a turn interrupted by a restart can resume. Pure – no I/O of its own.

Called by get_or_restore_session() when reviving a session from the sg:game_session hash, and by the Redis persistence tests.

Parameters:

raw (dict) – The raw Redis hash mapping (bytes or str keyed) as returned by HGETALL.

Returns:

The restored session instance.

Return type:

GameSession

async classmethod load_from_redis(channel_id, redis)

Load a session from the legacy JSON Redis key, or return None.

Reads the game:session:{channel_id} string written by _save_to_redis() and rebuilds the session via from_dict(). Returns None when the key is absent, and on any Redis or decode error logs and returns None rather than raising. This is the older, JSON-string persistence path; the newer hash path lives under sg:game_session and is read by from_redis().

Called by the message processor’s game-restore flow, get_or_restore_session() (as a fallback), and load_by_game_id().

Parameters:

• channel_id (str) – Channel whose session should be loaded.

• redis (Any) – The async Redis client to read through.

Returns:

The restored session, or None if missing or unreadable.

Return type:

GameSession | None

async classmethod delete_from_redis(channel_id, redis)

Delete all Redis state for a channel’s session.

Issues a single DEL covering every key tied to the channel’s game: the legacy JSON session key game:session:{channel_id}, its turn history list game:session:{channel_id}:history, and the newer hash sg:game_session:{channel_id}. Redis errors are caught and logged so a failed wipe never propagates. Note the global game:index entry is not touched here.

Called by the Redis persistence tests; reachable as part of game-wipe and teardown flows.

Parameters:

• channel_id (str) – Channel whose session keys should be removed.

• redis (Any) – The async Redis client to delete through.

Return type:

None

async game_session.persist_session(channel_id, session, redis)

Persist a session as a Redis hash with a four-hour TTL.

Stores the flat GameSession.to_redis() mapping under sg:game_session:{channel_id} via HSET and applies a 14400-second (4 hour) expiry so abandoned sessions self-clean. This is the hash-form counterpart to the JSON key written by GameSession._save_to_redis(), and the source get_or_restore_session() prefers when reviving a session after a restart. A None client is a no-op; Redis errors are caught and logged.

Called by GameSession._save_to_redis() on every save, by get_or_restore_session() when migrating a legacy session forward, and by the Redis persistence tests.

Parameters:

• channel_id (str) – Channel the session belongs to; forms the hash key.

• session (GameSession) – The GameSession to serialize and store.

• redis (Any) – Async Redis client, or None to skip persistence.

Return type:

None

game_session.get_session(channel_id)

Look up the in-memory session for a channel, if one is registered.

A pure read of the per-process _active_sessions registry; it does not touch Redis, so it returns None for a session that exists only on disk after a restart – use get_or_restore_session() when a Redis fallback is needed.

Called by the message processor and the game UI components.

Parameters:

channel_id (str) – Channel to look up.

Returns:

The live session, or None if not in memory.

Return type:

GameSession | None

async game_session.get_or_restore_session(channel_id, redis)

Get session from memory, or restore from Redis after restart. # 💀🔥

After a bot reboot the in-memory dict is empty, but Redis still has the persisted session. This transparently restores it.

Return type:

GameSession | None

Parameters:

• channel_id (str)

• redis (Any)

game_session.set_session(channel_id, session)

Register (or replace) the in-memory session for a channel.

Inserts session into the per-process _active_sessions registry so later get_session() and get_or_restore_session() calls find it without a Redis round-trip. Pure in-memory mutation; does not persist – the caller is responsible for saving via the session’s own Redis methods.

Called by the message processor, the Discord platform adapters (discord.py and discord_self.py), the game_controls and hot_swap_game tools, and the Redis persistence tests.

Parameters:

• channel_id (str) – Channel to bind the session to.

• session (GameSession) – The GameSession to register.

Return type:

None

game_session.remove_session(channel_id)

Remove and return the in-memory session for a channel.

Pops channel_id from the per-process _active_sessions registry, returning the evicted GameSession or None if none was registered. Only clears the live reference; it does not delete any Redis state (use GameSession.delete_from_redis() for that).

Called by the message processor, the game UI components, and the exit_game, hot_swap_game, and wipe_game_data tools.

Parameters:

channel_id (str) – Channel whose session should be unregistered.

Returns:

The removed session, or None if absent.

Return type:

GameSession | None

async game_session.list_all_games(redis)

List every saved game recorded in the global game index.

Reads the whole game:index hash with HGETALL, decoding each entry (handling both bytes and str from the Redis client) into its metadata dict, stamping the game_id from the hash field, and returning the list sorted by game name. Returns an empty list when the index is empty, and on any Redis or decode error logs and returns an empty list rather than raising.

Called by the lobby modal, several game UI components, and the message processor’s game-listing flow.

Parameters:

redis (Any) – The async Redis client to read the index through.

Returns:

One metadata dict per saved game (each including its game_id), sorted by game_name.

Return type:

list[dict[str, Any]]

async game_session.load_by_game_id(game_id, redis)

Resolve a game_id to its channel and load that session.

Performs an HGET against the global game:index hash to find the saved game’s metadata, extracts its channel_id, and then delegates to GameSession.load_from_redis() to rebuild the full session from that channel’s stored state. Returns None when the id is unknown or carries no channel; any Redis or decode error is logged and yields None instead of raising.

Called by the message processor’s load-by-id flow.

Parameters:

• game_id (str) – The stable game identifier to resolve.

• redis (Any) – The async Redis client to read through.

Returns:

The loaded session, or None if the id is unknown or its session could not be restored.

Return type:

GameSession | None

class game_session.GameSessionLockRegistry

Bases: object

Per-process pool of per-session asyncio.Lock objects.

Hands out one lock per session_id so concurrent coroutines operating on the same game session can be serialized without blocking unrelated sessions. The lock pool is created lazily and guarded by an internal _pool_lock so pool mutations are themselves atomic across coroutines. State is purely in-memory and process-local – it does not coordinate across the bot’s microservices. Instantiated and exercised by tests/test_game_engine_remediation.py; no module-level singleton is created here.

__init__()

Create an empty per-process registry of per-session locks.

Sets up the lazily populated mapping of session id to asyncio.Lock, plus a single guard lock (_pool_lock) that makes mutations of that mapping atomic across coroutines. Holds no external state and performs no I/O. Instantiated by callers that need to serialize concurrent operations on the same session (e.g. the test_game_engine_remediation test); no module-level singleton is created here.

Return type:

None

async get_session_lock(session_id)

Return the lock for a session, creating it on first request.

Looks up session_id in the internal pool and mints a fresh asyncio.Lock if none exists yet, always returning the same lock object for a given id so callers can async with it to serialize operations on that session. The lookup-and-create runs under _pool_lock so it is atomic with respect to concurrent callers and to cleanup_session().

Exercised by tests/test_game_engine_remediation.py, which asserts repeated calls for the same id return one identity-stable lock while different ids get distinct locks.

Parameters:

session_id (str) – Identifier of the session to obtain a lock for.

Returns:

The shared lock instance for this session.

Return type:

Lock

async cleanup_session(session_id)

Drop the lock associated with a finished session.

Removes session_id from the internal lock pool so the registry does not leak asyncio.Lock objects for sessions that have ended. The delete is performed while holding _pool_lock so it is atomic with respect to concurrent get_session_lock() calls; a subsequent request for the same id will mint a brand-new lock. Exercised by the test_multiplayer_session_in_memory_lock_registry test, which asserts the post-cleanup lock is a distinct object.

Parameters:

session_id (str) – Identifier of the session whose lock should be discarded. A no-op if no lock is currently registered for it.

Return type:

None