error_handlers.py

#!/usr/bin/env python3
"""
Error handling utilities for HuggingClaw - Cain.
Provides specific exception handlers for common operations.
"""
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Optional
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

# CRITICAL: Set up sys.path for agents imports at module load time
# This ensures `from agents import brain_minimal` works regardless of import order
# Dynamic path resolution with fallbacks for different Docker contexts
_script_dir = Path(os.path.abspath(os.path.dirname(__file__)))  # Absolute path of this script

# Try multiple possible locations for .openclaw directory
_possible_openclaw_paths = [
    _script_dir / ".openclaw",           # /app/.openclaw (legacy/flat structure)
    _script_dir / "openclaw" / ".openclaw",  # /app/openclaw/.openclaw (nested structure)
    Path("/app/openclaw/.openclaw"),     # Absolute Docker path (nested)
    Path("/app/.openclaw"),              # Absolute Docker path (flat)
]

# Add all valid paths to sys.path
for path_dir in _possible_openclaw_paths:
    path_str = str(path_dir)
    if path_str not in sys.path and path_dir.exists():
        sys.path.insert(0, path_str)


async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
    """
    FastAPI exception handler for HTTPException.

    Args:
        request: The incoming request.
        exc: The HTTPException that was raised.

    Returns:
        JSONResponse with error details.
    """
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": True,
            "message": exc.detail,
            "status_code": exc.status_code,
            "agent": "cain"
        },
    )


async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """
    FastAPI exception handler for generic exceptions.

    Args:
        request: The incoming request.
        exc: The Exception that was raised.

    Returns:
        JSONResponse with error details.
    """
    return JSONResponse(
        status_code=500,
        content={
            "error": True,
            "message": "Internal server error",
            "detail": str(exc),
            "type": type(exc).__name__,
            "agent": "cain"
        },
    )


# Configuration path - cain_status.json location
def _get_base_dir() -> Path:
    """Determine the correct base directory dynamically."""
    # Priority 1: OPENCLAW_DATA_DIR (set by Docker environment)
    data_dir = os.environ.get('OPENCLAW_DATA_DIR')
    if data_dir:
        return Path(data_dir)

    # Priority 2: /data (standard Docker volume mount)
    data_path = Path("/data")
    if data_path.exists():
        return data_path

    # Priority 3: Use this script's directory as fallback
    return Path(os.path.dirname(__file__))


def _resolve_status_file() -> Path:
    """Find cain_status.json by searching common locations."""
    # Priority 1: CAIN_STATUS_PATH env var (set by app.py from OPENCLAW_DATA_DIR)
    env_path = os.environ.get('CAIN_STATUS_PATH')
    if env_path:
        return Path(env_path)

    # Priority 2: CAIN_STATUS_FILE env var (legacy)
    env_path = os.environ.get('CAIN_STATUS_FILE')
    if env_path:
        return Path(env_path)

    # Priority 3: OPENCLAW_DATA_DIR + cain_status.json
    data_dir = os.environ.get('OPENCLAW_DATA_DIR')
    if data_dir:
        data_path = Path(data_dir) / "cain_status.json"
        if data_path.exists():
            return data_path

    # Priority 4: /data/cain_status.json (Docker volume mount, default)
    data_path = Path("/data/cain_status.json")
    if data_path.exists():
        return data_path

    # Priority 5: /app/.openclaw/agents/cain_status.json OR /app/openclaw/.openclaw/agents/cain_status.json
    _script_dir = Path(os.path.abspath(os.path.dirname(__file__)))
    _possible_status_paths = [
        Path("/app/.openclaw/agents/cain_status.json"),         # Flat structure
        Path("/app/openclaw/.openclaw/agents/cain_status.json"), # Nested structure
        _script_dir / ".openclaw" / "agents" / "cain_status.json",
        _script_dir / "openclaw" / ".openclaw" / "agents" / "cain_status.json",
    ]
    for openclaw_status in _possible_status_paths:
        if openclaw_status.exists():
            return openclaw_status

    # Priority 6: /app/cain_status.json (legacy location)
    app_path = Path("/app/cain_status.json")
    if app_path.parent.exists():
        return app_path

    # Priority 6.5: Memory directory paths (CRITICAL: can have stale 'unknown' errors)
    _script_dir = Path(os.path.abspath(os.path.dirname(__file__)))
    _memory_paths = [
        Path("/app/memory/cain_status.json"),
        Path("/data/memory/cain_status.json"),
        _script_dir / "memory" / "cain_status.json",
    ]
    for mem_path in _memory_paths:
        if mem_path.exists():
            return mem_path

    # Priority 7: Script directory as final fallback
    script_dir = Path(os.path.dirname(__file__))
    return script_dir / "cain_status.json"

STATUS_FILE = _resolve_status_file()


def handle_status_file_read() -> Dict[str, Any]:
    """
    Handle reading Cain's status file with specific exception handling.

    Returns:
        Status dictionary with current_state, last_updated, and agent fields.
    """
    try:
        with open(str(STATUS_FILE), "r") as f:
            data = json.load(f)

        # CRITICAL: Clean stale "unknown" error strings immediately after reading
        # This prevents "Error: unknown" display issues
        error = data.get('error')
        if isinstance(error, str) and error.strip().lower() in ('unknown', 'none', 'null', ''):
            data['error'] = None
            data['_cleaned_at'] = 'handle_status_file_read'
            # Write back the cleaned data
            try:
                with open(str(STATUS_FILE), "w") as f:
                    json.dump(data, f, indent=2)
            except Exception:
                pass  # Don't fail if we can't write back

        return data
    except FileNotFoundError:
        # Status file not found is not an error - return healthy default
        return {
            "current_state": "idle",
            "stage": "STATUS_FILE_NOT_FOUND",
            "last_updated": datetime.utcnow().isoformat() + "+00:00",
            "agent": "cain",
            "error": None,
            "status_file": str(STATUS_FILE),
            "note": "Status file not found - using default"
        }
    except PermissionError as e:
        return {
            "current_state": "error",
            "stage": "PERMISSION_ERROR",
            "last_updated": datetime.utcnow().isoformat() + "+00:00",
            "agent": "cain",
            "error": f"Permission denied reading status file: {str(e)}",
            "status_file": str(STATUS_FILE)
        }
    except json.JSONDecodeError as e:
        return {
            "current_state": "error",
            "stage": "JSON_DECODE_ERROR",
            "last_updated": datetime.utcnow().isoformat() + "+00:00",
            "agent": "cain",
            "error": f"Invalid JSON in status file: {str(e)}",
            "status_file": str(STATUS_FILE)
        }


def handle_brain_response(message: str) -> str:
    """
    Handle brain processing with specific exception handling.

    Args:
        message: The user message to process.

    Returns:
        Response string from the brain or error message.

    Raises:
        ImportError: If brain module cannot be imported.
        AttributeError: If required brain methods are missing.
    """
    try:
        # sys.path is already set up at module level (see top of file)
        # Import agents directly - no need to import openclaw first
        from agents import brain_minimal

        # Verify brain has required method before calling
        if not hasattr(brain_minimal, 'get_brain'):
            return "Brain interface error: get_brain method not found"

        brain = brain_minimal.get_brain(agent_name="cain", legacy_mode=True)
        result = brain._conversation_process(message)

        if result.get("success"):
            return result.get("response", f"Processed: {message}")
        else:
            return f"Error: {result.get('error', 'Unknown error')}"

    except ImportError as e:
        return f"Brain module import error: {str(e)}"
    except AttributeError as e:
        return f"Brain interface error: {str(e)}"
    except KeyError as e:
        return f"Brain response format error: missing key {str(e)}"
    except Exception as e:
        return f"Brain processing error: {type(e).__name__}: {str(e)}"


async def handle_websocket_send(websocket, status_data: dict) -> bool:
    """
    Handle websocket send with specific exception handling.

    Args:
        websocket: The WebSocket connection object.
        status_data: Status data to send.

    Returns:
        True if send succeeded, False if connection should close.
    """
    try:
        await websocket.send_json({
            "type": "heartbeat",
            "status": status_data
        })
        return True
    except (ConnectionError, RuntimeError, Exception):
        return False


def write_cain_status(status_data: Dict[str, Any]) -> bool:
    """
    Write Cain's status to the status file.

    Args:
        status_data: Dictionary with status information to write.

    Returns:
        True if write succeeded, False otherwise.
    """
    try:
        # Ensure directory exists
        STATUS_FILE.parent.mkdir(parents=True, exist_ok=True)

        # Add timestamp if not present
        if "last_updated" not in status_data:
            status_data["last_updated"] = datetime.utcnow().isoformat() + "+00:00"

        # Write to file
        with open(str(STATUS_FILE), "w") as f:
            json.dump(status_data, f, indent=2)

        return True
    except (PermissionError, OSError, json.JSONDecodeError) as e:
        print(f"Error writing status file: {e}")
        return False