Error Handling

from agent_sentinel import (
    AgentSentinelError,
    BudgetExceededError,
    PolicyViolationError,
    ReplayDivergenceError,
    NetworkError,
    SyncError,
    TimeoutError,
    ConfigurationError,
    # Enforcement-mode blocks (subclasses noted below)
    PolicyNotConfiguredError,    # PolicyViolationError
    NoPolicyForWriteError,       # PolicyViolationError
    AgentKilledError,
    KillSwitchUnavailableError,  # AgentKilledError
)

try:
    # Some operation
    ...
except AgentSentinelError as e:
    print(f"Error: {e.message}")
    print(f"Code: {e.error_code}")
    print(f"Details: {e.details}")
    print(f"Recoverable: {e.recoverable}")

    # Serialize for logging
    error_dict = e.to_dict()

from agent_sentinel import BudgetExceededError, guarded_action, PolicyEngine

PolicyEngine.configure(run_budget=1.0)

@guarded_action(name="expensive_call", cost_usd=0.60)
def call_api():
    return "result"

try:
    call_api()  # First call: $0.60 spent
    call_api()  # Second call: Would exceed $1.00 budget → raises BudgetExceededError
except BudgetExceededError as e:
    print(f"Budget exceeded: spent ${e.details['spent']}, limit ${e.details['limit']}")
    print(f"Budget type: {e.details['budget_type']}")  # "run", "session", or "action"

    # Graceful degradation
    if e.details['budget_type'] == 'action':
        # Switch to cheaper model
        use_cheaper_model()
    elif e.details['budget_type'] == 'run':
        # Stop the run
        terminate_run()

from agent_sentinel import PolicyViolationError, guarded_action, PolicyEngine

PolicyEngine.configure(
    denied_actions=["delete_production_db"],
    allowed_actions=["read_db", "write_db"],  # Allowlist mode
)

@guarded_action(name="delete_production_db")
def dangerous_action():
    return "never executed"

try:
    dangerous_action()
except PolicyViolationError as e:
    print(f"Policy violation: {e.message}")
    print(f"Rule: {e.details['policy_rule']}")  # "denied_action" or "allowlist" or "rate_limit"
    print(f"Action: {e.details['action_name']}")

    # Log security incident
    log_security_event(
        action=e.details['action_name'],
        rule=e.details['policy_rule'],
        agent_id=get_agent_id()
    )

from agent_sentinel import ReplayDivergenceError, replay_mode

try:
    with replay_mode(run_id="run-123", strict=True):
        # If agent behavior diverges from recording, raises ReplayDivergenceError
        run_agent()
except ReplayDivergenceError as e:
    print(f"Divergence detected: {e.message}")
    print(f"Expected: {e.details['expected']}")
    print(f"Actual: {e.details['actual']}")
    print(f"Action: {e.details['action_name']}")

    # Analyze divergence
    analyze_non_determinism(e.details)

from agent_sentinel import NetworkError, enable_remote_sync

try:
    enable_remote_sync(
        api_token="as_invalid_token",
        run_id="run-123"
    )
except NetworkError as e:
    print(f"Network error: {e.message}")
    print(f"Status code: {e.details['status_code']}")
    print(f"Endpoint: {e.details['endpoint']}")
    print(f"Recoverable: {e.recoverable}")  # True for 5xx, False for 4xx

    if e.recoverable:
        # Retry with backoff
        retry_with_backoff(enable_remote_sync, ...)
    else:
        # Authentication or configuration issue - don't retry
        log_error("Invalid API token or endpoint")

from agent_sentinel import SyncError

try:
    # Sync errors are typically raised internally and logged
    # but you can handle them if using manual flush
    sync_service.flush_now()
except SyncError as e:
    print(f"Sync failed: {e.message}")
    print(f"Batch size: {e.details['batch_size']}")
    print(f"Retry count: {e.details['retry_count']}")
    print(f"Recoverable: {e.recoverable}")  # Always True

    # Background sync will automatically retry
    # No action needed - just log for visibility
    log_warning(f"Sync retry {e.details['retry_count']}")

from agent_sentinel import TimeoutError, ApprovalClient

client = ApprovalClient(
    api_token="as_your_api_key_here",
)

try:
    response = client.request_approval_sync(
        action_name="expensive_operation",
        timeout_seconds=60,  # 1 minute timeout
        # ...
    )
except TimeoutError as e:
    print(f"Timeout: {e.message}")
    print(f"Operation: {e.details['operation']}")
    print(f"Timeout: {e.details['timeout_seconds']}s")
    print(f"Recoverable: {e.recoverable}")  # True

    # Handle timeout gracefully
    if e.details['operation'] == 'approval':
        # Maybe extend timeout or proceed without approval
        log_warning("Approval timed out - using default policy")

from agent_sentinel import ConfigurationError, PolicyEngine

try:
    PolicyEngine.configure(
        session_budget=-10.0,  # Invalid: negative budget
    )
except ConfigurationError as e:
    print(f"Configuration error: {e.message}")
    print(f"Config key: {e.details['config_key']}")
    print(f"Invalid value: {e.details.get('value')}")
    print(f"Recoverable: {e.recoverable}")  # False

    # Fix configuration
    PolicyEngine.configure(session_budget=10.0)

from agent_sentinel import NoPolicyForWriteError

try:
    issue_refund("ord_123", 40.0)  # impact="high", no policy loaded yet
except NoPolicyForWriteError as e:
    print(e.error_code)  # NO_POLICY_FOR_WRITE

from agent_sentinel import AgentKilledError, KillSwitchUnavailableError

try:
    run_step()
except KillSwitchUnavailableError:
    # platform unreachable under fail_closed; retry later or stop
    ...
except AgentKilledError as e:
    # an operator killed this target; stop the run
    print(e.details["target_type"], e.details["target_id"])

from agent_sentinel import BudgetExceededError, guarded_action

@guarded_action(name="premium_llm", cost_usd=0.50)
def call_premium_llm(prompt):
    return premium_model.generate(prompt)

@guarded_action(name="cheap_llm", cost_usd=0.01)
def call_cheap_llm(prompt):
    return cheap_model.generate(prompt)

def generate_with_fallback(prompt):
    try:
        return call_premium_llm(prompt)
    except BudgetExceededError:
        # Degrade to cheaper model
        logger.warning("Budget exceeded, using cheap model")
        return call_cheap_llm(prompt)

from agent_sentinel import NetworkError
import time

def call_with_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except NetworkError as e:
            if not e.recoverable or attempt == max_retries - 1:
                raise
            # Exponential backoff
            wait = (2 ** attempt) + (random.random() * 0.1)
            time.sleep(wait)

from agent_sentinel import AgentSentinelError

def monitor_errors(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except AgentSentinelError as e:
            # Send to monitoring service
            monitoring.track_error(
                error_code=e.error_code,
                message=e.message,
                details=e.details,
                recoverable=e.recoverable,
            )
            raise
    return wrapper