Update utils/async_runner.py

utils/async_runner.py

@@ -1,62 +1,147 @@
 """
-Async utilities for safe async/sync integration
+Enhanced Async Utilities for Safe Async/Sync Integration
+FIXED VERSION: Consolidated from app.py with nest_asyncio compatibility and better error handling
 """
 import asyncio
 import functools
-from typing import Any, Callable, Coroutine
+import time
 import logging
+from typing import Any, Callable, Coroutine, Optional, TypeVar, Union
+from contextlib import contextmanager
 
 logger = logging.getLogger(__name__)
+T = TypeVar('T')
 
 class AsyncRunner:
-    """Safely run async functions in sync context"""
+    """
+    Enhanced async runner with nest_asyncio compatibility and robust error handling
+    Consolidated from app.py local implementation
+    """
     
     @staticmethod
-    def run_async(coro: Coroutine) -> Any:
+    def run_async(coro: Coroutine[Any, Any, T], timeout: Optional[float] = 30.0) -> Union[T, dict]:
         """
-        Run async coroutine in sync context safely
+        FIXED: Run async coroutine in sync context with nest_asyncio compatibility
         
         Args:
             coro: Async coroutine to run
+            timeout: Maximum time to wait for coroutine completion (seconds)
             
         Returns:
-            Result of the coroutine
+            Result of the coroutine or error dictionary if failed
         """
+        start_time = time.time()
+        
         try:
-            # Try to get existing event loop
-            loop = asyncio.get_running_loop()
-            # We're already in async context - create task
-            logger.debug("Running in existing async context")
-            future = asyncio.create_task(coro)
-            return future
-        except RuntimeError:
-            # No running loop, create one
-            logger.debug("Creating new event loop for async execution")
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+            # Try to get running loop first (nest_asyncio compatible)
             try:
-                return loop.run_until_complete(coro)
-            finally:
-                loop.close()
+                loop = asyncio.get_running_loop()
+                logger.debug("✅ Running in existing async context (nest_asyncio detected)")
+                
+                # In a running loop, we need to schedule as a task
+                # This handles the "event loop already running" case
+                task = asyncio.create_task(coro)
+                
+                # For sync context, we need to run until complete
+                # Use asyncio.run_coroutine_threadsafe for thread safety
+                import concurrent.futures
+                future = asyncio.run_coroutine_threadsafe(coro, loop)
+                
+                try:
+                    result = future.result(timeout=timeout)
+                    logger.debug(f"✅ Async execution completed in {time.time() - start_time:.2f}s")
+                    return result
+                except concurrent.futures.TimeoutError:
+                    logger.error(f"❌ Async execution timed out after {timeout}s")
+                    future.cancel()
+                    return {
+                        "error": f"Async execution timed out after {timeout}s",
+                        "status": "failed",
+                        "timeout": True,
+                        "boundary_note": "Execution boundary reached - timeout"
+                    }
+                    
+            except RuntimeError:
+                # No running loop, create one
+                logger.debug("🔄 Creating new event loop for async execution")
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                
+                try:
+                    result = loop.run_until_complete(asyncio.wait_for(coro, timeout=timeout))
+                    logger.debug(f"✅ Async execution completed in {time.time() - start_time:.2f}s")
+                    return result
+                except asyncio.TimeoutError:
+                    logger.error(f"❌ Async execution timed out after {timeout}s")
+                    return {
+                        "error": f"Async execution timed out after {timeout}s",
+                        "status": "failed",
+                        "timeout": True,
+                        "boundary_note": "Execution boundary reached - timeout"
+                    }
+                finally:
+                    # Clean up the loop
+                    if not loop.is_closed():
+                        loop.close()
+            
+        except Exception as e:
+            logger.error(f"❌ Async execution failed: {e}", exc_info=True)
+            return {
+                "error": str(e),
+                "status": "failed",
+                "execution_time": time.time() - start_time,
+                "boundary_note": "Execution boundary reached",
+                "error_type": type(e).__name__
+            }
     
     @staticmethod
-    def sync_wrapper(async_func: Callable) -> Callable:
+    def async_to_sync(async_func: Callable[..., Coroutine[Any, Any, T]]) -> Callable[..., Union[T, dict]]:
         """
-        Decorator to make async function callable from sync context
+        FIXED: Decorator to convert async function to sync with enhanced error handling
         
-        Args:
-            async_func: Async function to wrap
-            
-        Returns:
-            Sync-compatible function
+        Usage:
+            @AsyncRunner.async_to_sync
+            async def my_async_function():
+                ...
+                
+            # Can now be called synchronously
+            result = my_async_function()
         """
         @functools.wraps(async_func)
-        def wrapper(*args, **kwargs) -> Any:
-            return AsyncRunner.run_async(async_func(*args, **kwargs))
+        def wrapper(*args, **kwargs) -> Union[T, dict]:
+            try:
+                # Create the coroutine
+                coro = async_func(*args, **kwargs)
+                
+                # Run it with timeout
+                return AsyncRunner.run_async(coro)
+                
+            except Exception as e:
+                logger.error(f"❌ Async to sync conversion failed: {e}", exc_info=True)
+                return {
+                    "error": str(e),
+                    "status": "failed",
+                    "boundary_context": "OSS advisory only - execution requires Enterprise",
+                    "error_type": type(e).__name__
+                }
         return wrapper
+    
+    @staticmethod
+    def is_async_context() -> bool:
+        """
+        Check if we're currently in an async context
+        
+        Returns:
+            True if in async context, False otherwise
+        """
+        try:
+            asyncio.get_running_loop()
+            return True
+        except RuntimeError:
+            return False
 
-
-def async_to_sync(async_func: Callable) -> Callable:
+# Convenience function for the decorator
+def async_to_sync(async_func: Callable[..., Coroutine[Any, Any, T]]) -> Callable[..., Union[T, dict]]:
     """
     Convenience decorator to convert async function to sync
     
@@ -68,32 +153,91 @@ def async_to_sync(async_func: Callable) -> Callable:
         # Can now be called synchronously
         result = my_async_function()
     """
-    return AsyncRunner.sync_wrapper(async_func)
+    return AsyncRunner.async_to_sync(async_func)
 
 
-class SafeEventLoop:
-    """Context manager for safe event loop handling"""
+@contextmanager
+def safe_event_loop():
+    """
+    Context manager for safe event loop handling with cleanup
+    
+    Usage:
+        with safe_event_loop() as loop:
+            result = loop.run_until_complete(async_function())
+    """
+    loop = None
+    try:
+        # Try to get existing loop
+        try:
+            loop = asyncio.get_running_loop()
+            logger.debug("Using existing event loop")
+            yield loop
+            return
+        except RuntimeError:
+            pass
+        
+        # Create new loop
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        logger.debug("Created new event loop")
+        yield loop
+        
+    finally:
+        # Cleanup
+        if loop and not loop.is_closed():
+            loop.close()
+            logger.debug("Closed event loop")
+
+
+class AsyncCircuitBreaker:
+    """
+    Circuit breaker pattern for async operations to prevent cascading failures
+    """
     
-    def __init__(self, create_new: bool = False):
-        self.create_new = create_new
-        self.loop = None
-        self.original_loop = None
+    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
+        self.failure_threshold = failure_threshold
+        self.recovery_timeout = recovery_timeout
+        self.failure_count = 0
+        self.last_failure_time = 0
+        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
+        
+    def can_execute(self) -> bool:
+        """Check if circuit breaker allows execution"""
+        if self.state == "OPEN":
+            # Check if recovery timeout has passed
+            if time.time() - self.last_failure_time > self.recovery_timeout:
+                self.state = "HALF_OPEN"
+                logger.info("Circuit breaker moving to HALF_OPEN state")
+                return True
+            return False
+        return True
     
-    def __enter__(self):
-        if self.create_new:
-            self.original_loop = asyncio.get_event_loop_policy().get_event_loop()
-            self.loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(self.loop)
-        else:
-            try:
-                self.loop = asyncio.get_running_loop()
-            except RuntimeError:
-                self.loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(self.loop)
-        return self.loop
+    def record_success(self):
+        """Record successful execution"""
+        if self.state == "HALF_OPEN":
+            self.state = "CLOSED"
+            logger.info("Circuit breaker reset to CLOSED state")
+        self.failure_count = 0
+    
+    def record_failure(self):
+        """Record failed execution"""
+        self.failure_count += 1
+        self.last_failure_time = time.time()
+        
+        if self.failure_count >= self.failure_threshold:
+            self.state = "OPEN"
+            logger.warning(f"Circuit breaker OPENED after {self.failure_count} failures")
     
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self.create_new and self.original_loop:
-            asyncio.set_event_loop(self.original_loop)
-            if not self.loop.is_closed():
-                self.loop.close()
+    @AsyncRunner.async_to_sync
+    async def execute(self, coro: Coroutine) -> Any:
+        """Execute async operation with circuit breaker protection"""
+        if not self.can_execute():
+            raise Exception("Circuit breaker is OPEN - operation blocked")
+        
+        try:
+            result = await coro
+            self.record_success()
+            return result
+        except Exception as e:
+            self.record_failure()
+            raise e