""" DungeonMaster AI - Game State Management Minimal GameState stub for Phase 1 (MCP Integration). This provides the interface that tool wrappers need. Phase 4 will extend this with full implementation. """ from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import datetime from typing import Protocol, runtime_checkable @runtime_checkable class GameStateProtocol(Protocol): """ Protocol for game state access. This interface allows MCP tool wrappers to update game state without depending on the full GameState implementation. Phase 4 will provide the complete implementation. """ session_id: str in_combat: bool party: list[str] recent_events: list[dict[str, object]] def add_event( self, event_type: str, description: str, data: dict[str, object], ) -> None: """Add an event to recent events.""" ... def get_character(self, character_id: str) -> dict[str, object] | None: """Get character data from cache.""" ... def update_character_cache( self, character_id: str, data: dict[str, object], ) -> None: """Update character data in cache.""" ... def set_combat_state(self, combat_state: dict[str, object] | None) -> None: """Set or clear combat state.""" ... @dataclass class GameState: """ Minimal game state stub for Phase 1. Provides the basic interface that MCP tool wrappers need. Phase 4 will extend this with: - Full Pydantic models (GameState, CombatState, etc.) - GameStateManager class - State persistence (save/load) - Story context building """ # Core state session_id: str = field(default_factory=lambda: str(uuid.uuid4())) started_at: datetime = field(default_factory=datetime.now) system: str = "dnd5e" # Party management party: list[str] = field(default_factory=list) active_character_id: str | None = None # Location current_location: str = "Unknown" current_scene: dict[str, object] = field(default_factory=dict) # Combat in_combat: bool = False _combat_state: dict[str, object] | None = field(default=None, repr=False) # Events recent_events: list[dict[str, object]] = field(default_factory=list) turn_count: int = 0 # Caches _character_cache: dict[str, dict[str, object]] = field( default_factory=dict, repr=False, ) _known_npcs: dict[str, dict[str, object]] = field( default_factory=dict, repr=False, ) # Story tracking story_flags: dict[str, object] = field(default_factory=dict) current_adventure: str | None = None # Metadata last_updated: datetime = field(default_factory=datetime.now) # Configuration max_recent_events: int = field(default=20, repr=False) def add_event( self, event_type: str, description: str, data: dict[str, object], ) -> None: """ Add an event to recent events list. Keeps only the most recent events (default: 20). Args: event_type: Type of event (roll, combat, dialogue, etc.) description: Human-readable description data: Event-specific data """ event = { "type": event_type, "description": description, "data": data, "timestamp": datetime.now().isoformat(), "turn": self.turn_count, } self.recent_events.append(event) # Trim to max size if len(self.recent_events) > self.max_recent_events: self.recent_events = self.recent_events[-self.max_recent_events :] self.last_updated = datetime.now() def get_character(self, character_id: str) -> dict[str, object] | None: """ Get character data from cache. Args: character_id: Character ID to look up Returns: Character data dict or None if not cached """ return self._character_cache.get(character_id) def update_character_cache( self, character_id: str, data: dict[str, object], ) -> None: """ Update character data in cache. Args: character_id: Character ID data: Character data to cache """ self._character_cache[character_id] = data self.last_updated = datetime.now() def set_combat_state(self, combat_state: dict[str, object] | None) -> None: """ Set or clear combat state. Args: combat_state: Combat state dict, or None to clear """ self._combat_state = combat_state self.in_combat = combat_state is not None self.last_updated = datetime.now() # Log combat state change if combat_state is not None: self.add_event( event_type="combat_start", description="Combat has begun", data={"combatants": combat_state.get("turn_order", [])}, ) else: self.add_event( event_type="combat_end", description="Combat has ended", data={}, ) @property def combat_state(self) -> dict[str, object] | None: """Get current combat state.""" return self._combat_state def add_character_to_party(self, character_id: str) -> None: """ Add a character to the party. Args: character_id: Character ID to add """ if character_id not in self.party: self.party.append(character_id) # Set as active if first character if self.active_character_id is None: self.active_character_id = character_id self.last_updated = datetime.now() def remove_character_from_party(self, character_id: str) -> None: """ Remove a character from the party. Args: character_id: Character ID to remove """ if character_id in self.party: self.party.remove(character_id) # Clear active if removed if self.active_character_id == character_id: self.active_character_id = self.party[0] if self.party else None # Clear from cache self._character_cache.pop(character_id, None) self.last_updated = datetime.now() def set_location( self, location: str, scene: dict[str, object] | None = None, ) -> None: """ Update current location. Args: location: Location name/description scene: Optional scene details """ self.current_location = location if scene: self.current_scene = scene self.add_event( event_type="movement", description=f"Moved to {location}", data={"location": location, "scene": scene or {}}, ) self.last_updated = datetime.now() def increment_turn(self) -> int: """ Increment turn counter. Returns: New turn count """ self.turn_count += 1 self.last_updated = datetime.now() return self.turn_count def set_story_flag(self, flag: str, value: object) -> None: """ Set a story/quest flag. Args: flag: Flag name value: Flag value """ self.story_flags[flag] = value self.last_updated = datetime.now() def get_story_flag(self, flag: str, default: object = None) -> object: """ Get a story/quest flag. Args: flag: Flag name default: Default value if not set Returns: Flag value or default """ return self.story_flags.get(flag, default) def add_known_npc(self, npc_id: str, data: dict[str, object]) -> None: """ Add or update a known NPC. Args: npc_id: NPC identifier data: NPC data """ self._known_npcs[npc_id] = data self.last_updated = datetime.now() def get_known_npc(self, npc_id: str) -> dict[str, object] | None: """ Get known NPC data. Args: npc_id: NPC identifier Returns: NPC data or None """ return self._known_npcs.get(npc_id) def get_recent_events_by_type( self, event_type: str, limit: int = 10, ) -> list[dict[str, object]]: """ Get recent events filtered by type. Args: event_type: Event type to filter limit: Maximum events to return Returns: List of matching events """ matching = [e for e in self.recent_events if e.get("type") == event_type] return matching[-limit:] def to_summary(self) -> dict[str, object]: """ Create a summary dict for LLM context. Returns: Summary dict with key state information """ return { "session_id": self.session_id, "system": self.system, "turn_count": self.turn_count, "party_size": len(self.party), "active_character": self.active_character_id, "location": self.current_location, "in_combat": self.in_combat, "combat_round": ( self._combat_state.get("round") if self._combat_state else None ), "recent_events_count": len(self.recent_events), "adventure": self.current_adventure, } def reset(self) -> None: """Reset game state for new game.""" self.session_id = str(uuid.uuid4()) self.started_at = datetime.now() self.party.clear() self.active_character_id = None self.current_location = "Unknown" self.current_scene.clear() self.in_combat = False self._combat_state = None self.recent_events.clear() self.turn_count = 0 self._character_cache.clear() self._known_npcs.clear() self.story_flags.clear() self.current_adventure = None self.last_updated = datetime.now()