|
|
""" |
|
|
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 |
|
|
""" |
|
|
|
|
|
|
|
|
session_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
|
|
started_at: datetime = field(default_factory=datetime.now) |
|
|
system: str = "dnd5e" |
|
|
|
|
|
|
|
|
party: list[str] = field(default_factory=list) |
|
|
active_character_id: str | None = None |
|
|
|
|
|
|
|
|
current_location: str = "Unknown" |
|
|
current_scene: dict[str, object] = field(default_factory=dict) |
|
|
|
|
|
|
|
|
in_combat: bool = False |
|
|
_combat_state: dict[str, object] | None = field(default=None, repr=False) |
|
|
|
|
|
|
|
|
recent_events: list[dict[str, object]] = field(default_factory=list) |
|
|
turn_count: int = 0 |
|
|
|
|
|
|
|
|
_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_flags: dict[str, object] = field(default_factory=dict) |
|
|
current_adventure: str | None = None |
|
|
|
|
|
|
|
|
last_updated: datetime = field(default_factory=datetime.now) |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
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() |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
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) |
|
|
|
|
|
if self.active_character_id == character_id: |
|
|
self.active_character_id = self.party[0] if self.party else None |
|
|
|
|
|
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() |
|
|
|