fix: Add asyncio CancelledError handling and graceful shutdown

- Add SIGTERM signal handler for graceful shutdown coordination
- Wrap initialization in try-except for CancelledError with stack trace logging
- Add 2-second timeout wrappers to async operations (state store, conversation storage)
- Add global shutdown flag to prevent race conditions during shutdown
- Update shutdown() method to use timeout wrappers

This prevents asyncio CancelledError noise during container shutdown when
background threads from state_manager and persistence are still using the
event loop.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.openclaw/agents/brain_minimal.py

@@ -16,9 +16,17 @@ from datetime import datetime
 from enum import Enum
 import threading
 import logging
+import signal
+import sys
+import traceback
+import asyncio
 
 logger = logging.getLogger(__name__)
 
+# Global shutdown flag for graceful shutdown
+_shutdown_in_progress = False
+_shutdown_lock = threading.Lock()
+
 # Import core modules for personality and formatting
 try:
     from .core.personality import get_personality, get_system_prompt
@@ -293,6 +301,92 @@ class BrainState(Enum):
     ERROR = "error"
 
 
+# ========== Signal Handler for Graceful Shutdown ==========
+
+def _sigterm_handler(signum, frame):
+    """
+    Handle SIGTERM signal for graceful shutdown.
+
+    This ensures all background tasks are properly closed before
+    the event loop closes, preventing CancelledError noise.
+    """
+    global _shutdown_in_progress
+    with _shutdown_lock:
+        if _shutdown_in_progress:
+            return  # Already shutting down
+        _shutdown_in_progress = True
+
+    logger.info(f"[SIGNAL] Received signal {signum}, initiating graceful shutdown...")
+
+    # Get the current brain instance and trigger shutdown
+    try:
+        global _global_brain_instance
+        if _global_brain_instance is not None:
+            logger.info("[SIGNAL] Shutting down brain gracefully...")
+            _global_brain_instance.shutdown()
+    except Exception as e:
+        logger.error(f"[SIGNAL] Error during brain shutdown: {e}")
+
+    # Also shutdown state store and conversation storage if available
+    try:
+        if STATE_PERSISTENCE_AVAILABLE:
+            from core.state_manager import get_state_store
+            state_store = get_state_store()
+            if state_store:
+                state_store.stop()
+                logger.info("[SIGNAL] State store stopped")
+    except Exception as e:
+        logger.warning(f"[SIGNAL] Error stopping state store: {e}")
+
+    try:
+        if CONVERSATION_PERSISTENCE_AVAILABLE:
+            from core.persistence import get_conversation_storage
+            storage = get_conversation_storage()
+            if storage:
+                storage.stop()
+                logger.info("[SIGNAL] Conversation storage stopped")
+    except Exception as e:
+        logger.warning(f"[SIGNAL] Error stopping conversation storage: {e}")
+
+    logger.info("[SIGNAL] Graceful shutdown complete")
+
+# Register signal handler for SIGTERM
+try:
+    signal.signal(signal.SIGTERM, _sigterm_handler)
+    logger.info("[INIT] SIGTERM handler registered for graceful shutdown")
+except ValueError:
+    # Signal handlers can only be set from main thread
+    logger.debug("[INIT] Could not set SIGTERM handler from non-main thread")
+
+
+def with_http_timeout(func):
+    """
+    Decorator to add 2-second timeout to HTTP calls.
+
+    This prevents hanging HTTP operations during shutdown.
+    """
+    def wrapper(*args, **kwargs):
+        import concurrent.futures
+        import time
+
+        def _execute():
+            return func(*args, **kwargs)
+
+        try:
+            # Use ThreadPoolExecutor with 2-second timeout
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(_execute)
+                return future.result(timeout=2.0)
+        except concurrent.futures.TimeoutError:
+            logger.warning(f"[HTTP_TIMEOUT] Operation {func.__name__} timed out after 2 seconds")
+            return None
+        except Exception as e:
+            logger.error(f"[HTTP_ERROR] Operation {func.__name__} failed: {e}")
+            raise
+
+    return wrapper
+
+
 class Tool:
     """Represents a tool that can be executed by the brain"""
 
@@ -346,52 +440,101 @@ class BrainMinimal:
             agent_name: Name of the agent (adam, eve, cain)
             legacy_mode: If True, operate in single-agent mode
         """
-        self.agent_name = agent_name or self._detect_agent()
-
-        # Load personality for this agent
-        self.personality = get_personality(self.agent_name) if CORE_MODULES_AVAILABLE else {"name": self.agent_name, "role": "Agent"}
-        self.legacy_mode = legacy_mode or not RBAC_AVAILABLE
-
-        # Initialize RBAC system
-        if RBAC_AVAILABLE and not self.legacy_mode:
-            self.rbac = MultiAgentSystem(legacy_mode=False)
-            self.current_role = self.rbac.current_role
-        else:
-            self.rbac = None
-            self.current_role = None
-            self.legacy_mode = True
+        # Wrap entire initialization in try-except for asyncio.CancelledError
+        try:
+            self.agent_name = agent_name or self._detect_agent()
+
+            # Load personality for this agent
+            self.personality = get_personality(self.agent_name) if CORE_MODULES_AVAILABLE else {"name": self.agent_name, "role": "Agent"}
+            self.legacy_mode = legacy_mode or not RBAC_AVAILABLE
+
+            # Initialize RBAC system
+            if RBAC_AVAILABLE and not self.legacy_mode:
+                self.rbac = MultiAgentSystem(legacy_mode=False)
+                self.current_role = self.rbac.current_role
+            else:
+                self.rbac = None
+                self.current_role = None
+                self.legacy_mode = True
+
+            # Initialize State persistence with timeout wrapper
+            self.state_store: Optional[StateStore] = None
+            if STATE_PERSISTENCE_AVAILABLE:
+                try:
+                    # Use timeout wrapper to prevent hanging during shutdown
+                    self.state_store = self._init_with_timeout(get_state_store, 2.0)
+                    logger.info(f"[{self.agent_name}] State persistence initialized")
+                except asyncio.CancelledError:
+                    logger.info(f"[{self.agent_name}] State persistence initialization cancelled (shutdown in progress)")
+                    raise
+                except Exception as e:
+                    logger.warning(f"[{self.agent_name}] State persistence initialization failed: {e}")
+
+            # Initialize Conversation persistence with timeout wrapper
+            self.conversation_storage: Optional[ConversationStorage] = None
+            if CONVERSATION_PERSISTENCE_AVAILABLE:
+                try:
+                    # Use timeout wrapper to prevent hanging during shutdown
+                    self.conversation_storage = self._init_with_timeout(
+                        lambda: get_conversation_storage(self.agent_name), 2.0
+                    )
+                    logger.info(f"[{self.agent_name}] Conversation persistence initialized")
+                except asyncio.CancelledError:
+                    logger.info(f"[{self.agent_name}] Conversation persistence initialization cancelled (shutdown in progress)")
+                    raise
+                except Exception as e:
+                    logger.warning(f"[{self.agent_name}] Conversation persistence initialization failed: {e}")
+
+            # Brain state - use AgentStateMachine for state management
+            self.brain_state = BrainState.IDLE  # Legacy brain state
+            self.state_machine = AgentStateMachine(agent_name or "cain")  # New state machine
+            self.tools: Dict[str, Tool] = {}
+            self.memory: Dict[str, Any] = {}
+            self.last_action = None
+            self.last_action_time = None
+
+            # Load memory from state store
+            self._load_memory_from_state()
+
+            # Register default tools
+            self._register_default_tools()
+
+        except asyncio.CancelledError:
+            # Log cancellation with stack trace for debugging
+            logger.info(f"[{self.agent_name if hasattr(self, 'agent_name') else 'brain'}] Initialization cancelled gracefully")
+            logger.debug(f"[CANCELLED] Stack trace:\n{''.join(traceback.format_stack())}")
+            # Re-raise to allow proper cleanup
+            raise
+        except Exception as e:
+            logger.error(f"[{agent_name or 'brain'}] Initialization failed: {e}")
+            raise
 
-        # Initialize State persistence
-        self.state_store: Optional[StateStore] = None
-        if STATE_PERSISTENCE_AVAILABLE:
-            try:
-                self.state_store = get_state_store()
-                logger.info(f"[{self.agent_name}] State persistence initialized")
-            except Exception as e:
-                logger.warning(f"[{self.agent_name}] State persistence initialization failed: {e}")
+    def _init_with_timeout(self, init_func, timeout: float = 2.0):
+        """
+        Initialize a component with timeout to prevent hanging during shutdown.
 
-        # Initialize Conversation persistence
-        self.conversation_storage: Optional[ConversationStorage] = None
-        if CONVERSATION_PERSISTENCE_AVAILABLE:
-            try:
-                self.conversation_storage = get_conversation_storage(self.agent_name)
-                logger.info(f"[{self.agent_name}] Conversation persistence initialized")
-            except Exception as e:
-                logger.warning(f"[{self.agent_name}] Conversation persistence initialization failed: {e}")
+        Args:
+            init_func: Function to call for initialization
+            timeout: Timeout in seconds
 
-        # Brain state - use AgentStateMachine for state management
-        self.brain_state = BrainState.IDLE  # Legacy brain state
-        self.state_machine = AgentStateMachine(agent_name or "cain")  # New state machine
-        self.tools: Dict[str, Tool] = {}
-        self.memory: Dict[str, Any] = {}
-        self.last_action = None
-        self.last_action_time = None
+        Returns:
+            Result of the initialization function
+        """
+        import concurrent.futures
 
-        # Load memory from state store
-        self._load_memory_from_state()
+        def _execute():
+            return init_func()
 
-        # Register default tools
-        self._register_default_tools()
+        try:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(_execute)
+                return future.result(timeout=timeout)
+        except concurrent.futures.TimeoutError:
+            logger.warning(f"[INIT_TIMEOUT] Initialization timed out after {timeout} seconds")
+            return None
+        except asyncio.CancelledError:
+            logger.info(f"[INIT_CANCELLED] Initialization cancelled by asyncio")
+            raise
 
     def _load_memory_from_state(self):
         """Load memory from state store on initialization."""
@@ -1179,25 +1322,46 @@ class BrainMinimal:
         """
         Shutdown the brain and sync all data.
 
+        Ensures graceful shutdown of all background tasks before
+        the event loop closes, preventing CancelledError noise.
+
         Returns:
             Result dict with shutdown status
         """
+        global _shutdown_in_progress
+        with _shutdown_lock:
+            _shutdown_in_progress = True
+
         results = {}
 
-        # Sync and stop conversation storage
+        logger.info(f"[{self.agent_name}] Starting graceful shutdown...")
+
+        # Sync and stop conversation storage with timeout
         if self.conversation_storage is not None:
             try:
-                self.conversation_storage.stop()
+                logger.debug(f"[{self.agent_name}] Stopping conversation storage...")
+                # Use a short timeout to prevent hanging
+                self._shutdown_with_timeout(self.conversation_storage.stop, 2.0)
                 results["conversation_storage"] = "stopped"
+            except asyncio.CancelledError:
+                logger.info(f"[{self.agent_name}] Conversation storage shutdown cancelled")
+                results["conversation_storage"] = "cancelled"
             except Exception as e:
+                logger.warning(f"[{self.agent_name}] Error stopping conversation storage: {e}")
                 results["conversation_storage"] = f"error: {e}"
 
-        # Stop state store
+        # Stop state store with timeout
         if self.state_store is not None:
             try:
-                self.state_store.stop()
+                logger.debug(f"[{self.agent_name}] Stopping state store...")
+                # Use a short timeout to prevent hanging
+                self._shutdown_with_timeout(self.state_store.stop, 2.0)
                 results["state_store"] = "stopped"
+            except asyncio.CancelledError:
+                logger.info(f"[{self.agent_name}] State store shutdown cancelled")
+                results["state_store"] = "cancelled"
             except Exception as e:
+                logger.warning(f"[{self.agent_name}] Error stopping state store: {e}")
                 results["state_store"] = f"error: {e}"
 
         logger.info(f"[{self.agent_name}] Brain shutdown complete")
@@ -1208,6 +1372,29 @@ class BrainMinimal:
             "shutdown_results": results
         }
 
+    def _shutdown_with_timeout(self, shutdown_func, timeout: float = 2.0):
+        """
+        Execute a shutdown function with timeout to prevent hanging.
+
+        Args:
+            shutdown_func: Function to call for shutdown
+            timeout: Timeout in seconds
+        """
+        import concurrent.futures
+
+        def _execute():
+            return shutdown_func()
+
+        try:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(_execute)
+                future.result(timeout=timeout)
+        except concurrent.futures.TimeoutError:
+            logger.warning(f"[SHUTDOWN_TIMEOUT] Shutdown operation timed out after {timeout} seconds")
+        except asyncio.CancelledError:
+            logger.info(f"[SHUTDOWN_CANCELLED] Shutdown cancelled by asyncio")
+            raise
+
     def get_info(self) -> Dict[str, Any]:
         """Get information about the brain"""
         return {