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.PlayerState(user_id, user_name, joined_turn=0)[source]

Bases: object

Per-player state within a game session.

Parameters:

  • user_id (str)

  • user_name (str)

  • joined_turn (int)

user_id: str
user_name: str
joined_turn: int = 0
to_dict()[source]
Return type:

dict[str, Any]

classmethod from_dict(d)[source]
Return type:

PlayerState

Parameters:

d (dict[str, Any])

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

Bases: object

A single turn in the game history.

Parameters:
turn: int
choices: dict[str, str]
narrative_summary: str = ''
timestamp: float
to_dict()[source]
Return type:

dict[str, Any]

classmethod from_dict(d)[source]
Return type:

TurnRecord

Parameters:

d (dict[str, Any])

class game_session.GameSession(channel_id, game_name='', game_id=None)[source]

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)

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
async boot(game_name, redis=None)[source]

Initialize a new game session.

Returns the boot message including title screen URL.

Return type:

str

Parameters:

  • game_name (str)

  • redis (Any | None)

async exit_game(redis=None)[source]

Save state and unload the session.

Return type:

str

Parameters:

redis (Any | None)

register_player(user_id, user_name)[source]

Ensure a player is tracked in this session.

Return type:

None

Parameters:

  • user_id (str)

  • user_name (str)

async submit_choice(user_id, user_name, choice)[source]

Register a player’s button press.

Returns (is_first_choice, seconds_remaining). If this is the first choice, the caller should start the countdown.

Return type:

tuple[bool, float]

Parameters:
async wait_for_countdown()[source]

Wait for the countdown to expire, then return collected choices.

Advances the turn number and clears pending choices.

Return type:

dict[str, str]

format_choices_as_input(choices)[source]

Format collected choices into a synthetic user message.

This gets fed back to the LLM as the next turn’s input.

Return type:

str

Parameters:

choices (dict[str, str])

async record_turn(choices, narrative_summary, redis=None)[source]

Record a completed turn in history.

Return type:

None

Parameters:
async get_turn_history(redis=None, count=10)[source]

Retrieve recent turn history.

Return type:

list[TurnRecord]

Parameters:

  • redis (Any | None)

  • count (int)

to_dict()[source]

Serialize session state to a dict.

Return type:

dict[str, Any]

classmethod from_dict(d)[source]

Restore a session from a serialized dict.

Return type:

GameSession

Parameters:

d (dict[str, Any])

async classmethod load_from_redis(channel_id, redis)[source]

Load a session from Redis, or return None.

Return type:

GameSession | None

Parameters:

  • channel_id (str)

  • redis (Any)

async classmethod delete_from_redis(channel_id, redis)[source]

Remove a session from Redis.

Return type:

None

Parameters:

  • channel_id (str)

  • redis (Any)

game_session.get_session(channel_id)[source]

Get the active game session for a channel, if any.

Return type:

GameSession | None

Parameters:

channel_id (str)

game_session.set_session(channel_id, session)[source]

Register a game session for a channel.

Return type:

None

Parameters:
game_session.remove_session(channel_id)[source]

Unregister and return the session for a channel.

Return type:

GameSession | None

Parameters:

channel_id (str)

async game_session.list_all_games(redis)[source]

List all saved games from the global index.

Return type:

list[dict[str, Any]]

Parameters:

redis (Any)

async game_session.load_by_game_id(game_id, redis)[source]

Load a game by its ID from the index.

Return type:

GameSession | None

Parameters: