""" DungeonMaster AI - Agent Models Pydantic models for agent responses, turn results, and game interactions. """ from __future__ import annotations from datetime import datetime from enum import Enum from pydantic import BaseModel, Field from src.mcp_integration.models import ( CombatStateResult, DiceRollResult, ) from src.voice.models import VoiceType # ============================================================================= # Enums # ============================================================================= class GameMode(str, Enum): """Game modes for DM agent context switching.""" EXPLORATION = "exploration" COMBAT = "combat" SOCIAL = "social" NARRATIVE = "narrative" class DegradationLevel(str, Enum): """System degradation levels for graceful fallbacks.""" FULL = "full" # All features working PARTIAL = "partial" # Some tools unavailable TEXT_ONLY = "text_only" # Voice unavailable MINIMAL = "minimal" # Fallback dice only, limited LLM class SpecialMomentType(str, Enum): """Types of special dramatic moments.""" CRITICAL_HIT = "critical_hit" CRITICAL_MISS = "critical_miss" DEATH_SAVE_SUCCESS = "death_save_success" DEATH_SAVE_FAILURE = "death_save_failure" DEATH_SAVE_NAT_20 = "death_save_nat_20" DEATH_SAVE_NAT_1 = "death_save_nat_1" KILLING_BLOW = "killing_blow" PLAYER_DEATH = "player_death" LEVEL_UP = "level_up" COMBAT_START = "combat_start" COMBAT_VICTORY = "combat_victory" class PacingStyle(str, Enum): """Response pacing styles.""" VERBOSE = "verbose" # New locations, world-building (3-5 sentences) STANDARD = "standard" # Normal gameplay (2-3 sentences) QUICK = "quick" # Combat actions (1-2 sentences) DRAMATIC = "dramatic" # Critical moments (short, punchy) # ============================================================================= # Voice Segment Models # ============================================================================= class VoiceSegment(BaseModel): """A segment of text with assigned voice profile.""" text: str = Field(description="Text content to synthesize") voice_type: VoiceType = Field( default=VoiceType.DM, description="Voice profile to use", ) pause_before_ms: int = Field( default=0, ge=0, description="Pause duration before this segment in milliseconds", ) pause_after_ms: int = Field( default=0, ge=0, description="Pause duration after this segment in milliseconds", ) emphasis: bool = Field( default=False, description="Whether to emphasize this segment", ) is_dialogue: bool = Field( default=False, description="Whether this is NPC/character dialogue", ) speaker_name: str | None = Field( default=None, description="Name of the speaker if dialogue", ) # ============================================================================= # Tool Call Models # ============================================================================= class ToolCallInfo(BaseModel): """Information about a tool call made during processing.""" tool_name: str = Field(description="Name of the tool called") arguments: dict[str, object] = Field( default_factory=dict, description="Arguments passed to the tool", ) result: object = Field( default=None, description="Result returned by the tool", ) success: bool = Field( default=True, description="Whether the tool call succeeded", ) error_message: str | None = Field( default=None, description="Error message if tool call failed", ) duration_ms: float = Field( default=0.0, description="Time taken by the tool call", ) timestamp: datetime = Field( default_factory=datetime.now, description="When the tool was called", ) # ============================================================================= # Special Moment Models # ============================================================================= class SpecialMoment(BaseModel): """A dramatic moment in the game requiring special handling.""" moment_type: SpecialMomentType = Field(description="Type of special moment") enhanced_narration: str = Field( default="", description="Dramatic narration for this moment", ) voice_type: VoiceType = Field( default=VoiceType.DM, description="Voice to use for narration", ) ui_effects: list[str] = Field( default_factory=list, description="UI effects to trigger (e.g., 'screen_shake', 'golden_glow')", ) pause_before_ms: int = Field( default=500, description="Dramatic pause before narration", ) sound_effect: str | None = Field( default=None, description="Optional sound effect to play", ) context: dict[str, object] = Field( default_factory=dict, description="Additional context (roll value, weapon, etc.)", ) # ============================================================================= # DM Response Models # ============================================================================= class DMResponse(BaseModel): """Response from the Dungeon Master agent.""" narration: str = Field( description="The narrative text response from the DM", ) voice_segments: list[VoiceSegment] = Field( default_factory=list, description="Text segments with voice assignments for multi-voice synthesis", ) tool_calls: list[ToolCallInfo] = Field( default_factory=list, description="Tools that were called during processing", ) dice_rolls: list[DiceRollResult] = Field( default_factory=list, description="Dice rolls that occurred", ) game_state_updates: dict[str, object] = Field( default_factory=dict, description="State changes to apply (hp_changes, location, combat, etc.)", ) special_moment: SpecialMoment | None = Field( default=None, description="Special dramatic moment if detected", ) ui_effects: list[str] = Field( default_factory=list, description="UI effects to trigger", ) game_mode: GameMode = Field( default=GameMode.EXPLORATION, description="Current game mode after processing", ) pacing: PacingStyle = Field( default=PacingStyle.STANDARD, description="Pacing style used for response", ) # Metadata processing_time_ms: float = Field( default=0.0, description="Total processing time", ) llm_provider: str = Field( default="", description="Which LLM provider was used", ) timestamp: datetime = Field( default_factory=datetime.now, description="When response was generated", ) # ============================================================================= # Rules Response Models # ============================================================================= class RulesResponse(BaseModel): """Response from the Rules Arbiter agent.""" answer: str = Field(description="The rules answer/explanation") sources: list[str] = Field( default_factory=list, description="Sources/citations for the rule", ) confidence: float = Field( default=1.0, ge=0.0, le=1.0, description="Confidence in the answer", ) tool_calls: list[ToolCallInfo] = Field( default_factory=list, description="Tools called to find the answer", ) from_cache: bool = Field( default=False, description="Whether result came from cache", ) # ============================================================================= # Turn Result Models # ============================================================================= class TurnResult(BaseModel): """Complete result of processing a player turn.""" # Core response narration_text: str = Field( description="The narrative text to display", ) narration_audio: bytes | None = Field( default=None, description="Synthesized audio (combined from all segments)", ) # Detailed information voice_segments: list[VoiceSegment] = Field( default_factory=list, description="Individual voice segments for multi-voice", ) tool_calls: list[ToolCallInfo] = Field( default_factory=list, description="All tools that were called", ) dice_rolls: list[DiceRollResult] = Field( default_factory=list, description="All dice rolls that occurred", ) # Game state game_mode: GameMode = Field( default=GameMode.EXPLORATION, description="Current game mode", ) combat_updates: CombatStateResult | None = Field( default=None, description="Combat state changes if in combat", ) # Special handling special_moment: SpecialMoment | None = Field( default=None, description="Special dramatic moment if any", ) ui_effects: list[str] = Field( default_factory=list, description="UI effects to trigger", ) # Status degradation_level: DegradationLevel = Field( default=DegradationLevel.FULL, description="System degradation level", ) voice_enabled: bool = Field( default=True, description="Whether voice was generated", ) error_message: str | None = Field( default=None, description="Error message if something went wrong", ) # Metadata turn_number: int = Field( default=0, description="Current turn number", ) processing_time_ms: float = Field( default=0.0, description="Total processing time", ) llm_provider: str = Field( default="", description="Which LLM was used", ) timestamp: datetime = Field( default_factory=datetime.now, description="When turn was processed", ) # ============================================================================= # Streaming Models # ============================================================================= class StreamChunk(BaseModel): """A chunk of streaming response.""" text: str = Field( default="", description="Incremental text content", ) is_complete: bool = Field( default=False, description="Whether this is the final chunk", ) tool_call: ToolCallInfo | None = Field( default=None, description="Tool call if one occurred in this chunk", ) accumulated_text: str = Field( default="", description="Full text accumulated so far", ) # ============================================================================= # Game Context Models # ============================================================================= class GameContext(BaseModel): """Context information passed to agents for decision making.""" # Core state session_id: str = Field(description="Current session ID") turn_number: int = Field(default=0, description="Current turn number") game_mode: GameMode = Field( default=GameMode.EXPLORATION, description="Current game mode", ) # Location current_location: str = Field( default="Unknown", description="Current location name", ) location_description: str = Field( default="", description="Description of current location", ) # Party active_character_id: str | None = Field( default=None, description="Currently controlled character", ) party_summary: str = Field( default="", description="Summary of party status (HP, conditions)", ) # Combat in_combat: bool = Field( default=False, description="Whether combat is active", ) combat_round: int | None = Field( default=None, description="Current combat round if in combat", ) current_combatant: str | None = Field( default=None, description="Whose turn it is in combat", ) # NPCs current_npc: dict[str, object] | None = Field( default=None, description="Current NPC being interacted with", ) npcs_present: list[str] = Field( default_factory=list, description="NPCs in current scene", ) # Recent history recent_events_summary: str = Field( default="", description="Summary of recent events for context", ) # Adventure adventure_name: str | None = Field( default=None, description="Name of current adventure", ) story_flags: dict[str, object] = Field( default_factory=dict, description="Active story/quest flags", ) # ============================================================================= # LLM Provider Models # ============================================================================= class LLMProviderHealth(BaseModel): """Health status of an LLM provider.""" provider_name: str = Field(description="Name of the provider") is_available: bool = Field( default=False, description="Whether provider is available", ) is_primary: bool = Field( default=False, description="Whether this is the primary provider", ) consecutive_failures: int = Field( default=0, description="Number of consecutive failures", ) last_success: datetime | None = Field( default=None, description="Last successful call", ) last_error: str | None = Field( default=None, description="Last error message", ) circuit_open: bool = Field( default=False, description="Whether circuit breaker is open", ) class LLMResponse(BaseModel): """Response from LLM provider chain.""" text: str = Field(description="Generated text") tool_calls: list[dict[str, object]] = Field( default_factory=list, description="Tool calls requested by LLM", ) provider_used: str = Field( default="", description="Which provider generated this response", ) model_used: str = Field( default="", description="Which model was used", ) input_tokens: int = Field( default=0, description="Input token count", ) output_tokens: int = Field( default=0, description="Output token count", ) latency_ms: float = Field( default=0.0, description="Response latency", ) from_fallback: bool = Field( default=False, description="Whether fallback provider was used", )