fix: wrapper layer gaps - async path observability, sync/async duplication, tool caching

src/praisonai/praisonai/agents_generator.py

@@ -253,6 +253,54 @@ def _get_framework_adapter(self, framework: str) -> FrameworkAdapter:
         """
         return self._adapter_registry.create(framework)
 
+    def _prepare_adapter(self, framework: str, config: dict, tools_dict: dict):
+        """
+        Single source of truth for adapter resolution + setup + observability.
+
+        Called by both sync and async entry points so they cannot drift.
+
+        Args:
+            framework: Framework name to prepare
+            config: Configuration dictionary
+            tools_dict: Tools dictionary
+
+        Returns:
+            Prepared and configured framework adapter
+        """
+        initial_adapter = self._get_framework_adapter(framework)
+
+        # Handle YAML-level autogen_version override
+        autogen_version_override = config.get('autogen_version')
+        original_env_value = None
+        if framework == 'autogen' and autogen_version_override:
+            original_env_value = os.environ.get('AUTOGEN_VERSION')
+            os.environ['AUTOGEN_VERSION'] = str(autogen_version_override)
+            self.logger.debug(f"Temporarily setting AUTOGEN_VERSION={autogen_version_override}")
+
+        try:
+            adapter = initial_adapter.resolve()  # autogen v0.2/v0.4, etc.
+        finally:
+            # Restore original environment variable
+            if autogen_version_override and original_env_value is not None:
+                os.environ['AUTOGEN_VERSION'] = original_env_value
+            elif autogen_version_override and original_env_value is None:
+                os.environ.pop('AUTOGEN_VERSION', None)
+
+        from .framework_adapters.validators import assert_framework_available
+        assert_framework_available(adapter.name)
+
+        from .observability.hooks import init_observability
+        init_observability(adapter.name)
+
+        adapter.setup(framework_tag=adapter.name)
+
+        self._validate_cli_backend_compatibility(config, adapter.name)
+
+        self.framework = adapter.name
+        self.framework_adapter = adapter
+        self.logger.info(f"Using framework: {adapter.name}")
+        return adapter
+
     def _merge_cli_config(self, config, cli_config):
         """
         Merge CLI configuration with YAML configuration.
@@ -594,27 +642,7 @@ def generate_crew_and_kickoff(self):
                 self.logger.debug("tools folder exists in the root directory")
 
         framework = self.framework or config.get('framework', 'crewai')
-
-        # Get initial adapter and resolve to concrete variant
-        initial_adapter = self._get_framework_adapter(framework)
-        adapter = initial_adapter.resolve()
-
-        # Validate framework availability early
-        from .framework_adapters.validators import assert_framework_available
-        assert_framework_available(adapter.name)
-
-        # Initialize observability hooks
-        from .observability.hooks import init_observability
-        init_observability(adapter.name)
-
-        # Run adapter setup hooks
-        adapter.setup(framework_tag=adapter.name)
-
-        # Update framework reference if resolution changed it
-        self.framework = adapter.name
-        self.framework_adapter = adapter
-
-        self.logger.info(f"Using framework: {adapter.name}")
+        adapter = self._prepare_adapter(framework, config, tools_dict)
         return adapter.run(
             config,
             self.config_list,
@@ -740,51 +768,8 @@ async def _arun_framework(self, config):
                 self.logger.debug("tools folder exists in the root directory")
 
         framework = self.framework or config.get('framework', 'crewai')
-
-        # AutoGen version selection logic
-        if framework == "autogen":
-            autogen_v4_adapter = self._get_framework_adapter("autogen_v4")
-            autogen_v2_adapter = self._get_framework_adapter("autogen")
-
-            autogen_version = str(
-                config.get('autogen_version', os.environ.get("AUTOGEN_VERSION", "auto"))
-            ).lower()
-            use_v4 = False
-
-            if autogen_version == "v0.4" and autogen_v4_adapter.is_available():
-                use_v4 = True
-            elif autogen_version == "v0.2" and autogen_v2_adapter.is_available():
-                use_v4 = False
-            elif autogen_version == "auto":
-                use_v4 = autogen_v4_adapter.is_available()
-            else:
-                use_v4 = autogen_v4_adapter.is_available() and not autogen_v2_adapter.is_available()
-
-            framework = "autogen_v4" if use_v4 else "autogen"
-
-        # Initialize AgentOps if configured
-        agentops_api_key = os.getenv("AGENTOPS_API_KEY")
-        if agentops_api_key:
-            try:
-                import agentops
-                agentops.init(agentops_api_key, default_tags=[framework])
-            except ImportError:
-                pass
-
-        # Update framework adapter if framework changed
-        if framework != self.framework:
-            self.framework = framework
-            self.framework_adapter = self._get_framework_adapter(framework)
-
-        # Validate framework availability
-        from .framework_adapters.validators import assert_framework_available
-        assert_framework_available(framework)
-
-        # Validate cli_backend compatibility
-        self._validate_cli_backend_compatibility(config, framework)
-
-        self.logger.info(f"Using framework: {framework}")
-        return await self.framework_adapter.arun(
+        adapter = self._prepare_adapter(framework, config, tools_dict)
+        return await adapter.arun(
             config,
             self.config_list,
             topic,

src/praisonai/praisonai/framework_adapters/base.py

@@ -64,34 +64,6 @@ def run(
         """
         ...
 
-    async def arun(
-        self,
-        config: Dict[str, Any],
-        llm_config: List[Dict],
-        topic: str,
-        *,
-        tools_dict: Optional[Dict[str, Any]] = None,
-        agent_callback: Optional[Callable] = None,
-        task_callback: Optional[Callable] = None,
-        cli_config: Optional[Dict[str, Any]] = None,
-    ) -> str:
-        """
-        Async-native execution. Default = offload sync run() to a thread.
-
-        Args:
-            config: Framework configuration
-            llm_config: LLM configuration list
-            topic: Topic for the tasks
-            tools_dict: Available tools dictionary
-            agent_callback: Callback for agent events
-            task_callback: Callback for task events
-            cli_config: CLI configuration
-
-        Returns:
-            Execution result as string
-        """
-        ...
-
     async def arun(
         self,
         config: Dict[str, Any],
@@ -157,29 +129,6 @@ def _sub(m):
         # Only substitute simple variable names like {topic}, not JSON like {"level":2}
         return re.sub(r'\{([a-zA-Z_][a-zA-Z0-9_]*)\}', _sub, template)
 
-    async def arun(
-        self,
-        config: Dict[str, Any],
-        llm_config: List[Dict],
-        topic: str,
-        *,
-        tools_dict: Optional[Dict[str, Any]] = None,
-        agent_callback: Optional[Callable] = None,
-        task_callback: Optional[Callable] = None,
-        cli_config: Optional[Dict[str, Any]] = None,
-    ) -> str:
-        """
-        Default async implementation that falls back to thread-offloaded sync.
-
-        Framework adapters with native async support should override this method.
-        """
-        import asyncio
-        return await asyncio.to_thread(
-            self.run, config, llm_config, topic,
-            tools_dict=tools_dict, agent_callback=agent_callback,
-            task_callback=task_callback, cli_config=cli_config
-        )
-
     def resolve(self) -> "FrameworkAdapter":
         """Default implementation returns self."""
         return self
@@ -200,8 +149,9 @@ async def arun(
         cli_config: Optional[Dict[str, Any]] = None,
     ) -> str:
         """
-        Safe default for sync-only adapters (crewai, autogen v0.2):
-        run the sync implementation in a worker thread, freeing the loop.
+        Safe default: run sync implementation in a worker thread.
+
+        Framework adapters with native async support should override this method.
         """
         import asyncio
         return await asyncio.to_thread(
@@ -211,6 +161,7 @@ async def arun(
             task_callback=task_callback,
             cli_config=cli_config
         )
+
     def cleanup(self) -> None:
         """Clean up resources - default implementation does nothing."""
         pass

src/praisonai/praisonai/tool_resolver.py

@@ -34,6 +34,16 @@
 
 logger = logging.getLogger(__name__)
 
+
+class _ResolveResult:
+    """Internal result wrapper to distinguish cacheable vs non-cacheable failures."""
+    __slots__ = ("tool", "cacheable")
+
+    def __init__(self, tool, cacheable=True):
+        self.tool = tool
+        self.cacheable = cacheable
+
+
 # Sentinel for cache - needed because None is a valid cached result (tool not found)
 _SENTINEL = object()
 
@@ -123,7 +133,7 @@ def _load_local_tools(self) -> Mapping[str, Callable]:
             self._local_tools_loaded = True
             return self._local_tools_cache
 
-    def _resolve_from_praisonaiagents(self, name: str) -> Optional[Callable]:
+    def _resolve_from_praisonaiagents(self, name: str) -> _ResolveResult:
         """Resolve tool from praisonaiagents.tools.TOOL_MAPPINGS.
 
         Uses lazy loading via __getattr__ in praisonaiagents.tools.
@@ -132,7 +142,7 @@ def _resolve_from_praisonaiagents(self, name: str) -> Optional[Callable]:
             name: Tool name to resolve
 
         Returns:
-            Callable if found, None otherwise
+            _ResolveResult with tool and cacheable flag
         """
         try:
             from praisonaiagents import tools as agent_tools
@@ -149,25 +159,29 @@ def _resolve_from_praisonaiagents(self, name: str) -> Optional[Callable]:
                         logger.warning(
                             f"Tool '{name}' exists in TOOL_MAPPINGS but failed to load: {e}"
                         )
-                        return None
+                        # IMPORTANT: do NOT cache. The dep may be installed later.
+                        return _ResolveResult(None, cacheable=False)
                     if tool is not None:
                         logger.debug(f"Resolved '{name}' from praisonaiagents.tools")
-                        return tool
+                        return _ResolveResult(tool)
 
             # Also try direct attribute access (for non-TOOL_MAPPINGS items)
             tool = getattr(agent_tools, name, None)
             if tool is not None and callable(tool):
                 logger.debug(f"Resolved '{name}' from praisonaiagents.tools (direct)")
-                return tool
+                return _ResolveResult(tool)
 
         except ImportError:
             logger.debug("praisonaiagents not available")
+            # SDK can be installed later
+            return _ResolveResult(None, cacheable=False)
         except AttributeError:
             pass
         except Exception as e:
             logger.debug(f"Error resolving '{name}' from praisonaiagents: {e}")
 
-        return None
+        # Genuinely not present
+        return _ResolveResult(None)
 
     def _resolve_from_praisonai_tools(self, name: str) -> Optional[Callable]:
         """Resolve tool from praisonai-tools package (external).
@@ -286,6 +300,9 @@ def resolve(self, name: str, instantiate: bool = False) -> Optional[Callable]:
                     return cached()
                 return cached
 
+            # Track if any source indicated a non-cacheable failure
+            allow_none_cache = True
+
             # 1. Check local tools.py first (highest priority)
             if name in local_tools:
                 logger.debug(f"Resolved '{name}' from local tools.py")
@@ -304,12 +321,16 @@ def resolve(self, name: str, instantiate: bool = False) -> Optional[Callable]:
                 return tool
 
             # 3. Check praisonaiagents.tools
-            tool = self._resolve_from_praisonaiagents(name)
-            if tool is not None:
-                self._resolve_cache[name] = tool
-                if instantiate and self._is_class(tool):
-                    return tool()
-                return tool
+            result = self._resolve_from_praisonaiagents(name)
+            if result.tool is not None:
+                if result.cacheable:
+                    self._resolve_cache[name] = result.tool
+                if instantiate and self._is_class(result.tool):
+                    return result.tool()
+                return result.tool
+            elif not result.cacheable:
+                # Track that a non-cacheable failure occurred
+                allow_none_cache = False
 
             # 4. Check praisonai-tools package
             tool = self._resolve_from_praisonai_tools(name)
@@ -327,9 +348,12 @@ def resolve(self, name: str, instantiate: bool = False) -> Optional[Callable]:
                     return tool()
                 return tool
 
-            # Cache the None result to avoid repeated failed lookups
-            logger.warning(f"Tool '{name}' not found in any source")
-            self._resolve_cache[name] = None
+            # Cache None only if the failure was not transient
+            if allow_none_cache:
+                logger.warning(f"Tool '{name}' not found in any source")
+                self._resolve_cache[name] = None
+            else:
+                logger.debug(f"Tool '{name}' failed transiently; not caching None")
             return None
 
     def _is_class(self, obj) -> bool: