Error Patterns

Table of Contents

• Overview
• When to Use
• Error Classification
• By Severity
• By Recoverability
• Quick Start
• Standard Error Handler
• Error Result
• Common Patterns
• Authentication Errors (401/403))
• Rate Limit Errors (429))
• Timeout Errors
• Context Too Large (400))
• Integration Pattern
• Detailed Resources
• Exit Criteria

Error Patterns

Overview

Standardized error handling patterns for consistent, production-grade behavior across plugins. Provides error classification, recovery strategies, and debugging workflows.

When To Use

• Building resilient integrations
• Need consistent error handling
• Want graceful degradation
• Debugging production issues

When NOT To Use

• Project doesn't use the leyline infrastructure patterns
• Simple scripts without service architecture needs

Error Classification

By Severity

By Recoverability

class ErrorCategory(Enum):
    TRANSIENT = "transient"      # Retry likely to succeed
    PERMANENT = "permanent"       # Retry won't help
    CONFIGURATION = "config"      # User action needed
    RESOURCE = "resource"         # Quota/limit issue

Verification: Run the command with --help flag to verify availability.

Quick Start

Standard Error Handler

from leyline.error_patterns import handle_error, ErrorCategory

try:
    result = service.execute(prompt)
except RateLimitError as e:
    return handle_error(e, ErrorCategory.RESOURCE, {
        "retry_after": e.retry_after,
        "service": "gemini"
    })
except AuthError as e:
    return handle_error(e, ErrorCategory.CONFIGURATION, {
        "action": "Run 'gemini auth login'"
    })

Verification: Run the command with --help flag to verify availability.

Error Result

@dataclass
class ErrorResult:
    category: ErrorCategory
    message: str
    recoverable: bool
    suggested_action: str
    metadata: dict

Verification: Run the command with --help flag to verify availability.

Common Patterns

Authentication Errors (401/403)

• Verify credentials exist
• Check token expiration
• Validate permissions/scopes
• Suggest re-authentication

Rate Limit Errors (429)

• Extract retry-after header
• Log for quota tracking
• Implement backoff
• Consider alternative service

Timeout Errors

• Increase timeout for retries
• Break into smaller requests
• Use async patterns
• Consider different model

Context Too Large (400)

• Estimate tokens before request
• Split into multiple requests
• Reduce input content
• Use larger context model

Integration Pattern

# In your skill's frontmatter
dependencies: [leyline:error-patterns]

Verification: Run the command with --help flag to verify availability.

Detailed Resources

• Classification: See modules/classification.md for error taxonomy
• Recovery: See modules/recovery-strategies.md for handling patterns
• Agent Damage Control: See modules/agent-damage-control.md for multi-agent error recovery and escalation

Exit Criteria

• Error classified correctly
• Appropriate recovery attempted
• User-actionable message provided
• Error logged for debugging

Agent Damage Control

Agent-level error recovery patterns for multi-agent coordination. Maps service-level error categories to agent-level equivalents and defines escalation ladders.

Error Categories

Agent-level error categories extend the service-level taxonomy from error-patterns:

Escalation Ladder

Three-tier escalation with clear handoff points:

Tier 1: Agent Self-Recovery
  Agent detects issue, attempts automated recovery
  (retry, context shed, conflict resolution)
        |
        | Fails after 2 attempts
        v
Tier 2: Lead Agent Intervention
  Lead reassigns tasks, spawns replacement agents,
  coordinates resolution across team
        |
        | Cannot resolve or CRITICAL severity
        v
Tier 3: Human Escalation
  Human notified with full context, recommended actions,
  and current state summary

Quick Reference

Recovery Patterns

Agent Crash Recovery

• Detect orphaned tasks via heartbeat monitoring
• Apply "replace don't wait" doctrine: spawn new agent immediately
• Recover state from last checkpoint or committed work
• Reassign orphaned tasks to replacement agent

Context Overflow Handling

• Detection signals: truncated responses, repeated content, loss of coherence
• Graceful handoff: summarize state, write to file, spawn continuation
• Progressive context shedding: drop least-relevant loaded modules first

Merge Conflict Resolution

• Stop all agents working on conflicting files
• Lead agent resolves conflicts using diff analysis
• Resume agents with updated base after resolution
• Prevention: assign non-overlapping file scopes

Partial Failure Handling

• Triage: categorize each sub-task result (success/failure/partial)
• Salvage: commit successful work first
• Retry: attempt failed tasks with fresh context
• Report: document what succeeded and what needs manual attention

Integration

Reference specific recovery patterns from orchestrator skills:

On agent crash: follow leyline:error-patterns/modules/agent-damage-control.md

Exit Criteria

• Failed agent identified and replaced (or escalated)
• Orphaned tasks reassigned to healthy agents
• Successful work preserved and committed
• Recovery actions logged for post-mortem

Error Classification

HTTP Status Code Mapping

ERROR_CLASSIFICATION = {
    # Authentication
    401: ErrorCategory.CONFIGURATION,
    403: ErrorCategory.CONFIGURATION,

    # Client errors
    400: ErrorCategory.PERMANENT,
    404: ErrorCategory.CONFIGURATION,
    422: ErrorCategory.PERMANENT,

    # Rate limits
    429: ErrorCategory.RESOURCE,

    # Server errors (transient)
    500: ErrorCategory.TRANSIENT,
    502: ErrorCategory.TRANSIENT,
    503: ErrorCategory.TRANSIENT,
    504: ErrorCategory.TRANSIENT,
}

Error Detection Patterns

By Message Content

def classify_by_message(error_msg: str) -> ErrorCategory:
    error_msg = error_msg.lower()

    if any(k in error_msg for k in ["rate limit", "quota", "exceeded"]):
        return ErrorCategory.RESOURCE

    if any(k in error_msg for k in ["auth", "token", "credential", "permission"]):
        return ErrorCategory.CONFIGURATION

    if any(k in error_msg for k in ["timeout", "connection", "network"]):
        return ErrorCategory.TRANSIENT

    if any(k in error_msg for k in ["invalid", "malformed", "syntax"]):
        return ErrorCategory.PERMANENT

    return ErrorCategory.TRANSIENT  # Default: assume retryable

Service-Specific Errors

Gemini Errors

Qwen Errors

Error Hierarchy

class LeylineError(Exception):
    """Base error for leyline infrastructure."""
    category = ErrorCategory.TRANSIENT

class AuthenticationError(LeylineError):
    """Authentication/authorization failures."""
    category = ErrorCategory.CONFIGURATION

class RateLimitError(LeylineError):
    """Rate limit or quota exceeded."""
    category = ErrorCategory.RESOURCE

class ServiceUnavailableError(LeylineError):
    """Service temporarily unavailable."""
    category = ErrorCategory.TRANSIENT

class ConfigurationError(LeylineError):
    """Missing or invalid configuration."""
    category = ErrorCategory.CONFIGURATION

Recovery Strategies

Strategy Selection

def select_recovery_strategy(error: LeylineError) -> RecoveryStrategy:
    strategies = {
        ErrorCategory.TRANSIENT: RetryWithBackoff(),
        ErrorCategory.RESOURCE: WaitOrSecondary(),
        ErrorCategory.CONFIGURATION: UserActionRequired(),
        ErrorCategory.PERMANENT: ReportAndAbort(),
    }
    return strategies[error.category]

Retry With Backoff

class RetryWithBackoff:
    def __init__(self, max_retries=3, base_delay=1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    def execute(self, operation, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return operation(*args, **kwargs)
            except LeylineError as e:
                if e.category != ErrorCategory.TRANSIENT:
                    raise
                if attempt < self.max_retries - 1:
                    delay = self.base_delay * (2 ** attempt)
                    time.sleep(delay)
        raise MaxRetriesExceededError()

Wait or Secondary

class WaitOrSecondary: def execute(self, error: RateLimitError, secondary_service=None): retry_after = error.metadata.get("retry_after", 60)

if retry_after < 30: # Short wait time.sleep(retry_after) return "retry"

if secondary_service: return ("secondary", secondary_service)

return ("defer", retry_after)

## User Action Required

class UserActionRequired: def execute(self, error: ConfigurationError) -> dict: actions = { "auth": "Run '{service} auth login' or set {ENV_VAR}", "model": "Check available models with '{service} --help'", "config": "Verify configuration in ~/.claude/leyline/", }

return { "status": "user_action_required", "message": error.message, "suggested_action": actions.get( error.metadata.get("type"), "Check service documentation" ) }

## Graceful Degradation

def execute_with_degradation( primary_fn, secondary_fn=None,

degraded_fn=None ): """Execute with graceful degradation.""" try: return primary_fn() except LeylineError as e: if e.category == ErrorCategory.TRANSIENT and secondary_fn: return secondary_fn() if degraded_fn: return degraded_fn(partial=True) raise

## Logging and Alerting

def log_error_with_context(error: LeylineError, context: dict): """Log error with full context for debugging.""" log_entry = { "timestamp": datetime.utcnow().isoformat(), "error_type": type(error).__name__, "category": error.category.value, "message": str(error), "recoverable": error.category in [ ErrorCategory.TRANSIENT, ErrorCategory.RESOURCE ], "context": context }

logger.error(json.dumps(log_entry))

Alert on critical errors

if error.category == ErrorCategory.PERMANENT: send_alert(log_entry)