mantimon-tcg/backend/app/core/models/game_state.py
Cal Corum 0c810e5b30 Add Phase 4 WebSocket infrastructure (WS-001 through GS-001)
WebSocket Message Schemas (WS-002):
- Add Pydantic models for all client/server WebSocket messages
- Implement discriminated unions for message type parsing
- Include JoinGame, Action, Resign, Heartbeat client messages
- Include GameState, ActionResult, Error, TurnStart server messages

Connection Manager (WS-003):
- Add Redis-backed WebSocket connection tracking
- Implement user-to-sid mapping with TTL management
- Support game room association and opponent lookup
- Add heartbeat tracking for connection health

Socket.IO Authentication (WS-004):
- Add JWT-based authentication middleware
- Support token extraction from multiple formats
- Implement session setup with ConnectionManager integration
- Add require_auth helper for event handlers

Socket.IO Server Setup (WS-001):
- Configure AsyncServer with ASGI mode
- Register /game namespace with event handlers
- Integrate with FastAPI via ASGIApp wrapper
- Configure CORS from application settings

Game Service (GS-001):
- Add stateless GameService for game lifecycle orchestration
- Create engine per-operation using rules from GameState
- Implement action-based RNG seeding for deterministic replay
- Add rng_seed field to GameState for replay support

Architecture verified:
- Core module independence (no forbidden imports)
- Config from request pattern (rules in GameState)
- Dependency injection (constructor deps, method config)
- All 1090 tests passing

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-28 22:21:20 -06:00

621 lines
23 KiB
Python

"""Game state models for the Mantimon TCG game engine.
This module defines the complete game state hierarchy:
- Zone: A collection of cards (deck, hand, bench, etc.) with operations
- PlayerState: All zones and state for a single player
- GameState: Complete game state including both players, rules, and turn tracking
The GameState is designed to be self-contained - it includes the card registry
so that the game can be serialized/deserialized without external dependencies.
This supports both live multiplayer and offline/standalone play.
Usage:
from app.core.config import RulesConfig
game = GameState(
game_id="match-123",
rules=RulesConfig(),
card_registry={card.id: card for card in cards},
players={
"player1": PlayerState(player_id="player1"),
"player2": PlayerState(player_id="player2"),
},
)
# Access player zones
player = game.players["player1"]
card = player.hand.draw()
player.deck.shuffle(rng)
"""
from typing import Any
from pydantic import BaseModel, Field
from app.core.config import RulesConfig
from app.core.enums import GameEndReason, TurnPhase
from app.core.models.card import CardDefinition, CardInstance
from app.core.rng import RandomProvider
class ForcedAction(BaseModel):
"""Represents an action a player must take before the game can proceed.
When a forced action is set, only the specified player can act, and only
with the specified action type. This is used for situations like:
- Selecting a new active Pokemon after a knockout
- Selecting prize cards to take
- Discarding cards when required by an effect
Attributes:
player_id: The player who must take the action.
action_type: The type of action required (e.g., "select_active", "select_prize").
reason: Human-readable explanation of why this action is required.
params: Additional parameters for the action (e.g., {"count": 2} for "discard 2 cards").
"""
player_id: str
action_type: str
reason: str
params: dict[str, Any] = Field(default_factory=dict)
class Zone(BaseModel):
"""A collection of cards representing a game zone.
Zones are the fundamental building blocks of game state. Each zone
(deck, hand, bench, etc.) is a Zone instance with appropriate operations.
The zone stores CardInstance objects by their instance_id. The actual
CardDefinition data is looked up from the GameState.card_registry.
Attributes:
cards: List of CardInstance objects in this zone.
zone_type: Identifier for this zone type (for debugging/logging).
"""
cards: list[CardInstance] = Field(default_factory=list)
zone_type: str = "generic"
def __len__(self) -> int:
"""Return the number of cards in this zone."""
return len(self.cards)
def __bool__(self) -> bool:
"""Return True if the zone has any cards."""
return len(self.cards) > 0
def __contains__(self, instance_id: str) -> bool:
"""Check if a card with the given instance_id is in this zone."""
return any(card.instance_id == instance_id for card in self.cards)
def is_empty(self) -> bool:
"""Check if the zone has no cards."""
return len(self.cards) == 0
def add(self, card: CardInstance) -> None:
"""Add a card to this zone (appends to the end)."""
self.cards.append(card)
def add_to_top(self, card: CardInstance) -> None:
"""Add a card to the top of this zone (index 0, like top of deck)."""
self.cards.insert(0, card)
def add_to_bottom(self, card: CardInstance) -> None:
"""Add a card to the bottom of this zone (end of list)."""
self.cards.append(card)
def remove(self, instance_id: str) -> CardInstance | None:
"""Remove and return a card by instance_id, or None if not found."""
for i, card in enumerate(self.cards):
if card.instance_id == instance_id:
return self.cards.pop(i)
return None
def get(self, instance_id: str) -> CardInstance | None:
"""Get a card by instance_id without removing it."""
for card in self.cards:
if card.instance_id == instance_id:
return card
return None
def draw(self) -> CardInstance | None:
"""Remove and return the top card (index 0), or None if empty.
This is the standard "draw from deck" operation.
"""
if self.cards:
return self.cards.pop(0)
return None
def draw_bottom(self) -> CardInstance | None:
"""Remove and return the bottom card, or None if empty."""
if self.cards:
return self.cards.pop()
return None
def peek(self, count: int = 1) -> list[CardInstance]:
"""Look at the top N cards without removing them.
Args:
count: Number of cards to peek at.
Returns:
List of cards from the top (may be fewer if zone has less).
"""
return self.cards[:count]
def peek_bottom(self, count: int = 1) -> list[CardInstance]:
"""Look at the bottom N cards without removing them."""
if count >= len(self.cards):
return list(self.cards)
return self.cards[-count:]
def shuffle(self, rng: RandomProvider) -> None:
"""Shuffle the cards in this zone using the provided RNG.
Args:
rng: RandomProvider instance for deterministic shuffling.
"""
rng.shuffle(self.cards)
def clear(self) -> list[CardInstance]:
"""Remove and return all cards from this zone."""
cards = self.cards.copy()
self.cards.clear()
return cards
def get_all(self) -> list[CardInstance]:
"""Return a copy of all cards in this zone."""
return list(self.cards)
def find_by_definition(self, definition_id: str) -> list[CardInstance]:
"""Find all cards with a specific definition_id."""
return [card for card in self.cards if card.definition_id == definition_id]
def count(self) -> int:
"""Return the number of cards (alias for __len__)."""
return len(self.cards)
class PlayerState(BaseModel):
"""Complete state for a single player in a game.
Contains all zones (deck, hand, active, bench, etc.) and per-player
state like score, turn counters, and action tracking.
Attributes:
player_id: Unique identifier for this player.
deck: The player's draw deck.
hand: Cards in the player's hand.
active: The active Pokemon zone (0 or 1 card).
bench: Benched Pokemon (up to max_bench_size from rules).
discard: Discard pile.
prizes: Prize cards (hidden until taken, in point-based mode this is unused).
energy_deck: Separate energy deck (Pokemon Pocket style).
score: Points scored (knockouts).
energy_attachments_this_turn: Number of energy cards attached this turn.
supporters_played_this_turn: Number of Supporter cards played this turn.
stadiums_played_this_turn: Number of Stadium cards played this turn.
items_played_this_turn: Number of Item cards played this turn.
retreats_this_turn: Number of retreats performed this turn.
gx_attack_used: Whether this player has used their GX attack (once per game).
vstar_power_used: Whether this player has used their VSTAR power (once per game).
"""
player_id: str
# Zones
deck: Zone = Field(default_factory=lambda: Zone(zone_type="deck"))
hand: Zone = Field(default_factory=lambda: Zone(zone_type="hand"))
active: Zone = Field(default_factory=lambda: Zone(zone_type="active"))
bench: Zone = Field(default_factory=lambda: Zone(zone_type="bench"))
discard: Zone = Field(default_factory=lambda: Zone(zone_type="discard"))
prizes: Zone = Field(default_factory=lambda: Zone(zone_type="prizes"))
energy_deck: Zone = Field(default_factory=lambda: Zone(zone_type="energy_deck"))
energy_zone: Zone = Field(default_factory=lambda: Zone(zone_type="energy_zone"))
# Score tracking (point-based system)
score: int = 0
# Per-turn action counters (reset at turn start)
# These are integers to support configurable rules (e.g., 2 energy per turn)
energy_attachments_this_turn: int = 0
supporters_played_this_turn: int = 0
stadiums_played_this_turn: int = 0
items_played_this_turn: int = 0
retreats_this_turn: int = 0
# Per-game flags
gx_attack_used: bool = False
vstar_power_used: bool = False
def get_active_pokemon(self) -> CardInstance | None:
"""Get the active Pokemon, or None if no active."""
if self.active.cards:
return self.active.cards[0]
return None
def has_active_pokemon(self) -> bool:
"""Check if this player has an active Pokemon."""
return len(self.active) > 0
def has_benched_pokemon(self) -> bool:
"""Check if this player has any benched Pokemon."""
return len(self.bench) > 0
def has_pokemon_in_play(self) -> bool:
"""Check if this player has any Pokemon in play (active or bench)."""
return self.has_active_pokemon() or self.has_benched_pokemon()
def get_all_pokemon_in_play(self) -> list[CardInstance]:
"""Get all Pokemon in play (active + bench)."""
pokemon = []
if self.active.cards:
pokemon.extend(self.active.cards)
pokemon.extend(self.bench.cards)
return pokemon
def can_attach_energy(self, rules: RulesConfig) -> bool:
"""Check if this player can attach energy based on rules and turn state.
Compares the number of energy attachments made this turn against the
rules.energy.attachments_per_turn limit.
Args:
rules: The RulesConfig governing this game.
Returns:
True if more energy can be attached this turn.
"""
return self.energy_attachments_this_turn < rules.energy.attachments_per_turn
def can_play_supporter(self, rules: RulesConfig) -> bool:
"""Check if this player can play a Supporter card.
Compares the number of Supporters played this turn against the
rules.trainer.supporters_per_turn limit.
Args:
rules: The RulesConfig governing this game.
Returns:
True if more Supporters can be played this turn.
"""
return self.supporters_played_this_turn < rules.trainer.supporters_per_turn
def can_play_stadium(self, rules: RulesConfig) -> bool:
"""Check if this player can play a Stadium card.
Compares the number of Stadiums played this turn against the
rules.trainer.stadiums_per_turn limit.
Args:
rules: The RulesConfig governing this game.
Returns:
True if more Stadiums can be played this turn.
"""
return self.stadiums_played_this_turn < rules.trainer.stadiums_per_turn
def can_play_item(self, rules: RulesConfig) -> bool:
"""Check if this player can play an Item card.
Compares the number of Items played this turn against the
rules.trainer.items_per_turn limit. If items_per_turn is None,
unlimited Items can be played.
Args:
rules: The RulesConfig governing this game.
Returns:
True if more Items can be played this turn.
"""
if rules.trainer.items_per_turn is None:
return True # Unlimited items
return self.items_played_this_turn < rules.trainer.items_per_turn
def can_retreat(self, rules: RulesConfig) -> bool:
"""Check if this player can retreat based on rules and turn state.
Compares the number of retreats performed this turn against the
rules.retreat.retreats_per_turn limit.
Args:
rules: The RulesConfig governing this game.
Returns:
True if more retreats are allowed this turn.
"""
return self.retreats_this_turn < rules.retreat.retreats_per_turn
def bench_space_available(self, rules: RulesConfig) -> int:
"""Return how many more Pokemon can be placed on the bench."""
return max(0, rules.bench.max_size - len(self.bench))
def can_bench_pokemon(self, rules: RulesConfig) -> bool:
"""Check if there's room on the bench for another Pokemon."""
return self.bench_space_available(rules) > 0
def reset_turn_state(self) -> None:
"""Reset all per-turn counters. Called at the start of each turn.
Resets all action counters (energy attachments, trainer plays, retreats)
to zero, and resets ability usage on all Pokemon in play.
"""
self.energy_attachments_this_turn = 0
self.supporters_played_this_turn = 0
self.stadiums_played_this_turn = 0
self.items_played_this_turn = 0
self.retreats_this_turn = 0
# Also reset ability usage on all Pokemon in play
for pokemon in self.get_all_pokemon_in_play():
pokemon.reset_turn_state()
class GameState(BaseModel):
"""Complete state of a game in progress.
This is the top-level model containing everything needed to represent
and resume a game. It's designed to be self-contained - the card_registry
holds all CardDefinitions used in this game, so the state can be
serialized and loaded without external dependencies.
Attributes:
game_id: Unique identifier for this game.
rules: The RulesConfig governing this game.
card_registry: Mapping of card definition IDs to CardDefinition objects.
players: Mapping of player IDs to PlayerState objects.
current_player_id: ID of the player whose turn it is.
turn_number: Current turn number (starts at 1).
phase: Current phase of the turn.
winner_id: ID of the winning player, if game has ended.
end_reason: Why the game ended, if it has ended.
stadium_in_play: The current Stadium card in play, if any.
stadium_owner_id: Player ID of who played the current stadium (for discard).
turn_order: List of player IDs in turn order.
first_turn_completed: Whether the very first turn of the game is done.
forced_actions: Queue of ForcedAction items that must be completed before game proceeds.
Actions are processed in FIFO order (first added = first to resolve).
action_log: Log of actions taken (for replays/debugging).
rng_seed: Optional seed for deterministic RNG. When set, enables replay capability.
"""
game_id: str
rules: RulesConfig = Field(default_factory=RulesConfig)
card_registry: dict[str, CardDefinition] = Field(default_factory=dict)
players: dict[str, PlayerState] = Field(default_factory=dict)
# Turn tracking
current_player_id: str = ""
turn_number: int = 0
phase: TurnPhase = TurnPhase.SETUP
# Game end state
winner_id: str | None = None
end_reason: GameEndReason | None = None
# Shared game state
stadium_in_play: CardInstance | None = None
stadium_owner_id: str | None = None
# Turn order (for 2+ player support)
turn_order: list[str] = Field(default_factory=list)
# First turn tracking
first_turn_completed: bool = False
# Forced actions queue (e.g., select new active after KO, select prizes)
# Actions are processed in FIFO order - first added is first to resolve
forced_actions: list[ForcedAction] = Field(default_factory=list)
# Optional action log for replays
action_log: list[dict[str, Any]] = Field(default_factory=list)
# Optional RNG seed for deterministic replays
rng_seed: int | None = None
def get_current_player(self) -> PlayerState:
"""Get the PlayerState for the current player.
Raises:
KeyError: If current_player_id is not in players dict.
"""
return self.players[self.current_player_id]
def get_opponent_id(self, player_id: str) -> str:
"""Get the opponent's player ID (assumes 2-player game).
Args:
player_id: The player whose opponent we want.
Returns:
The opponent's player ID.
Raises:
ValueError: If player_id is not in the game or game has != 2 players.
"""
if len(self.players) != 2:
raise ValueError("get_opponent_id only works for 2-player games")
if player_id not in self.players:
raise ValueError(f"Player {player_id} not found in game")
for pid in self.players:
if pid != player_id:
return pid
# This should be unreachable with 2 players where one is player_id
raise ValueError(f"Could not find opponent for {player_id}")
def get_opponent(self, player_id: str) -> PlayerState:
"""Get the PlayerState for a player's opponent (assumes 2-player game)."""
return self.players[self.get_opponent_id(player_id)]
def get_card_definition(self, definition_id: str) -> CardDefinition | None:
"""Look up a CardDefinition from the registry."""
return self.card_registry.get(definition_id)
def is_player_turn(self, player_id: str) -> bool:
"""Check if it's the specified player's turn."""
return self.current_player_id == player_id
def is_first_turn(self) -> bool:
"""Check if this is the very first turn of the game.
The first turn is turn 1, before first_turn_completed is set True.
"""
return self.turn_number == 1 and not self.first_turn_completed
def is_game_over(self) -> bool:
"""Check if the game has ended."""
return self.winner_id is not None
# =========================================================================
# Forced Action Queue Management
# =========================================================================
def has_forced_action(self) -> bool:
"""Check if there are any pending forced actions."""
return len(self.forced_actions) > 0
def get_current_forced_action(self) -> ForcedAction | None:
"""Get the current (first) forced action without removing it.
Returns:
The first forced action in the queue, or None if queue is empty.
"""
if self.forced_actions:
return self.forced_actions[0]
return None
def add_forced_action(self, action: ForcedAction) -> None:
"""Add a forced action to the queue.
Actions are processed in FIFO order, so this adds to the end.
Args:
action: The ForcedAction to add.
"""
self.forced_actions.append(action)
def pop_forced_action(self) -> ForcedAction | None:
"""Remove and return the current (first) forced action.
Returns:
The first forced action that was removed, or None if queue was empty.
"""
if self.forced_actions:
return self.forced_actions.pop(0)
return None
def clear_forced_actions(self) -> None:
"""Clear all pending forced actions."""
self.forced_actions.clear()
def get_player_count(self) -> int:
"""Return the number of players in this game."""
return len(self.players)
def advance_turn(self) -> None:
"""Advance to the next player's turn.
This handles:
- Setting first_turn_completed after turn 1
- Incrementing turn_number
- Switching current_player_id to the next player
- Resetting the new player's turn flags
- Setting phase to DRAW
"""
# Mark first turn as completed if we're ending turn 1
if self.turn_number == 1:
self.first_turn_completed = True
# Find next player in turn order
if self.turn_order:
current_index = self.turn_order.index(self.current_player_id)
next_index = (current_index + 1) % len(self.turn_order)
self.current_player_id = self.turn_order[next_index]
# Only increment turn number when we wrap back to first player
if next_index == 0:
self.turn_number += 1
else:
# Fallback for games without explicit turn order
self.turn_number += 1
# Reset the new player's turn state
self.get_current_player().reset_turn_state()
# Start at draw phase
self.phase = TurnPhase.DRAW
def set_winner(self, player_id: str, reason: GameEndReason) -> None:
"""Set the winner and end the game.
Args:
player_id: The winning player's ID.
reason: Why the game ended.
"""
self.winner_id = player_id
self.end_reason = reason
def log_action(self, action: dict[str, Any]) -> None:
"""Add an action to the action log for replay support."""
self.action_log.append(action)
def get_all_cards_in_play(self) -> list[CardInstance]:
"""Get all cards currently in play (all players' active and bench)."""
cards = []
for player in self.players.values():
cards.extend(player.get_all_pokemon_in_play())
return cards
def find_card_instance(self, instance_id: str) -> tuple[CardInstance | None, str | None]:
"""Find a CardInstance anywhere in the game.
Searches all zones of all players for the card, including cards attached
to Pokemon (energy, tools) and cards underneath evolved Pokemon.
Args:
instance_id: The instance_id to search for.
Returns:
Tuple of (CardInstance, zone_type) if found, (None, None) if not.
Zone types include standard zones plus:
- "attached_energy": Energy attached to a Pokemon in play
- "attached_tools": Tools attached to a Pokemon in play
- "cards_underneath": Cards in an evolution stack
"""
for player in self.players.values():
# Search standard zones
for zone_name in [
"deck",
"hand",
"active",
"bench",
"discard",
"prizes",
"energy_deck",
"energy_zone",
]:
zone: Zone = getattr(player, zone_name)
card = zone.get(instance_id)
if card:
return card, zone_name
# Search cards attached to Pokemon in play (active and bench)
for pokemon in player.get_all_pokemon_in_play():
# Check attached energy
for energy in pokemon.attached_energy:
if energy.instance_id == instance_id:
return energy, "attached_energy"
# Check attached tools
for tool in pokemon.attached_tools:
if tool.instance_id == instance_id:
return tool, "attached_tools"
# Check evolution stack (cards underneath)
for card in pokemon.cards_underneath:
if card.instance_id == instance_id:
return card, "cards_underneath"
return None, None