fix: Core SDK - resolve unsafe async patterns and global mutable state (#1450)

* fix: replace unsafe async callback dispatch and module-level mutable globals

- Replace unsafe asyncio.run() inside running event loops with run_coroutine_safely() utility
- Replace module-level mutable globals with thread-safe _AgentServerRegistry class
- Fixes race conditions in concurrent launch() calls
- Addresses issue #1449 points 2 and 3

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>

* fix: replace unsafe asyncio.Lock() creation with async context method

- Add _get_state_lock() method to create asyncio.Lock within async context
- Fixes unsafe lazy lock creation that could bind to wrong event loop
- Addresses issue #1449 point 4 (asyncio.Lock creation issue)

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>

* fix: make agent server route registration atomic and lock-safe

Agent-Logs-Url: https://github.com/MervinPraison/PraisonAI/sessions/b3b60c87-cf60-4e59-8852-07721eb9e9f6

Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>

* docs: clarify reserve_route return tuple semantics

Agent-Logs-Url: https://github.com/MervinPraison/PraisonAI/sessions/b3b60c87-cf60-4e59-8852-07721eb9e9f6

Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>

* fix: resolve critical server startup and thread-safety issues

- Fix uvicorn.Config duplicate log_level parameter causing TypeError
- Add thread-safe accessor methods (has_route, list_routes) to registry
- Move hot-path run_coroutine_safely import to module scope
- Add proper type annotations and noqa comment for host parameter
- Hoist threading import to module level for consistency

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>

Author: 3 people

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -2,8 +2,9 @@
 import time
 import json
 import logging
+import threading
 from praisonaiagents._logging import get_logger
-from typing import Any, Dict, Optional, List
+from typing import Any, Dict, Optional, List, Tuple
 from ..main import display_error, TaskOutput
 from ..agent.agent import Agent
 from ..task.task import Task
@@ -18,6 +19,12 @@
 except ImportError:
     _token_collector = None
 
+# Import async utility for hot-path usage
+try:
+    from ..approval.utils import run_coroutine_safely
+except ImportError:
+    run_coroutine_safely = None
+
 # Task status constants
 class TaskStatus(Enum):
     """Enumeration for task status values to ensure consistency"""
@@ -30,12 +37,101 @@ class TaskStatus(Enum):
 # Set up logger
 logger = get_logger(__name__)
 
-# Global variables for managing the shared servers with thread-safety
-import threading
-_agents_server_lock = threading.Lock()  # Protect all global server state mutations
-_agents_server_started = {}  # Dict of port -> started boolean
-_agents_registered_endpoints = {}  # Dict of port -> Dict of path -> endpoint_id
-_agents_shared_apps = {}  # Dict of port -> FastAPI app
+# Agent server registry for thread-safe server management
+
+
+class _AgentServerRegistry:
+    """Encapsulates all shared HTTP server state with proper synchronization."""
+    
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._started: Dict[int, bool] = {}
+        self._endpoints: Dict[int, Dict[str, str]] = {}
+        self._apps: Dict[int, Any] = {}  # FastAPI apps
+        self._ready_events: Dict[int, threading.Event] = {}
+    
+    def get_or_create_app(self, port: int, title: str = "AgentTeam API") -> Tuple[Any, bool]:
+        """Thread-safe app creation. Returns (app, is_new)."""
+        with self._lock:
+            if port not in self._apps:
+                # Lazy import to avoid optional dependency at module level
+                from fastapi import FastAPI
+                self._apps[port] = FastAPI(title=title)
+                self._endpoints[port] = {}
+                self._ready_events[port] = threading.Event()
+                return self._apps[port], True
+            return self._apps[port], False
+    
+    def register_route(self, port: int, path: str, endpoint_id: str = "registered") -> None:
+        """Thread-safe route registration tracking."""
+        with self._lock:
+            if port not in self._endpoints:
+                self._endpoints[port] = {}
+            self._endpoints[port][path] = endpoint_id
+
+    def reserve_route(self, port: int, path: str, endpoint_id: str) -> tuple[str, Optional[str]]:
+        """Atomically reserve and register a route.
+
+        Returns:
+            tuple[str, Optional[str]]: (final_path, original_path_if_collided).
+            If there is no collision, final_path equals the requested path and
+            original_path_if_collided is None.
+        """
+        with self._lock:
+            if port not in self._endpoints:
+                self._endpoints[port] = {}
+
+            original_path = path
+            while path in self._endpoints[port]:
+                path = f"{original_path}_{str(uuid.uuid4())[:6]}"
+
+            self._endpoints[port][path] = endpoint_id
+            return path, (original_path if path != original_path else None)
+
+    def list_routes(self, port: int) -> List[str]:
+        """Return a snapshot list of registered routes for a port."""
+        with self._lock:
+            return list(self._endpoints.get(port, {}).keys())
+    
+    def has_route(self, port: int, path: str) -> bool:
+        """Thread-safe check whether a path is registered on a port."""
+        with self._lock:
+            return path in self._endpoints.get(port, {})
+    
+    def is_server_started(self, port: int) -> bool:
+        """Check if server is started for this port."""
+        with self._lock:
+            return self._started.get(port, False)
+    
+    def start_server_if_needed(self, port: int, host: str = "0.0.0.0", **kwargs) -> bool:  # noqa: S104
+        """Start server with proper readiness signaling. Returns True if server was started."""
+        with self._lock:
+            if self._started.get(port, False):
+                return False  # Already started
+            self._started[port] = True
+            app = self._apps.get(port)
+            
+        if not app:
+            raise ValueError(f"No app registered for port {port}")
+        
+        ready_event = self._ready_events[port]
+        
+        def run_server():
+            import uvicorn
+            # Remove hardcoded log_level to avoid conflict with kwargs
+            config = uvicorn.Config(app, host=host, port=port, **kwargs)
+            server = uvicorn.Server(config)
+            ready_event.set()  # Signal readiness
+            server.run()
+        
+        thread = threading.Thread(target=run_server, daemon=True)
+        thread.start()
+        ready_event.wait(timeout=5.0)  # Deterministic wait instead of sleep(0.5)
+        return True
+
+
+# Module level — single registry instance
+_server_registry = _AgentServerRegistry()
 
 def encode_file_to_base64(file_path: str) -> str:
     """Base64-encode a file."""
@@ -933,12 +1029,10 @@ async def arun_task(self, task_id):
                     if task.callback:
                         try:
                             if asyncio.iscoroutinefunction(task.callback):
-                                try:
-                                    loop = asyncio.get_running_loop()
-                                    loop.create_task(task.callback(task_output))
-                                except RuntimeError:
-                                    # No event loop running, create new one
-                                    asyncio.run(task.callback(task_output))
+                                if run_coroutine_safely:
+                                    run_coroutine_safely(task.callback(task_output))
+                                else:
+                                    logger.warning("run_coroutine_safely not available, skipping async callback")
                             else:
                                 task.callback(task_output)
                         except Exception as e:
@@ -1164,12 +1258,10 @@ def run_task(self, task_id):
                     if task.callback:
                         try:
                             if asyncio.iscoroutinefunction(task.callback):
-                                try:
-                                    loop = asyncio.get_running_loop()
-                                    loop.create_task(task.callback(task_output))
-                                except RuntimeError:
-                                    # No event loop running, create new one
-                                    asyncio.run(task.callback(task_output))
+                                if run_coroutine_safely:
+                                    run_coroutine_safely(task.callback(task_output))
+                                else:
+                                    logger.warning("run_coroutine_safely not available, skipping async callback")
                             else:
                                 task.callback(task_output)
                         except Exception as e:
@@ -1643,7 +1735,7 @@ def launch(self, path: str = '/agents', port: int = 8000, host: str = '0.0.0.0',
             None
         """
         if protocol == "http":
-            global _agents_server_started, _agents_registered_endpoints, _agents_shared_apps
+            # Use centralized server registry
 
             if not self.agents:
                 logging.warning("No agents to launch for HTTP mode. Add agents to the Agents instance first.")
@@ -1674,58 +1766,43 @@ class AgentQuery(BaseModel):
                 print("pip install 'praisonaiagents[api]'")
                 return None
 
-            # Thread-safe initialization of port-specific collections
-            with _agents_server_lock:
-                # Initialize port-specific collections if needed
-                if port not in _agents_registered_endpoints:
-                    _agents_registered_endpoints[port] = {}
-                    
-                # Initialize shared FastAPI app if not already created for this port
-                if _agents_shared_apps.get(port) is None:
-                    _agents_shared_apps[port] = FastAPI(
-                        title=f"PraisonAI Agents API (Port {port})",
-                        description="API for interacting with multiple PraisonAI Agents"
-                    )
-                
+            # Thread-safe initialization of FastAPI app
+            app, is_new = _server_registry.get_or_create_app(
+                port, f"PraisonAI Agents API (Port {port})"
+            )
+            
+            if is_new:
                 # Add a root endpoint with a welcome message
-                @_agents_shared_apps[port].get("/")
+                @app.get("/")
                 async def root():
                     return {
                         "message": f"Welcome to PraisonAI Agents API on port {port}. See /docs for usage.",
-                        "endpoints": list(_agents_registered_endpoints[port].keys())
+                        "endpoints": _server_registry.list_routes(port)
                     }
 
                 # Add healthcheck endpoint
-                @_agents_shared_apps[port].get("/health")
+                @app.get("/health")
                 async def healthcheck():
                     return {
                         "status": "ok", 
-                        "endpoints": list(_agents_registered_endpoints[port].keys())
+                        "endpoints": _server_registry.list_routes(port)
                     }
 
             # Normalize path to ensure it starts with /
             if not path.startswith('/'):
                 path = f'/{path}'
 
-            # Thread-safe path registration 
-            with _agents_server_lock:
-                # Check if path is already registered for this port
-                if path in _agents_registered_endpoints[port]:
-                    logging.warning(f"Path '{path}' is already registered on port {port}. Please use a different path.")
-                    print(f"⚠️ Warning: Path '{path}' is already registered on port {port}.")
-                    # Use a modified path to avoid conflicts
-                    original_path = path
-                    instance_id = str(uuid.uuid4())[:6]
-                    path = f"{path}_{instance_id}"
-                    logging.warning(f"Using '{path}' instead of '{original_path}'")
-                    print(f"🔄 Using '{path}' instead")
-                
-                # Generate a unique ID for this agent group's endpoint
-                endpoint_id = str(uuid.uuid4())
-                _agents_registered_endpoints[port][path] = endpoint_id
+            # Generate a unique ID for this agent group's endpoint and reserve route atomically
+            endpoint_id = str(uuid.uuid4())
+            path, original_path = _server_registry.reserve_route(port, path, endpoint_id)
+            if original_path is not None:
+                logging.warning(f"Path '{original_path}' is already registered on port {port}. Please use a different path.")
+                print(f"⚠️ Warning: Path '{original_path}' is already registered on port {port}.")
+                logging.warning(f"Using '{path}' instead of '{original_path}'")
+                print(f"🔄 Using '{path}' instead")
 
             # Define the endpoint handler
-            @_agents_shared_apps[port].post(path)
+            @app.post(path)
             async def handle_query(request: Request, query_data: Optional[AgentQuery] = None):
                 # Handle both direct JSON with query field and form data
                 if query_data is None:
@@ -1804,7 +1881,7 @@ async def handle_query(request: Request, query_data: Optional[AgentQuery] = None
             agents_dict = {agent.display_name.lower().replace(' ', '_'): agent for agent in self.agents}
 
             # Add GET endpoint to list available agents
-            @_agents_shared_apps[port].get(f"{path}/list")
+            @app.get(f"{path}/list")
             async def list_agents():
                 return {
                     "agents": [
@@ -1850,45 +1927,19 @@ async def handle_single_agent(request: Request):
                             )
                     return handle_single_agent
 
-                # Register the endpoint with thread safety
-                _agents_shared_apps[port].post(agent_path)(create_agent_handler(agent_instance))
-                with _agents_server_lock:
-                    _agents_registered_endpoints[port][agent_path] = f"{endpoint_id}_{agent_id}"
+                # Register the endpoint
+                app.post(agent_path)(create_agent_handler(agent_instance))
+                _server_registry.register_route(port, agent_path, f"{endpoint_id}_{agent_id}")
 
             print(f"🔗 Per-agent endpoints: {', '.join([f'{path}/{aid}' for aid in agents_dict.keys()])}")
 
             # Start the server if it's not already running for this port
-            with _agents_server_lock:
-                if not _agents_server_started.get(port, False):
-                    # Mark the server as started first to prevent duplicate starts
-                    _agents_server_started[port] = True
-                    should_start_server = True
-                else:
-                    should_start_server = False
-            
-            if should_start_server:
+            if _server_registry.start_server_if_needed(port, host, log_level="debug" if debug else "info"):
+                print(f"✅ FastAPI server started at http://{host}:{port}")
+                print(f"📚 API documentation available at http://{host}:{port}/docs")
 
-                # Start the server in a separate thread
-                def run_server():
-                    try:
-                        print(f"✅ FastAPI server started at http://{host}:{port}")
-                        print(f"📚 API documentation available at http://{host}:{port}/docs")
-                        print(f"🔌 Registered HTTP endpoints on port {port}: {', '.join(list(_agents_registered_endpoints[port].keys()))}")
-                        uvicorn.run(_agents_shared_apps[port], host=host, port=port, log_level="debug" if debug else "info")
-                    except Exception as e:
-                        logging.error(f"Error starting server: {str(e)}", exc_info=True)
-                        print(f"❌ Error starting server: {str(e)}")
-                
-                # Run server in a background thread
-                server_thread = threading.Thread(target=run_server, daemon=True)
-                server_thread.start()
-                
-                # Wait for a moment to allow the server to start and register endpoints
-                time.sleep(0.5)
-            else:
-                # If server is already running, wait a moment to make sure the endpoint is registered
-                time.sleep(0.1)
-                print(f"🔌 Registered HTTP endpoints on port {port}: {', '.join(list(_agents_registered_endpoints[port].keys()))}")
+            endpoints = _server_registry.list_routes(port)
+            print(f"🔌 Registered HTTP endpoints on port {port}: {', '.join(endpoints)}")
 
             # Get the stack frame to check if this is the last launch() call in the script
             import inspect

src/praisonai-agents/praisonaiagents/process/process.py

@@ -46,7 +46,7 @@ def __init__(
         self.task_retry_counter: Dict[str, int] = {} # Initialize retry counter
         self.workflow_finished = False # ADDED: Workflow finished flag
         self.workflow_cancelled = False # ADDED: Workflow cancellation flag for timeout
-        self._state_lock_init = threading.Lock()  # Thread lock for async lock creation
+        self._state_lock_init = threading.Lock()  # Thread lock for synchronous shared-state updates
         self._state_lock = None # Lazy-initialized async lock for shared state protection
 
         # Resolve verbose from output= param (takes precedence) or legacy verbose= param
@@ -64,6 +64,12 @@ def __init__(
 
         logging.debug(f"Verbose mode: {self._verbose}")
 
+    async def _get_state_lock(self):
+        """Get or create the async state lock (must be called from async context)."""
+        if self._state_lock is None:
+            self._state_lock = asyncio.Lock()
+        return self._state_lock
+
     def _create_llm_instance(self):
         """Create and return a configured LLM instance for manager tasks."""
         return LLM(model=self.manager_llm, temperature=0.7)
@@ -608,12 +614,9 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
                     break
 
             # Reset completed task to "not started" so it can run again (atomic operation)
-            # Atomic state lock initialization
-            if self._state_lock is None:
-                with self._state_lock_init:
-                    if self._state_lock is None:  # Double-checked locking pattern
-                        self._state_lock = asyncio.Lock()
-            async with self._state_lock:
+            # Get async state lock (created within async context)
+            lock = await self._get_state_lock()
+            async with lock:
                 if self.tasks[task_id].status == "completed":
                     # Never reset loop tasks, decision tasks, or their subtasks if rerun is False
                     subtask_name = self.tasks[task_id].name
@@ -1547,4 +1550,4 @@ class ManagerInstructions(BaseModel):
         self.tasks[manager_task.id].status = "completed"
         if self.verbose >= 1:
             logging.info("All tasks completed under manager supervision.")
-        logging.info("Hierarchical task execution finished")
+        logging.info("Hierarchical task execution finished")