arch: Fix 3 key architectural gaps for DRY, protocol-driven, multi-agent safety

Gap 1: Enable unified LLM dispatch by default
- Changed _use_unified_llm_dispatch default from False to True
- Removed dead if False: code block in chat_mixin.py:583-607
- Eliminates triple execution paths for better sync/async feature parity

Gap 2: Extend provider adapter protocol for complete dispatch replacement
- Added 6 new methods to LLMProviderAdapterProtocol: parse_tool_calls,
  should_skip_streaming_with_tools, recover_tool_calls_from_text,
  inject_cache_control, extract_reasoning_tokens
- Implemented Ollama-specific tool recovery and Gemini streaming skip logic
- Foundation to replace 24+ hardcoded provider checks with adapter calls

Gap 3: Replace singletons with instance-based registries
- Converted IndexRegistry from __new__ singleton to normal class with default()
- Replaced module-level _memory_registry with get_default_memory_registry()
- Added ServerRegistry class to replace global _server_* dicts in agent.py
- Enables multi-agent safe execution and independent registries per agent

Fixes #1362

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

Author: praisonai-triage-agent[bot]

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

@@ -169,11 +169,51 @@ def _is_file_path(value: str) -> bool:
 # Applied even when context management is disabled to prevent runaway tool outputs
 DEFAULT_TOOL_OUTPUT_LIMIT = 16000
 
-# Global variables for API server (protected by _server_lock for thread safety)
-_server_lock = threading.Lock()
-_server_started = {}  # Dict of port -> started boolean
-_registered_agents = {}  # Dict of port -> Dict of path -> agent_id
-_shared_apps = {}  # Dict of port -> FastAPI app
+class ServerRegistry:
+    """Registry for API server state per-port."""
+    
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._server_started = {}  # Dict of port -> started boolean
+        self._registered_agents = {}  # Dict of port -> Dict of path -> agent_id  
+        self._shared_apps = {}  # Dict of port -> FastAPI app
+    
+    @staticmethod
+    def get_default_instance():
+        """Get default global registry for backward compatibility."""
+        if not hasattr(ServerRegistry, '_default_instance'):
+            ServerRegistry._default_instance = ServerRegistry()
+        return ServerRegistry._default_instance
+    
+    def is_server_started(self, port: int) -> bool:
+        with self._lock:
+            return self._server_started.get(port, False)
+    
+    def set_server_started(self, port: int, started: bool) -> None:
+        with self._lock:
+            self._server_started[port] = started
+    
+    def get_shared_app(self, port: int):
+        with self._lock:
+            return self._shared_apps.get(port)
+    
+    def set_shared_app(self, port: int, app) -> None:
+        with self._lock:
+            self._shared_apps[port] = app
+    
+    def register_agent(self, port: int, path: str, agent_id: str) -> None:
+        with self._lock:
+            if port not in self._registered_agents:
+                self._registered_agents[port] = {}
+            self._registered_agents[port][path] = agent_id
+    
+    def get_registered_agents(self, port: int) -> dict:
+        with self._lock:
+            return self._registered_agents.get(port, {}).copy()
+
+# Backward compatibility - use default instance
+def _get_default_server_registry() -> ServerRegistry:
+    return ServerRegistry.get_default_instance()
 
 # Don't import FastAPI dependencies here - use lazy loading instead
 

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

@@ -536,9 +536,9 @@ def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, r
         formatted_tools = self._format_tools_for_completion(tools)
 
         try:
-            # NEW: Unified protocol dispatch path (Issue #1304)
-            # Check if unified dispatch is enabled (opt-in for backward compatibility)
-            if getattr(self, '_use_unified_llm_dispatch', False):
+            # NEW: Unified protocol dispatch path (Issue #1304, #1362)
+            # Enable unified dispatch by default for DRY and feature parity
+            if getattr(self, '_use_unified_llm_dispatch', True):
                 # Use composition instead of runtime class mutation for safety
                 final_response = self._execute_unified_chat_completion(
                     messages=messages,
@@ -579,50 +579,24 @@ def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, r
                         reasoning_steps=reasoning_steps
                     )
                 else:
-                    # Non-streaming with custom LLM - don't show streaming-like behavior
-                    if False:  # Don't use display_generating when stream=False to avoid streaming-like behavior
-                        # This block is disabled to maintain consistency with the OpenAI path fix
-                        with _get_live()(
-                            _get_display_functions()['display_generating']("", start_time),
-                            console=self.console,
-                            refresh_per_second=4,
-                        ) as live:
-                            final_response = self.llm_instance.get_response(
-                                prompt=messages[1:],
-                                system_prompt=messages[0]['content'] if messages and messages[0]['role'] == 'system' else None,
-                                temperature=temperature,
-                                tools=formatted_tools if formatted_tools else None,
-                                verbose=self.verbose,
-                                markdown=self.markdown,
-                                stream=stream,
-                                console=self.console,
-                                execute_tool_fn=self.execute_tool,
-                                agent_name=self.name,
-                                agent_role=self.role,
-                                agent_tools=[getattr(t, '__name__', str(t)) for t in self.tools] if self.tools else None,
-                                task_name=task_name,
-                                task_description=task_description,
-                                task_id=task_id,
-                                reasoning_steps=reasoning_steps
-                            )
-                    else:
-                        final_response = self.llm_instance.get_response(
-                            prompt=messages[1:],
-                            system_prompt=messages[0]['content'] if messages and messages[0]['role'] == 'system' else None,
-                            temperature=temperature,
-                            tools=formatted_tools if formatted_tools else None,
-                            verbose=self.verbose,
-                            markdown=self.markdown,
-                            stream=stream,
-                            console=self.console,
-                            execute_tool_fn=self.execute_tool,
-                            agent_name=self.name,
-                            agent_role=self.role,
-                            agent_tools=[getattr(t, '__name__', str(t)) for t in self.tools] if self.tools else None,
-                            task_name=task_name,
-                            task_description=task_description,
-                            task_id=task_id,
-                            reasoning_steps=reasoning_steps
+                    # Non-streaming with custom LLM - direct execution
+                    final_response = self.llm_instance.get_response(
+                        prompt=messages[1:],
+                        system_prompt=messages[0]['content'] if messages and messages[0]['role'] == 'system' else None,
+                        temperature=temperature,
+                        tools=formatted_tools if formatted_tools else None,
+                        verbose=self.verbose,
+                        markdown=self.markdown,
+                        stream=stream,
+                        console=self.console,
+                        execute_tool_fn=self.execute_tool,
+                        agent_name=self.name,
+                        agent_role=self.role,
+                        agent_tools=[getattr(t, '__name__', str(t)) for t in self.tools] if self.tools else None,
+                        task_name=task_name,
+                        task_description=task_description,
+                        task_id=task_id,
+                        reasoning_steps=reasoning_steps
                         )
             else:
                 # Use the standard OpenAI client approach with tool support

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

@@ -101,13 +101,16 @@ def stats(self) -> IndexStats:
 class IndexRegistry:
     """Registry for index types."""
 
-    _instance: Optional["IndexRegistry"] = None
+    def __init__(self):
+        """Initialize a new registry instance."""
+        self._indices: Dict[str, Callable[..., IndexProtocol]] = {}
 
-    def __new__(cls) -> "IndexRegistry":
-        if cls._instance is None:
-            cls._instance = super().__new__(cls)
-            cls._instance._indices: Dict[str, Callable[..., IndexProtocol]] = {}
-        return cls._instance
+    @classmethod
+    def default(cls) -> "IndexRegistry":
+        """Get a default global registry instance for convenience."""
+        if not hasattr(cls, '_default_instance'):
+            cls._default_instance = cls()
+        return cls._default_instance
 
     def register(self, name: str, factory: Callable[..., IndexProtocol]) -> None:
         """Register an index factory."""

src/praisonai-agents/praisonaiagents/llm/adapters/__init__.py

@@ -57,6 +57,25 @@ def handle_empty_response_with_tools(self, state: Dict[str, Any]) -> bool:
 
     def get_default_settings(self) -> Dict[str, Any]:
         return {}  # No provider-specific defaults
+    
+    def parse_tool_calls(self, raw_response: Dict[str, Any]) -> Optional[List[Dict[str, Any]]]:
+        """Default tool call parsing - use OpenAI-style format."""
+        if "choices" in raw_response and len(raw_response["choices"]) > 0:
+            message = raw_response["choices"][0].get("message", {})
+            return message.get("tool_calls")
+        return None
+    
+    def should_skip_streaming_with_tools(self) -> bool:
+        return False  # Most providers support streaming with tools
+    
+    def recover_tool_calls_from_text(self, response_text: str, tools: List[Dict[str, Any]]) -> Optional[List[Dict[str, Any]]]:
+        return None  # No text recovery by default
+    
+    def inject_cache_control(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        return messages  # No cache control by default
+    
+    def extract_reasoning_tokens(self, response: Dict[str, Any]) -> int:
+        return 0  # No reasoning tokens by default
 
 
 class OllamaAdapter(DefaultAdapter):
@@ -99,6 +118,29 @@ def handle_empty_response_with_tools(self, state: Dict[str, Any]) -> bool:
             return True  # Signal that special handling is needed
         return False
 
+    def recover_tool_calls_from_text(self, response_text: str, tools: List[Dict[str, Any]]) -> Optional[List[Dict[str, Any]]]:
+        """Ollama-specific tool call recovery from response text."""
+        if not response_text or not tools:
+            return None
+        
+        try:
+            import json
+            response_json = json.loads(response_text.strip())
+            if isinstance(response_json, dict) and "name" in response_json:
+                # Convert Ollama format to standard tool_calls format
+                return [{
+                    "id": f"call_{response_json['name']}_{hash(response_text) % 10000}",
+                    "type": "function",
+                    "function": {
+                        "name": response_json["name"],
+                        "arguments": json.dumps(response_json.get("arguments", {}))
+                    }
+                }]
+        except (json.JSONDecodeError, TypeError, KeyError):
+            pass
+        
+        return None
+    
     def post_tool_iteration(self, state: Dict[str, Any]) -> None:
         # Replaces: Ollama-specific post-tool summary branches
         if (not state.get('response_text', '').strip() and 
@@ -142,6 +184,13 @@ class GeminiAdapter(DefaultAdapter):
     - Supports structured output
     """
 
+    def should_skip_streaming_with_tools(self) -> bool:
+        """Gemini should skip streaming when tools are present."""
+        return True
+    
+    def supports_structured_output(self) -> bool:
+        return True
+    
     def format_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         # Replaces: gemini_internal_tools handling in llm.py
         # Internal tool names match GEMINI_INTERNAL_TOOLS: {'googleSearch', 'urlContext', 'codeExecution'}

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

@@ -326,6 +326,26 @@ def handle_empty_response_with_tools(self, state: Dict[str, Any]) -> bool:
     def get_default_settings(self) -> Dict[str, Any]:
         """Get provider-specific default settings."""
         ...
+    
+    def parse_tool_calls(self, raw_response: Dict[str, Any]) -> Optional[List[Dict[str, Any]]]:
+        """Parse tool calls from provider-specific response format."""
+        ...
+    
+    def should_skip_streaming_with_tools(self) -> bool:
+        """Check if provider should skip streaming when tools are present."""
+        ...
+    
+    def recover_tool_calls_from_text(self, response_text: str, tools: List[Dict[str, Any]]) -> Optional[List[Dict[str, Any]]]:
+        """Attempt to recover tool calls from response text for providers that don't format them properly."""
+        ...
+    
+    def inject_cache_control(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Inject provider-specific cache control headers."""
+        ...
+    
+    def extract_reasoning_tokens(self, response: Dict[str, Any]) -> int:
+        """Extract reasoning token count from provider-specific response."""
+        ...
 
 
 @runtime_checkable  

src/praisonai-agents/praisonaiagents/memory/adapters/registry.py

@@ -33,28 +33,32 @@ def __init__(self):
         super().__init__(adapter_type_name="Memory")
 
 
-# Global registry instance
-_memory_registry = MemoryAdapterRegistry()
+# Default registry instance for backward compatibility
+def get_default_memory_registry() -> MemoryAdapterRegistry:
+    """Get the default global registry instance for convenience."""
+    if not hasattr(get_default_memory_registry, '_default_instance'):
+        get_default_memory_registry._default_instance = MemoryAdapterRegistry()
+    return get_default_memory_registry._default_instance
 
 
 def register_memory_adapter(name: str, adapter_class: Type[MemoryProtocol]) -> None:
     """Register a memory adapter class."""
-    _memory_registry.register_adapter(name, adapter_class)
+    get_default_memory_registry().register_adapter(name, adapter_class)
 
 
 def register_memory_factory(name: str, factory_func: Callable[..., MemoryProtocol]) -> None:
     """Register a memory adapter factory function."""
-    _memory_registry.register_factory(name, factory_func)
+    get_default_memory_registry().register_factory(name, factory_func)
 
 
 def get_memory_adapter(name: str, **kwargs) -> Optional[MemoryProtocol]:
     """Get memory adapter instance by name."""
-    return _memory_registry.get_adapter(name, **kwargs)
+    return get_default_memory_registry().get_adapter(name, **kwargs)
 
 
 def list_memory_adapters() -> List[str]:
     """List all registered memory adapter names."""
-    return _memory_registry.list_adapters()
+    return get_default_memory_registry().list_adapters()
 
 
 # Canonical aliases per AGENTS.md naming conventions