""" Memory System Exceptions Custom exception classes for the Agent Memory System. """ from typing import Any from uuid import UUID class MemoryError(Exception): """Base exception for all memory-related errors.""" def __init__( self, message: str, *, memory_type: str | None = None, scope_type: str | None = None, scope_id: str | None = None, details: dict[str, Any] | None = None, ) -> None: super().__init__(message) self.message = message self.memory_type = memory_type self.scope_type = scope_type self.scope_id = scope_id self.details = details or {} class MemoryNotFoundError(MemoryError): """Raised when a memory item is not found.""" def __init__( self, message: str = "Memory not found", *, memory_id: UUID | str | None = None, key: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.memory_id = memory_id self.key = key class MemoryCapacityError(MemoryError): """Raised when memory capacity limits are exceeded.""" def __init__( self, message: str = "Memory capacity exceeded", *, current_size: int = 0, max_size: int = 0, item_count: int = 0, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.current_size = current_size self.max_size = max_size self.item_count = item_count class MemoryExpiredError(MemoryError): """Raised when attempting to access expired memory.""" def __init__( self, message: str = "Memory has expired", *, key: str | None = None, expired_at: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.key = key self.expired_at = expired_at class MemoryStorageError(MemoryError): """Raised when memory storage operations fail.""" def __init__( self, message: str = "Memory storage operation failed", *, operation: str | None = None, backend: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.operation = operation self.backend = backend class MemoryConnectionError(MemoryError): """Raised when memory storage connection fails.""" def __init__( self, message: str = "Memory connection failed", *, backend: str | None = None, host: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.backend = backend self.host = host class MemorySerializationError(MemoryError): """Raised when memory serialization/deserialization fails.""" def __init__( self, message: str = "Memory serialization failed", *, content_type: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.content_type = content_type class MemoryScopeError(MemoryError): """Raised when memory scope operations fail.""" def __init__( self, message: str = "Memory scope error", *, requested_scope: str | None = None, allowed_scopes: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.requested_scope = requested_scope self.allowed_scopes = allowed_scopes or [] class MemoryConsolidationError(MemoryError): """Raised when memory consolidation fails.""" def __init__( self, message: str = "Memory consolidation failed", *, source_type: str | None = None, target_type: str | None = None, items_processed: int = 0, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.source_type = source_type self.target_type = target_type self.items_processed = items_processed class MemoryRetrievalError(MemoryError): """Raised when memory retrieval fails.""" def __init__( self, message: str = "Memory retrieval failed", *, query: str | None = None, retrieval_type: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.query = query self.retrieval_type = retrieval_type class EmbeddingError(MemoryError): """Raised when embedding generation fails.""" def __init__( self, message: str = "Embedding generation failed", *, content_length: int = 0, model: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.content_length = content_length self.model = model class CheckpointError(MemoryError): """Raised when checkpoint operations fail.""" def __init__( self, message: str = "Checkpoint operation failed", *, checkpoint_id: str | None = None, operation: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.checkpoint_id = checkpoint_id self.operation = operation class MemoryConflictError(MemoryError): """Raised when there's a conflict in memory (e.g., contradictory facts).""" def __init__( self, message: str = "Memory conflict detected", *, conflicting_ids: list[str | UUID] | None = None, conflict_type: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.conflicting_ids = conflicting_ids or [] self.conflict_type = conflict_type