logging_utils/core/error_handler.py

# -*- coding: utf-8 -*-
"""
Centralized Error Handling Utilities
=====================================
Provides consistent error logging, global exception handlers, and
optional error snapshot integration for robust debugging.
"""

import asyncio
import logging
import threading
from typing import Optional

from .context import request_id_var, source_var


def log_error(
    logger: logging.Logger,
    message: str,
    exception: Optional[BaseException] = None,
    *,
    save_snapshot: bool = False,
    req_id: str = "",
    exc_info: bool = True,
) -> None:
    """
    Centralized error logging with automatic stack trace capture.

    This function ensures consistent error logging across the codebase:
    - Always includes exc_info for full stack traces
    - Optionally saves error snapshots for debugging
    - Integrates with request context for tracing

    Args:
        logger: Logger instance to use
        message: Error message to log
        exception: Optional exception object (used for snapshot)
        save_snapshot: Whether to save an error snapshot (screenshot, DOM, etc.)
        req_id: Optional request ID override (uses context var if empty)
        exc_info: Whether to include exception info (default True)

    Usage:
        try:
            risky_operation()
        except Exception as e:
            log_error(logger, f"Operation failed: {e}", e, save_snapshot=True)
    """
    # Get request ID from context if not provided
    if not req_id:
        try:
            req_id = request_id_var.get()
        except LookupError:
            req_id = "unknown"

    # Log with exc_info for full stack trace
    logger.error(message, exc_info=exc_info)

    # Save error snapshot if requested
    if save_snapshot:
        try:
            # Lazy import to avoid circular dependencies
            from browser_utils.debug_utils import save_error_snapshot_enhanced

            # Generate error name from message (first 30 chars, sanitized)
            error_name = (
                message[:30].replace(" ", "_").replace(":", "").replace("/", "_")
            )
            # This is async, but we're in sync context - schedule it
            try:
                loop = asyncio.get_running_loop()
                loop.create_task(
                    save_error_snapshot_enhanced(
                        error_name=error_name,
                        error_exception=exception
                        if isinstance(exception, Exception)
                        else None,
                        error_stage="log_error",
                    )
                )
            except RuntimeError:
                # No running event loop - can't save async snapshot
                logger.debug(
                    "[Snapshot] Cannot save error snapshot: no running event loop"
                )
        except ImportError:
            # debug_utils not available (e.g., in tests or standalone mode)
            pass
        except Exception as snapshot_err:
            # Don't let snapshot failures break the main error handling
            logger.debug(f"[Snapshot] Failed to save error snapshot: {snapshot_err}")


def _asyncio_exception_handler(loop: asyncio.AbstractEventLoop, context: dict) -> None:
    """
    Global asyncio exception handler for uncaught exceptions in tasks.

    This catches exceptions that would otherwise be silently ignored when:
    - A Task raises an exception but is never awaited
    - A callback raises an exception

    Args:
        loop: The event loop that caught the exception
        context: Exception context dict with 'message', 'exception', etc.
    """
    # Skip logging during Python shutdown to avoid ImportError crashes
    import sys

    if sys.meta_path is None or sys.modules is None:
        return

    # Also check if logging module is still available
    try:
        import logging as _check

        del _check
    except (ImportError, TypeError):
        return

    logger = logging.getLogger("AIStudioProxyServer")

    # Extract exception info
    exception = context.get("exception")
    message = context.get("message", "Unhandled exception in asyncio task")

    # Get additional context
    task = context.get("task")
    future = context.get("future")

    # Build detailed error message
    error_parts = [f"[ASYNCIO EXCEPTION] {message}"]

    if task is not None:
        error_parts.append(
            f"Task: {task.get_name() if hasattr(task, 'get_name') else repr(task)}"
        )

    if future is not None and future is not task:
        error_parts.append(f"Future: {repr(future)}")

    # Get source from context if available
    try:
        source = source_var.get()
        error_parts.append(f"Source: {source}")
    except LookupError:
        pass

    try:
        req_id = request_id_var.get()
        if req_id.strip():
            error_parts.append(f"Request ID: {req_id}")
    except LookupError:
        pass

    full_message = " | ".join(error_parts)

    if exception is not None:
        # Log with full traceback
        logger.error(
            full_message, exc_info=(type(exception), exception, exception.__traceback__)
        )
    else:
        logger.error(full_message)


def _threading_exception_handler(args: threading.ExceptHookArgs) -> None:
    """
    Global threading exception handler for uncaught exceptions in threads.

    Args:
        args: ExceptHookArgs containing exc_type, exc_value, exc_traceback, thread
    """
    logger = logging.getLogger("AIStudioProxyServer")

    thread_name = args.thread.name if args.thread else "unknown"

    error_message = f"[THREAD EXCEPTION] Uncaught exception in thread '{thread_name}'"

    if args.exc_value is not None:
        logger.error(
            error_message,
            exc_info=(args.exc_type, args.exc_value, args.exc_traceback),
        )
    else:
        logger.error(error_message)


def setup_global_exception_handlers(
    *, install_asyncio: bool = True, install_threading: bool = True
) -> None:
    """
    Install global exception handlers for asyncio and threading.

    This ensures that uncaught exceptions in background tasks and threads
    are properly logged instead of being silently ignored.

    Should be called once during application startup.

    Args:
        install_asyncio: Install asyncio exception handler
        install_threading: Install threading exception handler

    Usage:
        # In application startup
        setup_global_exception_handlers()
    """
    logger = logging.getLogger("AIStudioProxyServer")

    if install_asyncio:
        try:
            loop = asyncio.get_running_loop()
            loop.set_exception_handler(_asyncio_exception_handler)
            logger.debug("[Init] Global exception handlers installed (Asyncio)")
        except RuntimeError:
            # No running event loop yet - will be installed when loop starts
            # This is common during module import
            pass

    if install_threading:
        # Python 3.8+
        if hasattr(threading, "excepthook"):
            threading.excepthook = _threading_exception_handler
            logger.debug("[Init] Global exception handlers installed (Threading)")


def install_asyncio_handler_on_loop(loop: asyncio.AbstractEventLoop) -> None:
    """
    Install the asyncio exception handler on a specific event loop.

    Use this when you need to install the handler after the loop is created.

    Args:
        loop: The event loop to install the handler on
    """
    loop.set_exception_handler(_asyncio_exception_handler)