Merge pull request #1277 from MervinPraison/claude/issue-1276-20260404-1738

fix: resolve critical architecture gaps for multi-agent safety

Author: MervinPraison

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

@@ -291,21 +291,6 @@ def get_summary(self) -> Dict[str, Any]:
             "final_stage": self._current_stage.name if self._current_stage else None,
         }
 
-# Global hooks instance (opt-in)
-_global_hooks: Optional[ObservabilityHooks] = None
-
-def get_hooks() -> Optional[ObservabilityHooks]:
-    """Get global observability hooks."""
-    return _global_hooks
-
-def enable_observability() -> ObservabilityHooks:
-    """Enable global observability."""
-    global _global_hooks
-    if _global_hooks is None:
-        _global_hooks = ObservabilityHooks(enabled=True)
-    return _global_hooks
-
-def disable_observability():
-    """Disable global observability."""
-    global _global_hooks
-    _global_hooks = None
+# NOTE: Observability is now per-agent/per-pipeline to comply with multi-agent safety.
+# Each agent creates its own ObservabilityHooks instance when observe=True.
+# See agent.py:2236 for the correct per-instance pattern.

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

@@ -8,6 +8,7 @@
 import json
 import logging
 import threading
+import asyncio
 from typing import Any, Dict, List, Optional
 from datetime import datetime
 
@@ -304,4 +305,86 @@ def learn(self):
                 self._learn_manager = LearnManager(config=self._learn_config)
             except ImportError:
                 logging.warning("Learn manager not available - install learn dependencies")
-        return self._learn_manager
+        return self._learn_manager
+
+    # Async variants to prevent event loop blocking
+    async def store_short_term_async(self, content: str, metadata: Optional[Dict] = None, quality_score: Optional[float] = None, 
+                                   user_id: Optional[str] = None, auto_promote: bool = True) -> str:
+        """
+        Async version of store_short_term to prevent event loop blocking.
+        
+        Args:
+            content: The content to store
+            metadata: Optional metadata dictionary
+            quality_score: Optional pre-calculated quality score
+            user_id: Optional user identifier
+            auto_promote: Whether to automatically promote to LTM if quality is high
+            
+        Returns:
+            The memory ID of the stored content
+        """
+        if not content.strip():
+            return ""
+        
+        # Calculate quality score if not provided
+        if quality_score is None:
+            quality_score = self.compute_quality_score(content, metadata)
+        
+        # Prepare metadata (mirror sync version's metadata construction including sanitization
+        # to ensure only JSON-serializable values are stored, preventing crashes across backends)
+        raw_metadata = metadata.copy() if metadata else {}
+        raw_metadata.update({
+            "timestamp": datetime.now().isoformat(),
+            "quality_score": quality_score,
+            "memory_type": "short_term"
+        })
+        if user_id:
+            raw_metadata["user_id"] = user_id
+        clean_metadata = self._sanitize_metadata(raw_metadata)
+
+        # Store in SQLite STM
+        memory_id = ""
+        try:
+            memory_id = await asyncio.to_thread(self._store_sqlite_stm, content, clean_metadata, quality_score)
+        except Exception as e:
+            logging.error(f"Failed to store in SQLite STM: {e}")
+            return ""
+        
+        # Auto-promote to long-term memory if quality is high (async)
+        if auto_promote and quality_score >= 7.5:  # High quality threshold
+            try:
+                await self.store_long_term_async(content, clean_metadata, quality_score, user_id)
+                self._log_verbose(f"Auto-promoted STM content to LTM (score: {quality_score:.2f})")
+            except Exception as e:
+                logging.warning(f"Failed to auto-promote to LTM: {e}")
+        
+        # Emit memory event
+        self._emit_memory_event("store", "short_term", content, clean_metadata)
+        
+        self._log_verbose(f"Stored in STM: {content[:100]}... (quality: {quality_score:.2f})")
+        
+        return memory_id or ""
+
+    async def store_long_term_async(self, content: str, metadata: Optional[Dict] = None, quality_score: Optional[float] = None,
+                                  user_id: Optional[str] = None) -> str:
+        """
+        Async version of store_long_term to prevent event loop blocking.
+        
+        Args:
+            content: The content to store
+            metadata: Optional metadata dictionary
+            quality_score: Optional pre-calculated quality score
+            user_id: Optional user identifier
+            
+        Returns:
+            The memory ID of the stored content
+        """
+        if not content.strip():
+            return ""
+        
+        # Calculate quality score if not provided
+        if quality_score is None:
+            quality_score = self.compute_quality_score(content, metadata)
+        
+        # Use sync version in thread to avoid blocking event loop
+        return await asyncio.to_thread(self.store_long_term, content, metadata, quality_score, user_id)

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

@@ -1,5 +1,6 @@
 import logging
 import asyncio
+import time
 import json
 from typing import Dict, Optional, List, Any, AsyncGenerator
 from pydantic import BaseModel, ConfigDict
@@ -25,6 +26,8 @@ def __init__(
         manager_llm: Optional[str] = None,
         verbose: bool = False,
         max_iter: int = 10,
+        max_retries: int = 3,
+        workflow_timeout: Optional[int] = None,  # seconds, None = no timeout
         output: Optional[str] = None,
     ):
         logging.debug(f"=== Initializing Process ===")
@@ -37,6 +40,8 @@ def __init__(
         self.agents = agents
         self.manager_llm = manager_llm
         self.max_iter = max_iter
+        self.max_retries = max_retries
+        self.workflow_timeout = workflow_timeout
         self.task_retry_counter: Dict[str, int] = {} # Initialize retry counter
         self.workflow_finished = False # ADDED: Workflow finished flag
         self._state_lock = asyncio.Lock() # Async lock for shared state protection
@@ -233,7 +238,7 @@ def _find_next_not_started_task(self) -> Optional[Task]:
             if hasattr(task, 'description') and 'Input data from previous tasks:' in task.description:
                 task.description = task.description.split('Input data from previous tasks:')[0].strip()
 
-        while fallback_attempts < Process.DEFAULT_RETRY_LIMIT and not temp_current_task:
+        while fallback_attempts < self.max_retries and not temp_current_task:
             fallback_attempts += 1
             logging.debug(f"Fallback attempt {fallback_attempts}: Trying to find next 'not started' task.")
             for task_candidate in self.tasks.values():
@@ -248,13 +253,13 @@ def _find_next_not_started_task(self) -> Optional[Task]:
                     if not leads_to_task and not task_candidate.next_tasks:
                         continue  # Skip if no valid path exists
 
-                    if self.task_retry_counter.get(task_candidate.id, 0) < Process.DEFAULT_RETRY_LIMIT:
+                    if self.task_retry_counter.get(task_candidate.id, 0) < self.max_retries:
                         self.task_retry_counter[task_candidate.id] = self.task_retry_counter.get(task_candidate.id, 0) + 1
                         temp_current_task = task_candidate
                         logging.debug(f"Fallback attempt {fallback_attempts}: Found 'not started' task: {temp_current_task.name}, retry count: {self.task_retry_counter[temp_current_task.id]}")
                         return temp_current_task # Return the found task immediately
                     else:
-                        logging.debug(f"Max retries reached for task {task_candidate.name} in fallback mode, marking as failed.")
+                        logging.debug(f"Max retries ({self.max_retries}) reached for task {task_candidate.name} in fallback mode, marking as failed.")
                         task_candidate.status = "failed"
             if not temp_current_task:
                 logging.debug(f"Fallback attempt {fallback_attempts}: No 'not started' task found within retry limit.")
@@ -383,6 +388,7 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
         """Async version of workflow method"""
         logging.debug("=== Starting Async Workflow ===")
         current_iter = 0  # Track how many times we've looped
+        workflow_start = time.monotonic()  # For timeout enforcement
         # Build workflow relationships first
         logging.debug("Building workflow relationships...")
         for task in self.tasks.values():
@@ -418,6 +424,13 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
                 logging.info(f"Max iteration limit {self.max_iter} reached, ending workflow.")
                 break
 
+            # Enforce workflow timeout if set
+            if self.workflow_timeout is not None:
+                elapsed = time.monotonic() - workflow_start
+                if elapsed > self.workflow_timeout:
+                    logging.warning(f"Workflow timeout ({self.workflow_timeout}s) exceeded after {elapsed:.1f}s, stopping.")
+                    break
+
             # ADDED: Check workflow finished flag at the start of each cycle
             if self.workflow_finished:
                 logging.info("Workflow finished early as all tasks are completed.")
@@ -455,11 +468,13 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
 Description length: {len(current_task.description)}
             """)
 
-            # Add context from previous tasks to description
+            # Build context and set description for this execution pass only
             context = self._build_task_context(current_task)
-            if context:
-                # Update task description with context
-                current_task.description = current_task.description + context
+            # Store original description if not already stored
+            if not hasattr(current_task, '_original_description'):
+                current_task._original_description = current_task.description
+            # Set description with context for execution; reset after yield to prevent accumulation
+            current_task.description = current_task._original_description + (context if context else "")
 
             # Skip execution for loop tasks, only process their subtasks
             if current_task.task_type == "loop":
@@ -566,6 +581,9 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
                 logging.debug(f"Task next_tasks: {current_task.next_tasks}")
                 yield task_id
                 visited_tasks.add(task_id)
+                # Reset description to original after execution to prevent context accumulation
+                if hasattr(current_task, '_original_description'):
+                    current_task.description = current_task._original_description
 
                 # Only end workflow if no next_tasks AND no conditions
                 if not current_task.next_tasks and not current_task.condition and not any(
@@ -577,28 +595,29 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
                     current_task = None
                     break
 
-            # Reset completed task to "not started" so it can run again
-            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
-                task_to_check = self.tasks[task_id]
-                logging.debug(f"=== Checking reset for completed task: {subtask_name} ===")
-                logging.debug(f"Task type: {task_to_check.task_type}")
-                logging.debug(f"Task status before reset check: {task_to_check.status}")
-                logging.debug(f"Task rerun: {getattr(task_to_check, 'rerun', True)}") # default to True if not set
-                logging.debug(f"Task async_execution: {task_to_check.async_execution}")
-
-                if (getattr(task_to_check, 'rerun', True) and # Corrected condition - reset only if rerun is True (or default True)
-                    task_to_check.task_type != "loop" and # Removed "decision" from exclusion
-                    not any(t.task_type == "loop" and subtask_name.startswith(t.name + "_")
-                           for t in self.tasks.values()) and
-                    not task_to_check.async_execution):  # Don't reset async parallel tasks
-                    logging.debug(f"=== Resetting non-loop, non-decision, non-parallel task {subtask_name} to 'not started' ===")
-                    self.tasks[task_id].status = "not started"
-                    logging.debug(f"Task status after reset: {self.tasks[task_id].status}")
-                else:
-                    logging.debug(f"=== Skipping reset for loop/decision/subtask/parallel or rerun=False: {subtask_name} ===")
-                    logging.debug(f"Keeping status as: {self.tasks[task_id].status}")
+            # Reset completed task to "not started" so it can run again (atomic operation)
+            async with self._state_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
+                    task_to_check = self.tasks[task_id]
+                    logging.debug(f"=== Checking reset for completed task: {subtask_name} ===")
+                    logging.debug(f"Task type: {task_to_check.task_type}")
+                    logging.debug(f"Task status before reset check: {task_to_check.status}")
+                    logging.debug(f"Task rerun: {getattr(task_to_check, 'rerun', True)}") # default to True if not set
+                    logging.debug(f"Task async_execution: {task_to_check.async_execution}")
+
+                    if (getattr(task_to_check, 'rerun', True) and # Corrected condition - reset only if rerun is True (or default True)
+                        task_to_check.task_type != "loop" and # Removed "decision" from exclusion
+                        not any(t.task_type == "loop" and subtask_name.startswith(t.name + "_")
+                               for t in self.tasks.values()) and
+                        not task_to_check.async_execution):  # Don't reset async parallel tasks
+                        logging.debug(f"=== Resetting non-loop, non-decision, non-parallel task {subtask_name} to 'not started' ===")
+                        self.tasks[task_id].status = "not started"
+                        logging.debug(f"Task status after reset: {self.tasks[task_id].status}")
+                    else:
+                        logging.debug(f"=== Skipping reset for loop/decision/subtask/parallel or rerun=False: {subtask_name} ===")
+                        logging.debug(f"Keeping status as: {self.tasks[task_id].status}")
 
             # Handle loop progression
             if current_task.task_type == "loop":
@@ -1116,11 +1135,13 @@ def workflow(self):
 Description length: {len(current_task.description)}
             """)
 
-            # Add context from previous tasks to description
+            # Build context and set description for this execution pass only
             context = self._build_task_context(current_task)
-            if context:
-                # Update task description with context
-                current_task.description = current_task.description + context
+            # Store original description if not already stored
+            if not hasattr(current_task, '_original_description'):
+                current_task._original_description = current_task.description
+            # Set description with context for execution; reset after yield to prevent accumulation
+            current_task.description = current_task._original_description + (context if context else "")
 
             # Skip execution for loop tasks, only process their subtasks
             if current_task.task_type == "loop":
@@ -1227,6 +1248,9 @@ def workflow(self):
                 logging.debug(f"Task next_tasks: {current_task.next_tasks}")
                 yield task_id
                 visited_tasks.add(task_id)
+                # Reset description to original after execution to prevent context accumulation
+                if hasattr(current_task, '_original_description'):
+                    current_task.description = current_task._original_description
 
                 # Only end workflow if no next_tasks AND no conditions
                 if not current_task.next_tasks and not current_task.condition and not any(

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

@@ -200,7 +200,16 @@
     'email_tools': ('.email_tools', None),
 }
 
-_instances = {}  # Cache for class instances (protected by _tools_lock)
+# Tool factory functions - caches classes but creates fresh instances
+# This prevents state leakage between concurrent agents while optimizing import performance
+_loaded_classes = {}  # Cache the Class, NOT the instance
+
+def _create_tool_instance(class_name: str, module_path: str):
+    """Create a new tool instance. Caches the class but returns fresh instances to prevent state sharing."""
+    if class_name not in _loaded_classes:
+        module = import_module(module_path, __package__)
+        _loaded_classes[class_name] = getattr(module, class_name)
+    return _loaded_classes[class_name]()  # Fresh instance safe for multi-agent
 
 # Profile exports (lazy loaded)
 _PROFILE_EXPORTS = frozenset({
@@ -325,17 +334,10 @@ def __getattr__(name: str) -> Any:
             return module  # Returns the callable module
         return getattr(module, name)
     else:
-        # Class method import (thread-safe)
-        if class_name not in _instances:
-            with _tools_lock:
-                # Double-check pattern to avoid race conditions
-                if class_name not in _instances:
-                    module = import_module(module_path, __package__)
-                    class_ = getattr(module, class_name)
-                    _instances[class_name] = class_()
-        
-        # Get the method and bind it to the instance
-        method = getattr(_instances[class_name], name)
+        # Create a fresh tool instance for each agent/session to prevent state leakage
+        # This factory pattern ensures multi-agent safety by avoiding shared mutable state
+        instance = _create_tool_instance(class_name, module_path)
+        method = getattr(instance, name)
         return method
 
 __all__ = list(TOOL_MAPPINGS.keys()) + [