Error Handling

Claif provides a comprehensive error handling system with structured exceptions, detailed error information, and recovery strategies.

Exception Hierarchy

Base Exception

class ClaifError(Exception):
    """Base exception for Claif framework."""
    
    def __init__(self, message: str, details: dict[str, Any] | None = None):
        super().__init__(message)
        self.message = message
        self.details = details or {}

All Claif exceptions inherit from ClaifError and include:

• Structured messages: Clear, actionable error descriptions
• Details dictionary: Additional context and debugging information
• Type hierarchy: Specific exception types for different error categories

Exception Types

ClaifError (base)
├── ProviderError              # Provider-specific errors
├── ConfigurationError         # Configuration issues
├── SessionError              # Session management errors
├── TransportError            # Network/transport errors
├── ClaifTimeoutError         # Timeout errors
└── ValidationError           # Input validation errors

Provider Errors

ProviderError

Used for provider-specific failures:

class ProviderError(ClaifError):
    """Error from a specific provider."""
    
    def __init__(self, provider: str, message: str, details: dict[str, Any] | None = None):
        super().__init__(f"{provider}: {message}", details)
        self.provider = provider

Common scenarios:

• API rate limiting
• Invalid API responses
• CLI tool failures
• Authentication errors

Example usage:

try:
    async for message in provider.query("Hello", options):
        yield message
except Exception as e:
    raise ProviderError(
        provider="claude",
        message="API request failed",
        details={
            "status_code": 429,
            "error_type": "rate_limit",
            "retry_after": 60
        }
    ) from e

Configuration Errors

ConfigurationError

Used for configuration-related issues:

class ConfigurationError(ClaifError):
    """Configuration-related error."""

Common scenarios:

• Missing API keys
• Invalid configuration files
• Malformed settings
• Missing required dependencies

Examples:

# Missing API key
raise ConfigurationError(
    "API key not found for provider 'claude'",
    details={
        "provider": "claude",
        "env_var": "ANTHROPIC_API_KEY",
        "config_file": "~/.claif/config.json"
    }
)

# Invalid configuration
raise ConfigurationError(
    "Invalid timeout value in configuration",
    details={
        "value": -1,
        "expected": "positive integer",
        "location": "providers.claude.timeout"
    }
)

Timeout Errors

ClaifTimeoutError

Used for operation timeouts:

class ClaifTimeoutError(ClaifError):
    """Operation timeout error."""

Common scenarios:

• Network request timeouts
• CLI process timeouts
• Session timeouts
• Response streaming timeouts

Example:

import asyncio

async def query_with_timeout(self, prompt: str, timeout: int):
    try:
        async with asyncio.timeout(timeout):
            return await self._execute_query(prompt)
    except asyncio.TimeoutError:
        raise ClaifTimeoutError(
            f"Query timed out after {timeout} seconds",
            details={
                "timeout": timeout,
                "provider": self.name,
                "prompt_length": len(prompt)
            }
        )

Transport Errors

TransportError

Used for network and communication failures:

class TransportError(ClaifError):
    """Transport layer error."""

Common scenarios:

• Network connectivity issues
• HTTP errors
• SSL/TLS failures
• DNS resolution failures

Example:

import aiohttp

try:
    async with session.post(url, json=payload) as response:
        return await response.json()
except aiohttp.ClientConnectorError as e:
    raise TransportError(
        "Failed to connect to provider API",
        details={
            "url": url,
            "error": str(e),
            "provider": self.name
        }
    ) from e

Validation Errors

ValidationError

Used for input validation failures:

class ValidationError(ClaifError):
    """Input validation error."""

Common scenarios:

• Empty or invalid prompts
• Invalid option values
• Malformed input data
• Type validation failures

Example:

def validate_prompt(prompt: str) -> None:
    if not prompt or not prompt.strip():
        raise ValidationError(
            "Prompt cannot be empty",
            details={
                "prompt_length": len(prompt),
                "prompt_stripped": len(prompt.strip()) if prompt else 0
            }
        )
    
    if len(prompt) > 100000:
        raise ValidationError(
            "Prompt exceeds maximum length",
            details={
                "prompt_length": len(prompt),
                "max_length": 100000
            }
        )

Session Errors

SessionError

Used for session management failures:

class SessionError(ClaifError):
    """Session management error."""

Common scenarios:

• Session file corruption
• Concurrent session access
• Session storage issues
• Session expiration

Example:

try:
    session = await self.load_session(session_id)
except FileNotFoundError:
    raise SessionError(
        f"Session not found: {session_id}",
        details={
            "session_id": session_id,
            "session_dir": self.session_dir,
            "available_sessions": await self.list_sessions()
        }
    )

Error Handling Patterns

Try-Catch with Specific Exceptions

from claif.common.errors import (
    ProviderError, 
    ConfigurationError, 
    ClaifTimeoutError,
    ValidationError
)

async def robust_query(client, prompt: str):
    try:
        response = await client.query(prompt)
        return response
        
    except ValidationError as e:
        # Handle user input errors - don't retry
        logger.error(f"Invalid input: {e.message}")
        raise
        
    except ConfigurationError as e:
        # Handle config errors - guide user to fix configuration
        logger.error(f"Configuration error: {e.message}")
        print("Please run 'claif config' to set up your configuration")
        raise
        
    except ClaifTimeoutError as e:
        # Handle timeouts - suggest retry with longer timeout
        logger.warning(f"Request timed out: {e.message}")
        logger.info("Try increasing timeout with --timeout option")
        raise
        
    except ProviderError as e:
        # Handle provider errors - attempt retry or fallback
        logger.warning(f"Provider error: {e.message}")
        if e.details.get("status_code") == 429:
            logger.info("Rate limited, waiting before retry...")
            await asyncio.sleep(e.details.get("retry_after", 60))
            # Retry logic here
        raise
        
    except Exception as e:
        # Handle unexpected errors
        logger.error(f"Unexpected error: {e}")
        raise

Context Manager for Error Handling

from contextlib import asynccontextmanager

@asynccontextmanager
async def error_context(operation: str, provider: str = None):
    """Context manager for consistent error handling."""
    try:
        logger.debug(f"Starting {operation}")
        yield
        logger.debug(f"Completed {operation}")
        
    except ClaifError:
        # Re-raise Claif errors as-is
        raise
        
    except Exception as e:
        # Convert unexpected errors to ClaifError
        error_msg = f"Unexpected error during {operation}: {e}"
        details = {"operation": operation}
        if provider:
            details["provider"] = provider
            
        logger.error(error_msg, extra=details)
        raise ClaifError(error_msg, details) from e

# Usage
async def query_provider(self, prompt: str):
    async with error_context("provider query", self.name):
        return await self._execute_query(prompt)

Retry Logic with Error Types

from tenacity import (
    retry, 
    stop_after_attempt, 
    wait_exponential,
    retry_if_exception_type
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((ProviderError, TransportError, ClaifTimeoutError))
)
async def query_with_retry(provider, prompt: str, options):
    """Query with automatic retry for transient errors."""
    try:
        return await provider.query(prompt, options)
    except (ValidationError, ConfigurationError):
        # Don't retry permanent errors
        raise
    except ProviderError as e:
        # Only retry certain provider errors
        if e.details.get("status_code") in [429, 502, 503, 504]:
            logger.warning(f"Retrying after provider error: {e.message}")
            raise
        else:
            # Don't retry permanent provider errors
            raise

Error Recovery Strategies

Automatic Fallback

async def query_with_fallback(prompt: str, providers: list[str]):
    """Try multiple providers with automatic fallback."""
    last_error = None
    
    for provider_name in providers:
        try:
            provider = get_provider(provider_name)
            return await provider.query(prompt)
            
        except (ProviderError, ClaifTimeoutError) as e:
            logger.warning(f"Provider {provider_name} failed: {e.message}")
            last_error = e
            continue
            
        except (ConfigurationError, ValidationError):
            # Don't try other providers for these errors
            raise
    
    # All providers failed
    raise ProviderError(
        "all_providers",
        "All providers failed",
        details={
            "tried_providers": providers,
            "last_error": str(last_error)
        }
    ) from last_error

Graceful Degradation

async def query_with_degradation(prompt: str, options):
    """Query with graceful degradation of features."""
    try:
        # Try full-featured query first
        return await provider.query(prompt, options)
        
    except ProviderError as e:
        if "tools_not_supported" in e.details:
            # Retry without tools
            logger.info("Retrying without tool support")
            options_no_tools = dataclasses.replace(options, enable_tools=False)
            return await provider.query(prompt, options_no_tools)
        raise
        
    except ClaifTimeoutError:
        # Retry with shorter prompt
        if len(prompt) > 1000:
            logger.info("Retrying with truncated prompt")
            short_prompt = prompt[:1000] + "..."
            return await provider.query(short_prompt, options)
        raise

Error Reporting

Structured Error Information

def format_error_report(error: ClaifError) -> dict:
    """Format error for reporting/logging."""
    return {
        "error_type": type(error).__name__,
        "message": error.message,
        "details": error.details,
        "timestamp": datetime.utcnow().isoformat(),
        "provider": getattr(error, 'provider', None),
        "stack_trace": traceback.format_exc() if logger.level <= logging.DEBUG else None
    }

User-Friendly Error Messages

def get_user_message(error: ClaifError) -> str:
    """Convert technical error to user-friendly message."""
    
    if isinstance(error, ConfigurationError):
        if "api_key" in error.message.lower():
            return (
                "API key not configured. "
                "Please run 'claif config' to set up your API keys."
            )
        return f"Configuration error: {error.message}"
    
    elif isinstance(error, ProviderError):
        if error.details.get("status_code") == 429:
            return (
                "Rate limit exceeded. "
                "Please wait a moment before trying again."
            )
        elif error.details.get("status_code") == 401:
            return (
                "Authentication failed. "
                "Please check your API key configuration."
            )
        return f"Provider error: {error.message}"
    
    elif isinstance(error, ClaifTimeoutError):
        return (
            "Request timed out. "
            "Try using a shorter prompt or increasing the timeout."
        )
    
    elif isinstance(error, ValidationError):
        return f"Invalid input: {error.message}"
    
    else:
        return f"An error occurred: {error.message}"

Debugging and Logging

Error Logging

import logging
from claif.common.utils import logger

def log_error(error: ClaifError, context: dict = None):
    """Log error with appropriate level and context."""
    
    log_data = {
        "error_type": type(error).__name__,
        "message": error.message,
        **error.details,
        **(context or {})
    }
    
    if isinstance(error, (ValidationError, ConfigurationError)):
        # User errors - info level
        logger.info("User error occurred", extra=log_data)
    elif isinstance(error, ProviderError):
        # Provider errors - warning level
        logger.warning("Provider error occurred", extra=log_data)
    else:
        # System errors - error level
        logger.error("System error occurred", extra=log_data)

Debug Information

def add_debug_info(error: ClaifError, **debug_info):
    """Add debug information to error details."""
    error.details.setdefault("debug", {})
    error.details["debug"].update(debug_info)
    return error

# Usage
try:
    result = await api_call()
except Exception as e:
    error = ProviderError(self.name, "API call failed")
    add_debug_info(
        error,
        api_endpoint=url,
        request_id=request_id,
        response_headers=response.headers
    )
    raise error from e

Testing Error Handling

Unit Tests

import pytest
from claif.common.errors import ProviderError, ConfigurationError

def test_provider_error():
    """Test ProviderError creation and attributes."""
    error = ProviderError(
        provider="test",
        message="Test error",
        details={"code": 500}
    )
    
    assert error.provider == "test"
    assert "test: Test error" in str(error)
    assert error.details["code"] == 500

@pytest.mark.asyncio
async def test_error_handling():
    """Test error handling in provider."""
    provider = TestProvider()
    
    with pytest.raises(ValidationError):
        await provider.query("", options)
    
    with pytest.raises(ConfigurationError):
        provider.api_key = None
        await provider.query("test", options)

Mock Error Scenarios

from unittest.mock import patch, AsyncMock

@pytest.mark.asyncio
async def test_timeout_handling():
    """Test timeout error handling."""
    provider = TestProvider()
    
    with patch.object(provider, '_execute_query') as mock_query:
        mock_query.side_effect = asyncio.TimeoutError()
        
        with pytest.raises(ClaifTimeoutError):
            await provider.query("test", options)

@pytest.mark.asyncio  
async def test_provider_error_handling():
    """Test provider error handling."""
    provider = TestProvider()
    
    with patch.object(provider, '_execute_query') as mock_query:
        mock_query.side_effect = Exception("API Error")
        
        with pytest.raises(ProviderError) as exc_info:
            await provider.query("test", options)
        
        assert "API Error" in str(exc_info.value)

Best Practices

1. Use Specific Exception Types

# Good - specific exception
raise ConfigurationError("Missing API key for Claude provider")

# Bad - generic exception  
raise Exception("Configuration error")

2. Include Helpful Details

# Good - detailed error information
raise ProviderError(
    provider="claude",
    message="Rate limit exceeded",
    details={
        "status_code": 429,
        "retry_after": 60,
        "limit_type": "requests_per_minute",
        "current_usage": 100
    }
)

# Bad - minimal information
raise ProviderError("claude", "Error")

3. Chain Exceptions

# Good - preserve original exception
try:
    result = api_call()
except HTTPError as e:
    raise ProviderError(
        provider="claude",
        message="API request failed"
    ) from e

# Bad - lose original exception information
try:
    result = api_call()
except HTTPError:
    raise ProviderError("claude", "API request failed")

4. Handle Errors at Appropriate Level

# Handle specific errors where they can be resolved
try:
    config = load_config()
except ConfigurationError:
    # Can handle here - prompt user to configure
    run_configuration_wizard()
    config = load_config()

# Let generic errors bubble up
try:
    response = await provider.query(prompt)
except ProviderError:
    # Let calling code handle provider errors
    raise

The error handling system provides comprehensive coverage of failure scenarios while maintaining clear error types and actionable information for debugging and recovery.