temp/.opencode/agents/shared/error_handling.md

Shared Error Handling Patterns

This document provides standardized error handling patterns that all agents MUST use to ensure consistent, reliable operation.

Core Principles

1. Fail Fast - Detect errors early and fail immediately
2. Be Specific - Use appropriate exception types
3. Log Context - Include relevant information for debugging
4. Clean Up - Always clean up resources on failure
5. Retry Intelligently - Use exponential backoff for transient errors

Standard Error Types

# Network/API errors that might be transient
class NetworkError(Exception):
    """Network or API communication failure."""
    pass

# Git operation failures
class GitError(Exception):
    """Git command or operation failure."""
    pass

class GitConflictError(GitError):
    """Git merge or rebase conflict."""
    pass

# Forgejo API specific errors
class ForgejoError(Exception):
    """Forgejo API operation failure."""
    pass

# Agent coordination errors
class CoordinationError(Exception):
    """Agent coordination or claiming failure."""
    pass

class AlreadyClaimedError(CoordinationError):
    """Work item is already claimed by another agent."""
    pass

# Configuration/setup errors
class ConfigurationError(Exception):
    """Missing or invalid configuration."""
    pass

# Operation failures after retries
class OperationFailedError(Exception):
    """Operation failed after all retry attempts."""
    def __init__(self, operation_name, last_error):
        self.operation_name = operation_name
        self.last_error = last_error
        super().__init__(f"{operation_name} failed: {last_error}")

Standard Recovery Pattern

def safe_operation_with_recovery(operation_name, operation_func, 
                                recovery_func=None, max_retries=3):
    """
    Standard pattern for operations that might fail.
    
    Args:
        operation_name: Human-readable operation description
        operation_func: Function to execute (callable)
        recovery_func: Optional recovery function(error_type, error)
        max_retries: Maximum retry attempts
        
    Returns:
        Result of operation_func
        
    Raises:
        OperationFailedError: If all attempts fail
    """
    last_error = None
    
    for attempt in range(max_retries):
        try:
            print(f"[ATTEMPT {attempt+1}/{max_retries}] {operation_name}")
            result = operation_func()
            print(f"[SUCCESS] {operation_name} completed")
            return result
            
        except GitConflictError as e:
            print(f"[GIT CONFLICT] {operation_name}: {e}")
            if recovery_func:
                try:
                    recovery_func("git_conflict", e)
                except Exception as recovery_error:
                    print(f"[RECOVERY FAILED] {recovery_error}")
            last_error = e
            
        except NetworkError as e:
            print(f"[NETWORK ERROR] {operation_name}: {e}")
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff: 1, 2, 4...
                print(f"[RETRY] Waiting {wait_time}s before retry...")
                time.sleep(wait_time)
            last_error = e
            
        except ForgejoError as e:
            print(f"[FORGEJO ERROR] {operation_name}: {e}")
            # Forgejo errors might be rate limits
            if "rate limit" in str(e).lower() and attempt < max_retries - 1:
                wait_time = 60  # Wait longer for rate limits
                print(f"[RATE LIMIT] Waiting {wait_time}s...")
                time.sleep(wait_time)
            last_error = e
            
        except Exception as e:
            print(f"[UNEXPECTED ERROR] {operation_name}: {type(e).__name__}: {e}")
            last_error = e
            # Don't retry unexpected errors unless recovery possible
            if recovery_func:
                try:
                    recovery_func("unexpected", e)
                    continue  # Try again after recovery
                except:
                    break
            break
    
    # All attempts failed
    print(f"[FAILED] {operation_name} after {attempt + 1} attempts")
    raise OperationFailedError(operation_name, last_error)

Cleanup Manager

class CleanupManager:
    """Context manager that ensures cleanup happens even on failure."""
    
    def __init__(self, cleanup_func, cleanup_description="cleanup"):
        """
        Args:
            cleanup_func: Function to call for cleanup
            cleanup_description: Human-readable description
        """
        self.cleanup_func = cleanup_func
        self.cleanup_description = cleanup_description
        self.cleaned = False
        self.error_during_cleanup = None
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if not self.cleaned:
            try:
                print(f"[CLEANUP] Performing {self.cleanup_description}")
                self.cleanup_func()
                self.cleaned = True
                print(f"[CLEANUP] {self.cleanup_description} completed")
            except Exception as e:
                self.error_during_cleanup = e
                print(f"[CLEANUP FAILED] {self.cleanup_description}: {e}")
                # Don't suppress the original exception
        
        # Don't suppress any exception
        return False
    
    def cleanup_now(self):
        """Manual cleanup trigger."""
        if not self.cleaned:
            self.cleanup_func()
            self.cleaned = True

Usage Examples

Git Operations

def clone_with_retry(repo_url, target_dir):
    """Clone a repository with automatic retry and cleanup."""
    
    def do_clone():
        result = subprocess.run(
            ['git', 'clone', repo_url, target_dir],
            capture_output=True, text=True
        )
        if result.returncode != 0:
            if "fatal: unable to access" in result.stderr:
                raise NetworkError(f"Network error cloning: {result.stderr}")
            elif "already exists and is not an empty directory" in result.stderr:
                raise GitError(f"Directory already exists: {target_dir}")
            else:
                raise GitError(f"Clone failed: {result.stderr}")
        return target_dir
    
    def recovery(error_type, error):
        if error_type == "git_conflict" and os.path.exists(target_dir):
            print(f"[RECOVERY] Removing existing directory {target_dir}")
            shutil.rmtree(target_dir)
    
    with CleanupManager(
        lambda: shutil.rmtree(target_dir) if os.path.exists(target_dir) else None,
        f"remove {target_dir}"
    ):
        return safe_operation_with_recovery(
            f"clone {repo_url}",
            do_clone,
            recovery
        )

Forgejo API Operations

def create_issue_with_retry(owner, repo, title, body):
    """Create a Forgejo issue with retry logic."""
    
    def do_create():
        try:
            issue = forgejo_create_issue(owner, repo, title, body)
            return issue
        except Exception as e:
            if "rate limit" in str(e).lower():
                raise ForgejoError(f"Rate limit exceeded: {e}")
            elif "network" in str(e).lower():
                raise NetworkError(f"Network error: {e}")
            else:
                raise ForgejoError(f"API error: {e}")
    
    return safe_operation_with_recovery(
        f"create issue '{title}'",
        do_create,
        max_retries=5  # More retries for API operations
    )

Work Claiming with Cleanup

def work_with_claim(owner, repo, issue_number, work_func):
    """Execute work with proper claim management."""
    
    claim_id = None
    
    def cleanup_claim():
        if claim_id:
            try:
                release_claim(owner, repo, issue_number, claim_id, "cleanup")
            except:
                pass  # Best effort
    
    with CleanupManager(cleanup_claim, "release work claim"):
        # Claim the work
        success, claim_id, error = claim_work_item(
            owner, repo, "issue", issue_number, "my-agent", "session-123"
        )
        
        if not success:
            raise AlreadyClaimedError(error)
        
        # Do the actual work
        try:
            result = work_func()
            # Success - release with completed status
            release_claim(owner, repo, issue_number, claim_id, "completed")
            return result
        except Exception as e:
            # Failure - release with failed status
            release_claim(owner, repo, issue_number, claim_id, "failed")
            raise

Complex Operation with Multiple Steps

def implement_feature_with_recovery(issue_number, subtasks):
    """Implement a feature with recovery for each step."""
    
    completed_subtasks = []
    clone_dir = f"/tmp/impl-{issue_number}"
    
    def cleanup_all():
        if os.path.exists(clone_dir):
            shutil.rmtree(clone_dir)
    
    with CleanupManager(cleanup_all, "remove working directory"):
        # Step 1: Clone
        safe_operation_with_recovery(
            "clone repository",
            lambda: git_clone(repo_url, clone_dir)
        )
        
        # Step 2: Implement each subtask
        for subtask in subtasks:
            try:
                safe_operation_with_recovery(
                    f"implement {subtask['title']}",
                    lambda: implement_subtask(subtask, clone_dir),
                    recovery_func=lambda t, e: rollback_subtask(subtask, clone_dir)
                )
                completed_subtasks.append(subtask)
            except OperationFailedError as e:
                print(f"[ERROR] Subtask {subtask['title']} failed: {e}")
                # Decide whether to continue or abort
                if subtask.get('required', True):
                    raise  # Abort on required subtask failure
        
        # Step 3: Commit and push
        if completed_subtasks:
            safe_operation_with_recovery(
                "commit and push changes",
                lambda: git_commit_and_push(clone_dir, completed_subtasks)
            )
        
        return completed_subtasks

Best Practices

1. Always use specific exception types - Never catch bare Exception unless re-raising
2. Include context in errors - Add relevant details (file paths, URLs, etc.)
3. Log before operations - So you know what was attempted if it fails
4. Use with statements - For automatic resource cleanup
5. Implement recovery - But only for errors you can actually recover from
6. Set appropriate timeouts - Don't wait forever for operations
7. Test error paths - Error handling code needs testing too

Anti-Patterns to Avoid

# ❌ WRONG - Swallowing all errors silently
try:
    do_something()
except:
    pass

# ✓ CORRECT - Specific handling with logging
try:
    do_something()
except NetworkError as e:
    print(f"[NETWORK ERROR] Operation failed: {e}")
    raise

# ❌ WRONG - No cleanup on error
temp_dir = create_temp_dir()
do_work(temp_dir)  # If this fails, temp_dir is leaked
cleanup(temp_dir)

# ✓ CORRECT - Guaranteed cleanup
with CleanupManager(lambda: cleanup(temp_dir)):
    do_work(temp_dir)

# ❌ WRONG - Infinite retry
while True:
    try:
        do_operation()
        break
    except:
        continue

# ✓ CORRECT - Limited retries with backoff
safe_operation_with_recovery("operation", do_operation, max_retries=3)

Integration with Agents

All agents should import and use these patterns:

from shared.error_handling import (
    safe_operation_with_recovery,
    CleanupManager,
    NetworkError,
    GitError,
    ForgejoError,
    OperationFailedError
)

# Then use throughout the agent code