""" Safety Framework Exceptions Custom exception classes for the safety and guardrails framework. """ from typing import Any class SafetyError(Exception): """Base exception for all safety-related errors.""" def __init__( self, message: str, *, action_id: str | None = None, agent_id: str | None = None, details: dict[str, Any] | None = None, ) -> None: super().__init__(message) self.message = message self.action_id = action_id self.agent_id = agent_id self.details = details or {} class PermissionDeniedError(SafetyError): """Raised when an action is not permitted.""" def __init__( self, message: str = "Permission denied", *, action_type: str | None = None, resource: str | None = None, required_permission: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.action_type = action_type self.resource = resource self.required_permission = required_permission class BudgetExceededError(SafetyError): """Raised when cost budget is exceeded.""" def __init__( self, message: str = "Budget exceeded", *, budget_type: str = "session", current_usage: float = 0.0, budget_limit: float = 0.0, unit: str = "tokens", **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.budget_type = budget_type self.current_usage = current_usage self.budget_limit = budget_limit self.unit = unit class RateLimitExceededError(SafetyError): """Raised when rate limit is exceeded.""" def __init__( self, message: str = "Rate limit exceeded", *, limit_type: str = "actions", limit_value: int = 0, window_seconds: int = 60, retry_after_seconds: float = 0.0, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.limit_type = limit_type self.limit_value = limit_value self.window_seconds = window_seconds self.retry_after_seconds = retry_after_seconds class LoopDetectedError(SafetyError): """Raised when an action loop is detected.""" def __init__( self, message: str = "Loop detected", *, loop_type: str = "exact", repetition_count: int = 0, action_pattern: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.loop_type = loop_type self.repetition_count = repetition_count self.action_pattern = action_pattern or [] class ApprovalRequiredError(SafetyError): """Raised when human approval is required.""" def __init__( self, message: str = "Human approval required", *, approval_id: str | None = None, reason: str | None = None, timeout_seconds: int = 300, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.approval_id = approval_id self.reason = reason self.timeout_seconds = timeout_seconds class ApprovalDeniedError(SafetyError): """Raised when human explicitly denies an action.""" def __init__( self, message: str = "Approval denied by human", *, approval_id: str | None = None, denied_by: str | None = None, denial_reason: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.approval_id = approval_id self.denied_by = denied_by self.denial_reason = denial_reason class ApprovalTimeoutError(SafetyError): """Raised when approval request times out.""" def __init__( self, message: str = "Approval request timed out", *, approval_id: str | None = None, timeout_seconds: int = 300, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.approval_id = approval_id self.timeout_seconds = timeout_seconds class RollbackError(SafetyError): """Raised when rollback fails.""" def __init__( self, message: str = "Rollback failed", *, checkpoint_id: str | None = None, failed_actions: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.checkpoint_id = checkpoint_id self.failed_actions = failed_actions or [] class CheckpointError(SafetyError): """Raised when checkpoint creation fails.""" def __init__( self, message: str = "Checkpoint creation failed", *, checkpoint_type: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.checkpoint_type = checkpoint_type class ValidationError(SafetyError): """Raised when action validation fails.""" def __init__( self, message: str = "Validation failed", *, validation_rules: list[str] | None = None, failed_rules: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.validation_rules = validation_rules or [] self.failed_rules = failed_rules or [] class ContentFilterError(SafetyError): """Raised when content filtering detects prohibited content.""" def __init__( self, message: str = "Prohibited content detected", *, filter_type: str | None = None, detected_patterns: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.filter_type = filter_type self.detected_patterns = detected_patterns or [] class SandboxError(SafetyError): """Raised when sandbox execution fails.""" def __init__( self, message: str = "Sandbox execution failed", *, exit_code: int | None = None, stderr: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.exit_code = exit_code self.stderr = stderr class SandboxTimeoutError(SandboxError): """Raised when sandbox execution times out.""" def __init__( self, message: str = "Sandbox execution timed out", *, timeout_seconds: int = 300, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.timeout_seconds = timeout_seconds class EmergencyStopError(SafetyError): """Raised when emergency stop is triggered.""" def __init__( self, message: str = "Emergency stop triggered", *, stop_type: str = "kill", triggered_by: str | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.stop_type = stop_type self.triggered_by = triggered_by class PolicyViolationError(SafetyError): """Raised when an action violates a safety policy.""" def __init__( self, message: str = "Policy violation", *, policy_name: str | None = None, violated_rules: list[str] | None = None, **kwargs: Any, ) -> None: super().__init__(message, **kwargs) self.policy_name = policy_name self.violated_rules = violated_rules or []