""" DungeonMaster AI - Agent Exceptions Exception hierarchy for agent-related errors. """ from __future__ import annotations class AgentError(Exception): """Base exception for all agent-related errors.""" def __init__(self, message: str, recoverable: bool = True) -> None: super().__init__(message) self.message = message self.recoverable = recoverable # ============================================================================= # LLM Provider Exceptions # ============================================================================= class LLMProviderError(AgentError): """Base exception for LLM provider errors.""" def __init__( self, message: str, provider: str = "", recoverable: bool = True, ) -> None: super().__init__(message, recoverable) self.provider = provider class LLMRateLimitError(LLMProviderError): """LLM provider rate limit exceeded.""" def __init__( self, provider: str, retry_after: float | None = None, ) -> None: super().__init__( f"Rate limit exceeded for {provider}", provider=provider, recoverable=True, ) self.retry_after = retry_after class LLMTimeoutError(LLMProviderError): """LLM request timed out.""" def __init__( self, provider: str, timeout_seconds: float, ) -> None: super().__init__( f"Request to {provider} timed out after {timeout_seconds}s", provider=provider, recoverable=True, ) self.timeout_seconds = timeout_seconds class LLMAuthenticationError(LLMProviderError): """LLM authentication failed (invalid API key).""" def __init__(self, provider: str) -> None: super().__init__( f"Authentication failed for {provider}. Check API key.", provider=provider, recoverable=False, ) class LLMQuotaExhaustedError(LLMProviderError): """LLM quota/credits exhausted.""" def __init__(self, provider: str) -> None: super().__init__( f"Quota exhausted for {provider}", provider=provider, recoverable=False, ) class LLMAllProvidersFailedError(AgentError): """All LLM providers failed.""" def __init__(self, errors: dict[str, str]) -> None: providers = ", ".join(errors.keys()) super().__init__( f"All LLM providers failed: {providers}", recoverable=False, ) self.errors = errors class LLMCircuitBreakerOpenError(LLMProviderError): """Circuit breaker is open for this provider.""" def __init__( self, provider: str, reset_after: float | None = None, ) -> None: super().__init__( f"Circuit breaker open for {provider}", provider=provider, recoverable=True, ) self.reset_after = reset_after # ============================================================================= # Agent Processing Exceptions # ============================================================================= class AgentProcessingError(AgentError): """Error during agent processing.""" def __init__( self, message: str, agent_name: str = "", recoverable: bool = True, ) -> None: super().__init__(message, recoverable) self.agent_name = agent_name class DMAgentError(AgentProcessingError): """Error in Dungeon Master agent.""" def __init__(self, message: str, recoverable: bool = True) -> None: super().__init__(message, agent_name="DungeonMaster", recoverable=recoverable) class RulesAgentError(AgentProcessingError): """Error in Rules Arbiter agent.""" def __init__(self, message: str, recoverable: bool = True) -> None: super().__init__(message, agent_name="RulesArbiter", recoverable=recoverable) class VoiceNarratorError(AgentProcessingError): """Error in Voice Narrator agent.""" def __init__(self, message: str, recoverable: bool = True) -> None: super().__init__(message, agent_name="VoiceNarrator", recoverable=recoverable) # ============================================================================= # Tool Execution Exceptions # ============================================================================= class ToolExecutionError(AgentError): """Error executing a tool.""" def __init__( self, tool_name: str, message: str, original_error: Exception | None = None, recoverable: bool = True, ) -> None: super().__init__( f"Tool '{tool_name}' failed: {message}", recoverable=recoverable, ) self.tool_name = tool_name self.original_error = original_error class ToolTimeoutError(ToolExecutionError): """Tool execution timed out.""" def __init__( self, tool_name: str, timeout_seconds: float, ) -> None: super().__init__( tool_name, f"Execution timed out after {timeout_seconds}s", recoverable=True, ) self.timeout_seconds = timeout_seconds class ToolNotFoundError(ToolExecutionError): """Requested tool not found.""" def __init__(self, tool_name: str) -> None: super().__init__( tool_name, "Tool not found in available tools", recoverable=False, ) # ============================================================================= # Orchestration Exceptions # ============================================================================= class OrchestratorError(AgentError): """Error in agent orchestration.""" pass class OrchestratorNotInitializedError(OrchestratorError): """Orchestrator not properly initialized.""" def __init__(self) -> None: super().__init__( "Orchestrator not initialized. Call setup() first.", recoverable=True, ) class TurnProcessingError(OrchestratorError): """Error processing a player turn.""" def __init__( self, message: str, partial_result: object | None = None, ) -> None: super().__init__(message, recoverable=True) self.partial_result = partial_result # ============================================================================= # State Exceptions # ============================================================================= class GameStateError(AgentError): """Error with game state.""" pass class StateConsistencyError(GameStateError): """Game state became inconsistent.""" def __init__( self, message: str, state_snapshot: dict[str, object] | None = None, ) -> None: super().__init__(message, recoverable=True) self.state_snapshot = state_snapshot # ============================================================================= # Graceful Error Messages # ============================================================================= GRACEFUL_ERROR_MESSAGES: dict[str, str] = { "llm_timeout": ( "The magical winds of computation blow slowly today. " "Please try again in a moment." ), "llm_rate_limit": ( "The ethereal realm is experiencing heavy traffic. " "Take a breath and try again shortly." ), "llm_all_failed": ( "The arcane servers are temporarily unreachable. " "Your adventure continues in text mode." ), "tool_failed": ( "The mystical tools encountered interference. " "Let's try a different approach..." ), "voice_unavailable": ( "The voice of the narrator is temporarily silenced. " "Continuing with text narration." ), "mcp_unavailable": ( "The game mechanics server is resting. " "Using simplified rules for now." ), "general_error": ( "An unexpected twist in the fabric of reality occurred. " "The adventure continues nonetheless." ), } def get_graceful_message(error_type: str) -> str: """ Get a user-friendly error message. Args: error_type: Type of error (key in GRACEFUL_ERROR_MESSAGES) Returns: User-friendly error message """ return GRACEFUL_ERROR_MESSAGES.get( error_type, GRACEFUL_ERROR_MESSAGES["general_error"], )