fix: wrapper layer gaps - async path observability, sync/async duplication, tool caching (fixes #1860)

- Gap 1: Extract shared _prepare_adapter() method used by both sync/async paths
  Ensures observability init and adapter setup are consistent between code paths
- Gap 2: Remove duplicate arun() method definitions in Protocol and BaseFrameworkAdapter
- Gap 3: Fix ToolResolver to distinguish cacheable vs non-cacheable failures
  Transient import errors no longer permanently cached, allowing deps installed later

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

Author: praisonai-triage-agent[bot]

src/praisonai/praisonai/agents_generator.py

@@ -253,6 +253,47 @@ 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)
+        adapter = initial_adapter.resolve()  # autogen v0.2/v0.4, etc.
+
+        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)
+
+        # AgentOps init lives here too -- once, not in two places.
+        api_key = os.getenv("AGENTOPS_API_KEY")
+        if api_key:
+            try:
+                import agentops
+                agentops.init(api_key, default_tags=[adapter.name])
+            except ImportError:
+                pass
+
+        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 +635,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 +761,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(

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).
@@ -304,12 +318,13 @@ 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
 
             # 4. Check praisonai-tools package
             tool = self._resolve_from_praisonai_tools(name)