Release v4.6.45

feat: PraisonAIUI host integration (Patterns B/C) + core protocol bridges

api.md

@@ -756,7 +756,7 @@ Methods:
 
 Types:
 ```python
-from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__
+from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__, arun, run
 ```
 
 # CLI

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.43" \
+    "praisonai>=4.6.45" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.43" \
+    "praisonai>=4.6.45" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.43" \
+    "praisonai>=4.6.45" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/python/a2ui/send_surface.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+"""Minimal A2UI example — agent + send_a2ui_messages tool.
+
+Install:
+    pip install praisonaiagents[a2ui]
+
+Run:
+    python send_surface.py
+
+For live rendering, use PraisonAIUI with AG-UI (/agui) — see integrate-a2ui-frontend docs.
+"""
+
+from praisonaiagents import Agent
+from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
+
+agent = Agent(
+    name="ui-assistant",
+    instructions=(
+        "When asked for UI, call send_a2ui_messages with A2UI v0.9 JSON "
+        "(createSurface on surface id 'main')."
+    ),
+    tools=[send_a2ui_messages],
+)
+
+if __name__ == "__main__":
+    result = agent.start("Show a simple welcome card on surface main")
+    print(result)

examples/yaml/a2ui-canvas.yaml

@@ -0,0 +1,19 @@
+# A2UI canvas agent — YAML example
+#
+# Usage:
+#   pip install praisonaiagents[a2ui]
+#   praisonai agent run --file a2ui-canvas.yaml --prompt "Show a welcome card"
+#
+# For live A2UI rendering, use PraisonAIUI (AG-UI / surfaces):
+#   pip install praisonaiui
+#   See examples/python/a2ui/send_surface.py and PraisonAIUI example 29.
+
+agent:
+  name: "ui-assistant"
+  instructions: |
+    You are a helpful assistant. When the user asks for UI elements
+    (buttons, cards, forms), call send_a2ui_messages with valid A2UI v0.9 JSON.
+    Use surface id "main" unless told otherwise.
+  llm: gpt-4o-mini
+  tools:
+    - send_a2ui_messages

src/praisonai-agents/praisonaiagents/__init__.py

@@ -343,6 +343,7 @@ _LAZY_IMPORTS = {
     # UI (optional)
     'AGUI': ('praisonaiagents.ui.agui', 'AGUI'),
     'A2A': ('praisonaiagents.ui.a2a', 'A2A'),
+    'A2UI': ('praisonaiagents.ui.a2ui', 'A2UI'),
     
     # Feature configs
     'MemoryConfig': ('praisonaiagents.config.feature_configs', 'MemoryConfig'),
@@ -814,6 +815,6 @@ def __dir__():
 # - Workflow, Task, Route, Parallel, Loop, Repeat, Pipeline, etc.
 # - MCP, FlowDisplay, track_workflow, FastContext, etc.
 # - Plan, PlanStep, TodoList, PlanningAgent, ApprovalCallback, etc.
-# - RAG, RAGConfig, RAGResult, AGUI, A2A, etc.
+# - RAG, RAGConfig, RAGResult, AGUI, A2A, A2UI, etc.
 # - All telemetry, display, and utility functions
 # ============================================================================

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

@@ -4616,60 +4616,57 @@ Answer:"""
                 error=f"Agent guardrail validation error: {str(e)}"
             )
 
-    def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
-        """Apply guardrail validation with retry logic.
-        
-        Args:
-            response_text: The response to validate
-            prompt: Original prompt for regeneration if needed
-            temperature: Temperature for regeneration
-            tools: Tools for regeneration
-            
-        Returns:
-            str: The validated response text or None if validation fails after retries
-        """
+    def _validate_with_guardrail(self, response_text):
+        """Validate response with guardrail. Returns (success, result, error)."""
         if not self._guardrail_fn:
-            return response_text
+            return True, response_text, None
             
         from ..main import TaskOutput
         
+        task_output = TaskOutput(
+            description="Agent response output",
+            raw=response_text,
+            agent=self.name
+        )
+        
+        guardrail_result = self._process_guardrail(task_output)
+        
+        if guardrail_result.success:
+            # Return the potentially modified result
+            if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
+                return True, guardrail_result.result.raw, None
+            elif guardrail_result.result:
+                return True, str(guardrail_result.result), None
+            else:
+                return True, response_text, None
+        else:
+            return False, None, guardrail_result.error
+
+    def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
+        """Apply guardrail validation with retry logic (sync version)."""
         retry_count = 0
         current_response = response_text
         
         while retry_count <= self.max_guardrail_retries:
-            # Create TaskOutput object
-            task_output = TaskOutput(
-                description="Agent response output",
-                raw=current_response,
-                agent=self.name
-            )
+            success, result, error = self._validate_with_guardrail(current_response)
             
-            # Process guardrail
-            guardrail_result = self._process_guardrail(task_output)
-            
-            if guardrail_result.success:
+            if success:
                 logging.info(f"Agent {self.name}: Guardrail validation passed")
-                # Return the potentially modified result
-                if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
-                    return guardrail_result.result.raw
-                elif guardrail_result.result:
-                    return str(guardrail_result.result)
-                else:
-                    return current_response
+                return result
             
             # Guardrail failed
             if retry_count >= self.max_guardrail_retries:
                 raise Exception(
                     f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
-                    f"Last error: {guardrail_result.error}"
+                    f"Last error: {error}"
                 )
             
             retry_count += 1
-            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {guardrail_result.error}")
+            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
             
             # Regenerate response for retry
             try:
-                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {guardrail_result.error}. Please provide an improved response."
+                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
                 response = self._chat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
                 if response and response.choices:
                     content = response.choices[0].message.content
@@ -4678,10 +4675,46 @@ Answer:"""
                     raise Exception("Failed to generate retry response")
             except Exception as e:
                 logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
-                # If we can't regenerate, fail the guardrail
+                raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
+        
+        return current_response
+
+    async def _aapply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
+        """Apply guardrail validation with retry logic (async version)."""
+        retry_count = 0
+        current_response = response_text
+        
+        while retry_count <= self.max_guardrail_retries:
+            success, result, error = self._validate_with_guardrail(current_response)
+            
+            if success:
+                logging.info(f"Agent {self.name}: Guardrail validation passed")
+                return result
+            
+            # Guardrail failed
+            if retry_count >= self.max_guardrail_retries:
                 raise Exception(
-                    f"Agent {self.name} guardrail retry failed: {e}"
+                    f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
+                    f"Last error: {error}"
                 )
+            
+            retry_count += 1
+            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
+            
+            # Regenerate response for retry (async version)
+            try:
+                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
+                response = await self._execute_unified_achat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
+                if response and hasattr(response, 'choices') and response.choices:
+                    content = response.choices[0].message.content
+                    current_response = content.strip() if content else ""
+                elif isinstance(response, str):
+                    current_response = response.strip()
+                else:
+                    raise Exception("Failed to generate retry response")
+            except Exception as e:
+                logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
+                raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
         
         return current_response
     

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

@@ -2079,7 +2079,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                     
                     # Apply guardrail validation for custom LLM response
                     try:
-                        validated_response = self._apply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
+                        validated_response = await self._aapply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
                         # Execute callback after validation
                         self._execute_callback_and_display(normalized_content, validated_response, time.time() - start_time, task_name, task_description, task_id)
                         return await self._atrigger_after_agent_hook(prompt, validated_response, start_time)
@@ -2187,22 +2187,216 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                             task_id=task_id,
                             response_format=response_format
                         )
-                        return await self._finalize_unified_achat_response(
-                            response=response,
-                            original_prompt=original_prompt,
-                            temperature=temperature,
-                            tools=tools,
-                            output_json=output_json,
-                            output_pydantic=output_pydantic,
-                            reasoning_steps=reasoning_steps,
-                            stream=stream,
-                            messages=messages,
-                            chat_history_length=chat_history_length,
-                            start_time=start_time,
-                            task_name=task_name,
-                            task_description=task_description,
-                            task_id=task_id,
-                        )
+                        
+                        # Process response - mirror sync _chat_impl behavior (lines ~1544-1602)
+                        if not response:
+                            # Rollback chat history on response failure
+                            self._truncate_chat_history(chat_history_length)
+                            return None
+
+                        # Extract response content using the same method as sync
+                        response_text = self._extract_llm_response_content(response)
+                        if isinstance(response_text, str):
+                            response_text = response_text.strip()
+
+                        # Handle output_json or output_pydantic if specified
+                        if output_json or output_pydantic:
+                            # Add to chat history and return raw response
+                            # User message already added before LLM call via _build_messages
+                            self._append_to_chat_history({"role": "assistant", "content": response_text})
+                            # Persist assistant message to DB
+                            self._persist_message("assistant", response_text)
+                            # Apply guardrail validation even for JSON output
+                            try:
+                                validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                # Execute callback after validation
+                                self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                            except Exception as e:
+                                logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
+                                # Rollback chat history on guardrail failure
+                                self._truncate_chat_history(chat_history_length)
+                                return None
+
+                        # For regular responses (no self-reflection)
+                        if not self.self_reflect:
+                            # User message already added before LLM call via _build_messages
+                            self._append_to_chat_history({"role": "assistant", "content": response_text})
+                            # Persist assistant message to DB (non-reflect path)
+                            self._persist_message("assistant", response_text)
+                            if self.verbose:
+                                logging.debug(f"Agent {self.name} final response: {response_text}")
+                            # Return only reasoning content if reasoning_steps is True
+                            if reasoning_steps and hasattr(response, 'choices') and response.choices and hasattr(response.choices[0].message, 'reasoning_content') and response.choices[0].message.reasoning_content:
+                                # Apply guardrail to reasoning content
+                                try:
+                                    validated_reasoning = await self._aapply_guardrail_with_retry(response.choices[0].message.reasoning_content, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                    # Execute callback after validation
+                                    self._execute_callback_and_display(original_prompt, validated_reasoning, time.time() - start_time, task_name, task_description, task_id)
+                                    return await self._atrigger_after_agent_hook(original_prompt, validated_reasoning, start_time)
+                                except Exception as e:
+                                    logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
+                                    # Rollback chat history on guardrail failure
+                                    self._truncate_chat_history(chat_history_length)
+                                    return None
+                            else:
+                                # Apply guardrail to regular response content
+                                try:
+                                    validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                    # Execute callback after validation
+                                    self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                    return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                                except Exception as e:
+                                    logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
+                                    # Rollback chat history on guardrail failure
+                                    self._truncate_chat_history(chat_history_length)
+                                    return None
+                        
+                        # If self-reflection is enabled, implement reflection logic
+                        if self.self_reflect:
+                            # Implement async self-reflection similar to sync path
+                            reflection_prompt = f"""
+Reflect on your previous response: '{response_text}'.
+{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
+Provide a "satisfactory" status ('yes' or 'no').
+Output MUST be JSON with 'reflection' and 'satisfactory'.
+                            """
+                            
+                            # Add reflection prompt to messages
+                            reflection_messages = messages + [
+                                {"role": "assistant", "content": response_text},
+                                {"role": "user", "content": reflection_prompt}
+                            ]
+                            
+                            reflection_count = 0
+                            
+                            while True:
+                                try:
+                                    # Check if OpenAI client is available for self-reflection
+                                    if self._openai_client is None:
+                                        # For custom LLMs, self-reflection with structured output is not supported
+                                        if self.verbose:
+                                            _get_display_functions()['display_self_reflection'](f"Agent {self.name}: Self-reflection with structured output is not supported for custom LLM providers. Skipping reflection.", console=self.console)
+                                        # Return the original response without reflection
+                                        self._append_to_chat_history({"role": "assistant", "content": response_text})
+                                        # Persist assistant message to DB
+                                        self._persist_message("assistant", response_text)
+                                        try:
+                                            validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                            # Execute callback after validation
+                                            self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                            return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                                        except Exception as e:
+                                            logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
+                                            # Rollback chat history on guardrail failure
+                                            self._truncate_chat_history(chat_history_length)
+                                            return None
+                                    
+                                    reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
+                                        model=self.reflect_llm if self.reflect_llm else self.llm,
+                                        messages=reflection_messages,
+                                        temperature=temperature,
+                                        response_format=_get_display_functions()['ReflectionOutput']
+                                    )
+                                    
+                                    reflection_output = reflection_response.choices[0].message.parsed
+                                    
+                                    if self.verbose:
+                                        _get_display_functions()['display_self_reflection'](f"Agent {self.name} self reflection (using {self.reflect_llm if self.reflect_llm else self.llm}): reflection='{reflection_output.reflection}' satisfactory='{reflection_output.satisfactory}'", console=self.console)
+                                    
+                                    # Only consider satisfactory after minimum reflections
+                                    if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
+                                        if self.verbose:
+                                            _get_display_functions()['display_self_reflection']("Agent marked the response as satisfactory after meeting minimum reflections", console=self.console)
+                                        # Add to chat history and return
+                                        self._append_to_chat_history({"role": "assistant", "content": response_text})
+                                        # Persist assistant message to DB
+                                        self._persist_message("assistant", response_text)
+                                        try:
+                                            validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                            # Execute callback after validation
+                                            self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                            return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                                        except Exception as e:
+                                            logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
+                                            # Rollback chat history on guardrail failure
+                                            self._truncate_chat_history(chat_history_length)
+                                            return None
+                                    
+                                    # Check if we've hit max reflections
+                                    if reflection_count >= self.max_reflect - 1:
+                                        if self.verbose:
+                                            _get_display_functions()['display_self_reflection']("Maximum reflection count reached, returning current response", console=self.console)
+                                        # Add to chat history and return
+                                        self._append_to_chat_history({"role": "assistant", "content": response_text})
+                                        # Persist assistant message to DB
+                                        self._persist_message("assistant", response_text)
+                                        try:
+                                            validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                            # Execute callback after validation
+                                            self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                            return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                                        except Exception as e:
+                                            logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
+                                            # Rollback chat history on guardrail failure
+                                            self._truncate_chat_history(chat_history_length)
+                                            return None
+                                    
+                                    # Regenerate response based on reflection
+                                    regenerate_messages = reflection_messages + [
+                                        {"role": "assistant", "content": f"Self Reflection: {reflection_output.reflection} Satisfactory?: {reflection_output.satisfactory}"},
+                                        {"role": "user", "content": "Now regenerate your response using the reflection you made"}
+                                    ]
+                                    
+                                    new_response = await self._execute_unified_achat_completion(
+                                        messages=regenerate_messages,
+                                        temperature=temperature,
+                                        tools=formatted_tools,
+                                        stream=stream,
+                                        reasoning_steps=reasoning_steps,
+                                        task_name=task_name,
+                                        task_description=task_description,
+                                        task_id=task_id,
+                                        response_format=response_format
+                                    )
+                                    
+                                    if new_response:
+                                        new_response_text = self._extract_llm_response_content(new_response)
+                                        if isinstance(new_response_text, str):
+                                            response_text = new_response_text.strip()
+                                        # Update reflection_messages to include the new response for next iteration
+                                        reflection_messages = regenerate_messages + [
+                                            {"role": "assistant", "content": response_text}
+                                        ]
+                                    
+                                    reflection_count += 1
+                                    
+                                except Exception as e:
+                                    if self.verbose:
+                                        _get_display_functions()['display_error'](f"Error in parsing self-reflection json {e}. Retrying", console=self.console)
+                                    logging.error("Reflection parsing failed.", exc_info=True)
+                                    reflection_count += 1
+                                    if reflection_count >= self.max_reflect:
+                                        # Return original response after max reflection attempts
+                                        self._append_to_chat_history({"role": "assistant", "content": response_text})
+                                        # Persist assistant message to DB
+                                        self._persist_message("assistant", response_text)
+                                        try:
+                                            validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                                            # Execute callback after validation
+                                            self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
+                                            return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
+                                        except Exception as guard_e:
+                                            logging.error(f"Agent {self.name}: Guardrail validation failed after reflection error: {guard_e}")
+                                            # Rollback chat history on guardrail failure
+                                            self._truncate_chat_history(chat_history_length)
+                                            return None
+                                    continue
+                        
+                        # This should never be reached due to the returns above
+                        # But adding as safety fallback
+                        logging.warning(f"Agent {self.name}: Unexpected code path reached in unified dispatch")
+                        return None
                     else:
                         # LEGACY: Check if OpenAI client is available
                         if self._openai_client is None:
@@ -2339,7 +2533,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                         
                         # Apply guardrail validation for OpenAI client response
                         try:
-                            validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
+                            validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
                             # Execute callback after validation
                             self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
                             return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)

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

@@ -403,9 +403,33 @@ class MemoryMixin:
                     self._session_store.add_user_message(self._session_id, content)
                 elif role == "assistant":
                     self._session_store.add_assistant_message(self._session_id, content)
+                self._persist_session_stats()
             except Exception as e:
                 logging.warning(f"Failed to persist message to session store: {e}")
 
+    def _persist_session_stats(self):
+        """Flush agent cost/token stats into session JSON metadata."""
+        store = getattr(self, "_session_store", None)
+        session_id = getattr(self, "_session_id", None)
+        if store is None or session_id is None:
+            return
+        if not hasattr(store, "update_session_metadata"):
+            return
+        try:
+            summary = getattr(self, "cost_summary", {}) or {}
+            llm = getattr(self, "llm", None)
+            model = str(llm) if llm is not None else None
+            store.update_session_metadata(
+                session_id,
+                model=model,
+                total_tokens=summary.get("tokens_in", 0) + summary.get("tokens_out", 0),
+                cost=summary.get("cost", 0),
+                source=getattr(self, "_session_source", "chat"),
+                agent_id=getattr(self, "agent_id", None) or getattr(self, "_agent_id", None),
+            )
+        except Exception as e:
+            logging.debug(f"Session stats persist skipped: {e}")
+
     def session_id(self) -> Optional[str]:
         """Get the current session ID."""
         return self._session_id

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

@@ -26,6 +26,21 @@ if TYPE_CHECKING:
 class ToolExecutionMixin:
     """Mixin providing toolexecution methods for the Agent class."""
 
+    def _get_existing_stream_emitter(self):
+        """Return an already-initialized stream emitter without creating one."""
+        emitter = getattr(self, "_stream_emitter", None)
+        if emitter is not None:
+            return emitter
+
+        # Support name-mangled private attributes across class renames/inheritance.
+        for cls in type(self).mro():
+            mangled = f"_{cls.__name__}__stream_emitter"
+            if hasattr(self, mangled):
+                emitter = getattr(self, mangled, None)
+                if emitter is not None:
+                    return emitter
+        return None
+
     def _resolve_tool_names(self, tool_names):
         """Resolve tool names to actual tool instances from registry.
         
@@ -160,8 +175,9 @@ class ToolExecutionMixin:
         
         # Emit TOOL_CALL_START to stream_emitter (for AIUI/AG-UI consumers)
         # Zero overhead when no callbacks registered
-        if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
-            getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
+        _stream_emitter = self._get_existing_stream_emitter()
+        if _stream_emitter is not None and _stream_emitter.has_callbacks:
+            _stream_emitter.emit(StreamEvent(
                 type=StreamEventType.TOOL_CALL_START,
                 timestamp=_tool_start_perf,
                 tool_call={
@@ -275,10 +291,10 @@ class ToolExecutionMixin:
             
             # Emit TOOL_CALL_RESULT to stream_emitter (for AIUI/AG-UI consumers)
             # Zero overhead when no callbacks registered
-            if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
+            if _stream_emitter is not None and _stream_emitter.has_callbacks:
                 # Truncate result for stream event (keep it reasonable for UI display)
                 result_summary = str(result)[:500] if result else None
-                getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
+                _stream_emitter.emit(StreamEvent(
                     type=StreamEventType.TOOL_CALL_RESULT,
                     timestamp=_time.perf_counter(),
                     tool_call={
@@ -290,6 +306,18 @@ class ToolExecutionMixin:
                     agent_id=self.name,
                     metadata={"duration_ms": _duration_ms},
                 ))
+                _stream_emitter.emit(StreamEvent(
+                    type=StreamEventType.TOOL_CALL_END,
+                    timestamp=_time.perf_counter(),
+                    tool_call={
+                        "name": function_name,
+                        "arguments": arguments,
+                        "result": result_summary,
+                        "id": tool_call_id,
+                    },
+                    agent_id=self.name,
+                    metadata={"duration_ms": _duration_ms},
+                ))
             
             # Trigger AFTER_TOOL hook
             from ..hooks import HookEvent, AfterToolInput
@@ -926,4 +954,3 @@ class ToolExecutionMixin:
     def pending_approval_count(self) -> int:
         """Number of approval requests still waiting."""
         return len(self._pending_approvals)
-

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

@@ -15,9 +15,9 @@ from enum import Enum
 
 # Import token tracking
 try:
-    from ..telemetry.token_collector import _token_collector
+    from ..telemetry.token_collector import get_token_collector
 except ImportError:
-    _token_collector = None
+    get_token_collector = None
 
 # Import async utility for hot-path usage
 try:
@@ -1826,18 +1826,19 @@ class AgentTeam:
 
     def get_token_usage_summary(self) -> Dict[str, Any]:
         """Get a summary of token usage across all agents and tasks."""
-        if not _token_collector:
+        if not get_token_collector:
             return {"error": "Token tracking not available"}
         
-        return _token_collector.get_session_summary()
+        return get_token_collector().get_session_summary()
     
     def get_detailed_token_report(self) -> Dict[str, Any]:
         """Get a detailed token usage report."""
-        if not _token_collector:
+        if not get_token_collector:
             return {"error": "Token tracking not available"}
         
-        summary = _token_collector.get_session_summary()
-        recent = _token_collector.get_recent_interactions(limit=20)
+        collector = get_token_collector()
+        summary = collector.get_session_summary()
+        recent = collector.get_recent_interactions(limit=20)
         
         # Calculate cost estimates (example rates)
         cost_per_1k_input = 0.0005  # $0.0005 per 1K input tokens
@@ -1861,11 +1862,11 @@ class AgentTeam:
     
     def display_token_usage(self):
         """Display token usage in a formatted table."""
-        if not _token_collector:
+        if not get_token_collector:
             print("Token tracking not available")
             return
         
-        summary = _token_collector.get_session_summary()
+        summary = get_token_collector().get_session_summary()
         
         print("\n" + "="*50)
         print("TOKEN USAGE SUMMARY")

src/praisonai-agents/praisonaiagents/hooks/query_protocol.py

@@ -0,0 +1,13 @@
+"""Hooks query protocol for UI/API consumers."""
+
+from __future__ import annotations
+
+from typing import Dict, List, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class HooksQueryProtocol(Protocol):
+    """Read path for registered hooks."""
+
+    def list_hooks_for_api(self) -> List[Dict]:
+        ...

src/praisonai-agents/praisonaiagents/hooks/registry.py

@@ -439,3 +439,13 @@ def has_hook(event: Union[str, HookEvent]) -> bool:
             )
     
     return get_default_registry().has_hooks(event)
+
+
+def list_hooks_for_api() -> List[Dict]:
+    """Flat hook list suitable for REST/UI consumers."""
+    registry = get_default_registry()
+    flat: List[Dict] = []
+    for event, hooks in registry.list_hooks().items():
+        for h in hooks:
+            flat.append({"event": event, "source": "sdk", **h})
+    return flat

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

@@ -73,10 +73,10 @@ def _get_live():
 
 # Import token tracking
 try:
-    from ..telemetry.token_collector import TokenMetrics, _token_collector
+    from ..telemetry.token_collector import TokenMetrics, get_token_collector
 except ImportError:
     TokenMetrics = None
-    _token_collector = None
+    get_token_collector = None
 
 # Logging is already configured in _logging.py via __init__.py
 
@@ -4519,7 +4519,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
 
     def _track_token_usage(self, response: Dict[str, Any], model: str) -> Optional[TokenMetrics]:
         """Extract and track token usage from LLM response."""
-        if not TokenMetrics or not _token_collector:
+        if not TokenMetrics or not get_token_collector:
             return None
         
         # Note: metrics check moved to call sites for performance
@@ -4551,7 +4551,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
             # Track in global collector
             # Extract provider from model string (e.g., 'openai' from 'openai/gpt-4o-mini')
             provider = model.split('/')[0] if '/' in model else 'litellm'
-            _token_collector.track_tokens(
+            get_token_collector().track_tokens(
                 model=model,
                 agent=self.current_agent_name,
                 metrics=metrics,

src/praisonai-agents/praisonaiagents/scheduler/README.md

@@ -0,0 +1,29 @@
+# Scheduler — runner vs loop
+
+## ScheduleRunner (stateless)
+
+`ScheduleRunner` is **stateless**. Call `get_due_jobs()` whenever you want to check for jobs that should fire. It does not run background threads.
+
+Use when:
+- CLI one-shot tick (`praisonai schedule tick`)
+- Host integration that polls on its own timer
+- Tests
+
+## ScheduleLoop (optional daemon)
+
+`ScheduleLoop` is an **optional background daemon**. You must call `.start()` explicitly; it polls the runner and invokes `on_trigger` callbacks.
+
+Use when:
+- Long-running host process (Pattern B/C via `setup_bridges()`)
+- Standalone scheduler service
+
+```python
+from praisonaiagents.scheduler import FileScheduleStore, ScheduleRunner, ScheduleLoop
+
+store = FileScheduleStore()
+runner = ScheduleRunner(store)
+loop = ScheduleLoop(runner)
+loop.start()  # optional daemon
+```
+
+The wrapper host calls `ensure_schedule_runner()` which starts `ScheduleLoop` when the scheduler optional dependency is available.

src/praisonai-agents/praisonaiagents/session/README.md

@@ -32,7 +32,22 @@ When you provide a `session_id` to an Agent:
 
 ### Default Storage Location
 
-Sessions are stored in: `~/.praison/sessions/{session_id}.json`
+Sessions are stored in: `~/.praisonai/sessions/{session_id}.json`
+
+### Session metadata contract (Issue #48)
+
+List endpoints and UI dashboards may expose these optional root-level fields:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `model` | string | LLM model used in the session |
+| `total_tokens` | int | Cumulative input+output tokens |
+| `cost` | float | Estimated USD cost |
+| `agent_id` | string | Gateway or registry agent id |
+| `source` | string | Origin: `chat`, `gateway`, `cli`, `api` |
+| `agent_name` | string | Human-readable agent name |
+
+Fields are persisted in session JSON `metadata` and mirrored at root on save. Populated automatically after each assistant turn when using the default session store.
 
 ### Session File Format
 
@@ -45,7 +60,13 @@ Sessions are stored in: `~/.praison/sessions/{session_id}.json`
   ],
   "created_at": "2026-01-02T04:00:00+00:00",
   "updated_at": "2026-01-02T04:01:00+00:00",
-  "agent_name": "Assistant"
+  "agent_name": "Assistant",
+  "agent_id": "agent-abc",
+  "source": "chat",
+  "model": "gpt-4o-mini",
+  "total_tokens": 128,
+  "cost": 0.0004,
+  "metadata": {}
 }
 ```
 

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

@@ -129,3 +129,16 @@ class SessionStoreProtocol(Protocol):
             True if the session exists.
         """
         ...
+
+
+@runtime_checkable
+class CheckpointQueryProtocol(Protocol):
+    """Read path for session checkpoints / rollback snapshots."""
+
+    def list_checkpoints(self, session_id: str) -> List[Dict[str, Any]]:
+        """Return checkpoint metadata for a session."""
+        ...
+
+    def get_checkpoint(self, session_id: str, checkpoint_id: str) -> Optional[Dict[str, Any]]:
+        """Return a single checkpoint payload."""
+        ...

src/praisonai-agents/praisonaiagents/session/store.py

@@ -81,7 +81,7 @@ class SessionData:
     
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary for JSON serialization."""
-        return {
+        data = {
             "session_id": self.session_id,
             "messages": [m.to_dict() for m in self.messages],
             "created_at": self.created_at,
@@ -92,6 +92,10 @@ class SessionData:
             "gateway_session_id": self.gateway_session_id,
             "agent_id": self.agent_id,
         }
+        for key in ("model", "llm", "total_tokens", "token_count", "cost", "source"):
+            if key in self.metadata:
+                data[key] = self.metadata[key]
+        return data
     
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "SessionData":
@@ -470,7 +474,20 @@ class DefaultSessionStore:
             self._cache[session_id] = session
         
         return self._save_session(session)
-    
+
+    def update_session_metadata(self, session_id: str, **fields: Any) -> bool:
+        """Merge run stats / metadata fields into a persisted session."""
+        session = self._load_session(session_id)
+        with self._lock:
+            for key, value in fields.items():
+                if value is None:
+                    continue
+                session.metadata[key] = value
+                if key in ("agent_id", "agent_name", "user_id"):
+                    setattr(session, key, value)
+            self._cache[session_id] = session
+        return self._save_session(session)
+
     def delete_session(self, session_id: str) -> bool:
         """Delete a session completely."""
         filepath = self._get_session_path(session_id)
@@ -500,10 +517,16 @@ class DefaultSessionStore:
                             data = json.load(f)
                         sessions.append({
                             "session_id": data.get("session_id", filename[:-5]),
+                            "id": data.get("session_id", filename[:-5]),
                             "agent_name": data.get("agent_name"),
+                            "agent_id": data.get("agent_id") or (data.get("metadata") or {}).get("agent_id"),
+                            "source": data.get("source") or (data.get("metadata") or {}).get("source"),
                             "created_at": data.get("created_at"),
                             "updated_at": data.get("updated_at"),
                             "message_count": len(data.get("messages", [])),
+                            "model": data.get("model") or data.get("llm") or (data.get("metadata") or {}).get("model"),
+                            "total_tokens": data.get("total_tokens") or data.get("token_count") or (data.get("metadata") or {}).get("total_tokens"),
+                            "cost": data.get("cost") or (data.get("metadata") or {}).get("cost"),
                         })
                     except (json.JSONDecodeError, IOError):
                         continue

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

@@ -7,7 +7,7 @@ plugin bundle, enterprise policy) without touching the core.
 
 from __future__ import annotations
 
-from typing import Protocol, Iterable, Optional, runtime_checkable
+from typing import Protocol, Iterable, Optional, runtime_checkable, List, Dict, Any
 
 from .models import SkillProperties
 
@@ -45,7 +45,8 @@ class SkillMutatorProtocol(Protocol):
     """Agent-managed skill CRUD operations.
     
     Allows agents to create, edit, and manage their own skills at runtime.
-    Implementations should provide safe-by-default behavior (e.g., propose mode).
+    Implementations should provide safe-by-default behaviour: use ``propose=True``
+    (default) to stage mutations for human approval before writing to disk.
     """
 
     def create(self, name: str, content: str, category: Optional[str] = None,
@@ -163,3 +164,30 @@ class SkillMutatorProtocol(Protocol):
             Status message describing the action taken
         """
         ...
+
+
+@runtime_checkable
+class SkillsCatalogProtocol(Protocol):
+    """Metadata listing for skills — UI catalog pages."""
+
+    def list_skills(self) -> List[Dict[str, Any]]:
+        """Return skill metadata dicts (name, description, location)."""
+        ...
+
+
+def list_skills_for_api() -> List[Dict[str, Any]]:
+    """Default catalog adapter using SkillManager when available."""
+    try:
+        from .manager import SkillManager
+
+        manager = SkillManager()
+        return [
+            {
+                "name": s.name,
+                "description": s.description,
+                "location": getattr(s, "location", ""),
+            }
+            for s in manager.get_available_skills()
+        ]
+    except Exception:
+        return []

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

@@ -100,3 +100,107 @@ class InMemoryTokenUsageSink:
     def clear(self) -> None:
         """Clear all records."""
         self.records.clear()
+
+
+@runtime_checkable
+class UsageQueryProtocol(Protocol):
+    """Read path for token usage — UI dashboards and analytics."""
+
+    def get_summary(self) -> Dict[str, Any]:
+        """Aggregate totals and breakdowns by model/agent."""
+        ...
+
+    def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
+        """Recent usage records, newest last."""
+        ...
+
+    def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
+        ...
+
+    def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
+        ...
+
+
+class InMemoryUsageQuery:
+    """Query adapter backed by InMemoryTokenUsageSink records."""
+
+    def __init__(self, sink: InMemoryTokenUsageSink):
+        self._sink = sink
+
+    def get_summary(self) -> Dict[str, Any]:
+        records = self._sink.records
+        total_in = 0
+        total_out = 0
+        total_all = 0
+        by_model: Dict[str, Dict[str, int]] = {}
+        by_agent: Dict[str, Dict[str, int]] = {}
+        
+        # Single-pass optimization: calculate everything in one loop
+        for r in records:
+            r_in = r.get("input_tokens", 0)
+            r_out = r.get("output_tokens", 0)
+            r_total = r.get("total_tokens", r_in + r_out)
+            
+            total_in += r_in
+            total_out += r_out
+            total_all += r_total
+            
+            model = r.get("model", "unknown")
+            agent = r.get("agent_name", "unknown")
+            
+            # Update by_model and by_agent efficiently
+            for bucket, key in ((by_model, model), (by_agent, agent)):
+                stats = bucket.setdefault(key, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0})
+                stats["input_tokens"] += r_in
+                stats["output_tokens"] += r_out
+                stats["total_tokens"] += r_total
+        return {
+            "total_requests": len(records),
+            "total_input_tokens": total_in,
+            "total_output_tokens": total_out,
+            "total_tokens": total_all,  # Use consistent calculation
+            "by_model": by_model,
+            "by_agent": by_agent,
+        }
+
+    def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
+        return self._sink.records[-limit:]
+
+    def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
+        return self._sink.get_by_task(task_id)
+
+    def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
+        return self._sink.get_by_agent(agent_name)
+
+
+class TokenCollectorUsageQuery:
+    """Query adapter reading from the global TokenCollector + optional sink."""
+
+    def __init__(self, collector: Any):
+        self._collector = collector
+
+    def get_summary(self) -> Dict[str, Any]:
+        summary = self._collector.get_session_summary()
+        return {
+            "total_requests": summary.get("total_interactions", 0),
+            "total_input_tokens": summary.get("total_metrics", {}).get("input_tokens", 0),
+            "total_output_tokens": summary.get("total_metrics", {}).get("output_tokens", 0),
+            "total_tokens": summary.get("total_tokens", 0),
+            "by_model": summary.get("by_model", {}),
+            "by_agent": summary.get("by_agent", {}),
+        }
+
+    def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
+        return self._collector.get_recent_interactions(limit)
+
+    def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
+        sink = getattr(self._collector, "_sink", None)
+        if sink and hasattr(sink, "get_by_task"):
+            return sink.get_by_task(task_id)
+        return []
+
+    def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
+        sink = getattr(self._collector, "_sink", None)
+        if sink and hasattr(sink, "get_by_agent"):
+            return sink.get_by_agent(agent_name)
+        return []

src/praisonai-agents/praisonaiagents/telemetry/token_collector.py

@@ -191,4 +191,9 @@ class TokenCollector:
 
 
 # Global token collector instance
-_token_collector = TokenCollector()
+_token_collector = TokenCollector()
+
+
+def get_token_collector() -> TokenCollector:
+    """Return the global TokenCollector singleton."""
+    return _token_collector

src/praisonai-agents/praisonaiagents/telemetry/token_telemetry.py

@@ -7,7 +7,7 @@ import logging
 from praisonaiagents._logging import get_logger
 
 # Import dependencies
-from .token_collector import _token_collector, TokenMetrics
+from .token_collector import get_token_collector, TokenMetrics
 from .telemetry import get_telemetry
 
 logger = get_logger(__name__)
@@ -56,11 +56,12 @@ class TokenTelemetryBridge:
     
     def export_token_metrics(self) -> Dict[str, Any]:
         """Export token metrics for telemetry reporting."""
-        if not _token_collector:
+        collector = get_token_collector()
+        if not collector:
             return {}
         
         try:
-            metrics = _token_collector.export_metrics()
+            metrics = collector.export_metrics()
             
             # Prepare telemetry-friendly format
             return {
@@ -77,8 +78,7 @@ class TokenTelemetryBridge:
     
     def reset_token_metrics(self):
         """Reset token metrics collection."""
-        if _token_collector:
-            _token_collector.reset()
+        get_token_collector().reset()
 
 # Global telemetry bridge instance
 _token_telemetry_bridge = TokenTelemetryBridge()

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

@@ -14,10 +14,31 @@ import json
 from typing import Any, Dict, List, Union
 
 from praisonaiagents.tools.decorator import tool
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
+
+
+def _unwrap_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> List[Dict[str, Any]]:
+    """Minimal unwrap for common LLM arg shapes (not full UI normalisation)."""
+    if isinstance(messages, str):
+        try:
+            messages = json.loads(messages)
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"messages JSON string is not valid JSON: {exc}") from exc
+
+    if isinstance(messages, dict):
+        if "messages" in messages:
+            messages = messages["messages"]
+        else:
+            messages = [messages]
+
+    if not isinstance(messages, list):
+        raise ValueError("messages must be a list of A2UI message dicts or a JSON string")
+
+    return messages
 
 
 @tool
-def send_a2ui_messages(messages: Union[List[Dict[str, Any]], str]) -> Dict[str, Any]:
+def send_a2ui_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> Dict[str, Any]:
     """
     Send validated A2UI JSON messages to a connected A2UI renderer client.
 
@@ -32,21 +53,14 @@ def send_a2ui_messages(messages: Union[List[Dict[str, Any]], str]) -> Dict[str,
     """
     from praisonaiagents.ui.a2ui.adapter import create_a2ui_part
 
-    if isinstance(messages, str):
-        try:
-            messages = json.loads(messages)
-        except json.JSONDecodeError as exc:
-            raise ValueError(f"messages JSON string is not valid JSON: {exc}") from exc
-
-    if not isinstance(messages, list):
-        raise ValueError("messages must be a list of A2UI message dicts or a JSON string")
+    messages = _unwrap_messages(messages)
 
     part = create_a2ui_part({"messages": messages})
     data = getattr(getattr(part, "root", part), "data", messages)
     metadata = getattr(getattr(part, "root", part), "metadata", {}) or {}
 
     return {
-        "mime_type": metadata.get("mimeType", "application/json+a2ui"),
+        "mime_type": metadata.get("mimeType", A2UI_MIME_TYPE),
         "messages": messages,
         "a2ui_part": data,
     }

src/praisonai-agents/praisonaiagents/ui/a2a/a2a.py

@@ -253,7 +253,7 @@ class A2A:
         from praisonaiagents.ui.a2a.conversion import (
             extract_user_input,
             praisonai_to_a2a_message,
-            create_artifact,
+            create_response_artifact,
         )
         
         # Validate params
@@ -286,7 +286,8 @@ class A2A:
                     else:
                         raise RuntimeError("No agent or agents configured")
                     response_msg = praisonai_to_a2a_message(str(response), task_id=task_id)
-                    artifact = create_artifact(str(response))
+                    history = getattr(self.agent, "chat_history", None) if self.agent else None
+                    artifact = create_response_artifact(str(response), chat_history=history)
                     self.task_store.add_to_history(task_id, response_msg)
                     self.task_store.add_artifact(task_id, artifact)
                     self.task_store.update_status(task_id, TaskState.COMPLETED)
@@ -322,7 +323,8 @@ class A2A:
             response_msg = praisonai_to_a2a_message(
                 str(response), task_id=task.id
             )
-            artifact = create_artifact(str(response))
+            history = getattr(self.agent, "chat_history", None) if self.agent else None
+            artifact = create_response_artifact(str(response), chat_history=history)
             
             # Update task store
             self.task_store.add_to_history(task.id, response_msg)

src/praisonai-agents/praisonaiagents/ui/a2a/conversion.py

@@ -16,6 +16,7 @@ from praisonaiagents.ui.a2a.types import (
     DataPart,
     Artifact,
 )
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
 
 
 def a2a_to_praisonai_messages(messages: List[Message]) -> List[Dict[str, Any]]:
@@ -107,6 +108,75 @@ def praisonai_to_a2a_message(
     )
 
 
+def find_a2ui_tool_result(
+    chat_history: Optional[List[Dict[str, Any]]],
+) -> Optional[Dict[str, Any]]:
+    """Return the last A2UI tool result dict from agent chat history, if any."""
+    if not chat_history:
+        return None
+    for msg in reversed(chat_history):
+        if msg.get("role") != "tool":
+            continue
+        content = msg.get("content")
+        if content is None:
+            continue
+        data: Any = content if isinstance(content, dict) else None
+        if data is None:
+            try:
+                data = json.loads(content)
+            except (json.JSONDecodeError, TypeError):
+                continue
+        if isinstance(data, dict) and data.get("mime_type") == A2UI_MIME_TYPE:
+            return data
+    return None
+
+
+def create_a2ui_artifact(
+    result: Dict[str, Any],
+    artifact_id: Optional[str] = None,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+) -> Artifact:
+    """Create an A2A Artifact with a DataPart for an A2UI tool result."""
+    art_id = artifact_id or f"art-{uuid.uuid4().hex[:12]}"
+    messages = result.get("messages") or []
+    a2ui_data = result.get("a2ui_part")
+    if not isinstance(a2ui_data, dict):
+        a2ui_data = {"messages": messages}
+    elif "messages" not in a2ui_data:
+        a2ui_data = {**a2ui_data, "messages": messages}
+
+    return Artifact(
+        artifact_id=art_id,
+        name=name,
+        description=description,
+        parts=[
+            DataPart(
+                data=a2ui_data,
+                metadata={"mimeType": result.get("mime_type", A2UI_MIME_TYPE)},
+            )
+        ],
+    )
+
+
+def create_response_artifact(
+    content: str,
+    chat_history: Optional[List[Dict[str, Any]]] = None,
+    artifact_id: Optional[str] = None,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+) -> Artifact:
+    """Build artifact from response text, preferring A2UI DataPart when present in history."""
+    a2ui = find_a2ui_tool_result(chat_history)
+    if a2ui:
+        return create_a2ui_artifact(
+            a2ui, artifact_id=artifact_id, name=name, description=description
+        )
+    return create_artifact(
+        content, artifact_id=artifact_id, name=name, description=description
+    )
+
+
 def create_artifact(
     content: str,
     artifact_id: Optional[str] = None,

src/praisonai-agents/praisonaiagents/ui/a2ui/__init__.py

@@ -12,9 +12,12 @@ from praisonaiagents.ui.a2ui.adapter import (
     is_a2ui_part,
     parse_a2ui_response,
 )
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
 
 __all__ = [
     "A2UI",
+    "A2UI_MIME_TYPE",
+    "A2UIToolResultProtocol",
     "create_a2ui_part",
     "generate_a2ui_system_prompt",
     "get_schema_manager",

src/praisonai-agents/praisonaiagents/ui/agui/streaming.py

@@ -12,6 +12,7 @@ from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Set
 
 from praisonaiagents.ui.agui.types import (
     BaseEvent,
+    CustomEvent,
     TextMessageStartEvent,
     TextMessageContentEvent,
     TextMessageEndEvent,
@@ -26,6 +27,7 @@ from praisonaiagents.ui.agui.types import (
     StepFinishedEvent,
     StateSnapshotEvent,
 )
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
 
 
 @dataclass
@@ -292,13 +294,41 @@ def stream_event_to_agui_events(
 
     if event.type == StreamEventType.TOOL_CALL_RESULT and event.tool_call:
         tool_call_id = event.tool_call.get("id") or str(uuid.uuid4())
-        content = event.content if event.content is not None else str(event.tool_call.get("result", ""))
+        raw_result = event.tool_call.get("result")
+        content = event.content if event.content is not None else str(raw_result or "")
         buffer.end_tool_call(tool_call_id)
-        return [create_tool_result_event(tool_call_id, str(content), message_id)]
+        events: List[BaseEvent] = [
+            create_tool_result_event(tool_call_id, str(content), message_id)
+        ]
+        if isinstance(raw_result, dict) and raw_result.get("mime_type") == A2UI_MIME_TYPE:
+            events.append(
+                CustomEvent(
+                    name="a2ui",
+                    value={
+                        "mime_type": A2UI_MIME_TYPE,
+                        "messages": raw_result.get("messages", []),
+                        "surface_id": _infer_a2ui_surface_id(raw_result),
+                    },
+                )
+            )
+        return events
 
     return []
 
 
+def _infer_a2ui_surface_id(result: Dict[str, Any]) -> str:
+    """Infer surface id from send_a2ui_messages result (inline, no UI-layer deps)."""
+    for msg in result.get("messages") or []:
+        if isinstance(msg, dict):
+            create = msg.get("createSurface")
+            if isinstance(create, dict):
+                sid = create.get("surfaceId") or create.get("id")
+                if sid:
+                    return str(sid)
+    sid = result.get("surface_id") or result.get("surfaceId")
+    return str(sid) if sid else "main"
+
+
 async def async_stream_agent_response(
     agent,
     user_input: str,

src/praisonai-agents/praisonaiagents/ui/agui/types.py

@@ -133,6 +133,13 @@ class StateSnapshotEvent(BaseEvent):
     snapshot: Dict[str, Any]
 
 
+class CustomEvent(BaseEvent):
+    """Custom extension event (e.g. structured A2UI payloads)."""
+    type: Literal["CUSTOM"] = "CUSTOM"
+    name: str
+    value: Any
+
+
 # Message Types
 class FunctionCall(BaseModel):
     """Function call in a tool call."""
@@ -175,6 +182,7 @@ Event = Union[
     ToolCallEndEvent,
     ToolCallResultEvent,
     StateSnapshotEvent,
+    CustomEvent,
     RunStartedEvent,
     RunFinishedEvent,
     RunErrorEvent,

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

@@ -0,0 +1,26 @@
+"""
+UI protocol types — lightweight contracts for agent-to-UI integration.
+
+Zero runtime cost: TypedDict and constants only. No optional dependencies.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TypedDict, Union
+
+# Standard MIME type for A2UI payloads (Google A2UI / A2A transport).
+A2UI_MIME_TYPE = "application/json+a2ui"
+
+
+class A2UIToolResultProtocol(TypedDict, total=False):
+    """
+    Output shape from ``send_a2ui_messages`` — the integrator contract for any UI.
+
+    Example:
+        if result.get("mime_type") == A2UI_MIME_TYPE:
+            render(result["messages"])
+    """
+
+    mime_type: str
+    messages: List[Dict[str, Any]]
+    a2ui_part: Union[Dict[str, Any], List[Any]]

src/praisonai-agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "1.6.43"
+version = "1.6.45"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 readme = "README.md"
 requires-python = ">=3.10"

src/praisonai-agents/tests/a2a/test_a2a.py

@@ -233,6 +233,47 @@ class TestA2AJsonRpc:
         assert result["status"]["state"] == "completed"
         assert result["artifacts"] is not None
         assert len(result["artifacts"]) > 0
+
+    def test_post_a2a_message_send_a2ui_artifact(self):
+        """message/send with A2UI tool result in chat_history → DataPart artifact."""
+        import json
+
+        from fastapi.testclient import TestClient
+
+        from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
+
+        app, a2a = _make_app()
+        a2ui_payload = {
+            "mime_type": A2UI_MIME_TYPE,
+            "messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
+            "a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
+        }
+        a2a.agent.chat = MagicMock(return_value="Surface ready.")
+        a2a.agent.chat_history = [
+            {"role": "tool", "content": json.dumps(a2ui_payload)},
+        ]
+
+        client = TestClient(app)
+        response = client.post("/a2a", json={
+            "jsonrpc": "2.0",
+            "method": "message/send",
+            "id": "a2ui-1",
+            "params": {
+                "message": {
+                    "messageId": "msg-a2ui",
+                    "role": "user",
+                    "parts": [{"text": "Show UI"}],
+                }
+            },
+        })
+
+        assert response.status_code == 200
+        result = response.json()["result"]
+        artifact = result["artifacts"][0]
+        parts = artifact["parts"]
+        data_parts = [p for p in parts if "data" in p]
+        assert data_parts, "Expected DataPart in artifact"
+        assert data_parts[0]["metadata"]["mimeType"] == A2UI_MIME_TYPE
     
     def test_post_a2a_message_send_creates_task_in_store(self):
         """message/send should create a task in TaskStore."""

src/praisonai-agents/tests/a2a/test_a2ui_conversion.py

@@ -0,0 +1,66 @@
+"""Tests for A2UI artifact conversion in A2A transport."""
+
+import json
+
+import pytest
+
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
+
+
+SAMPLE_A2UI_RESULT = {
+    "mime_type": A2UI_MIME_TYPE,
+    "messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
+    "a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
+}
+
+
+class TestFindA2uiToolResult:
+    def test_finds_last_a2ui_tool_message(self):
+        from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
+
+        history = [
+            {"role": "user", "content": "show ui"},
+            {"role": "tool", "content": json.dumps({"mime_type": "text/plain", "data": "x"})},
+            {"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)},
+        ]
+        result = find_a2ui_tool_result(history)
+        assert result is not None
+        assert result["mime_type"] == A2UI_MIME_TYPE
+
+    def test_returns_none_without_a2ui(self):
+        from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
+
+        assert find_a2ui_tool_result([]) is None
+        assert find_a2ui_tool_result([{"role": "tool", "content": "plain text"}]) is None
+
+
+class TestCreateA2uiArtifact:
+    def test_creates_datapart_with_mime(self):
+        from praisonaiagents.ui.a2a.conversion import create_a2ui_artifact
+        from praisonaiagents.ui.a2a.types import DataPart
+
+        artifact = create_a2ui_artifact(SAMPLE_A2UI_RESULT, name="ui-surface")
+        assert artifact.name == "ui-surface"
+        assert len(artifact.parts) == 1
+        part = artifact.parts[0]
+        assert isinstance(part, DataPart)
+        assert part.metadata.get("mimeType") == A2UI_MIME_TYPE
+        assert "messages" in part.data
+
+
+class TestCreateResponseArtifact:
+    def test_prefers_a2ui_from_history(self):
+        from praisonaiagents.ui.a2a.conversion import create_response_artifact
+        from praisonaiagents.ui.a2a.types import DataPart
+
+        history = [{"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)}]
+        artifact = create_response_artifact("Done.", chat_history=history)
+        assert isinstance(artifact.parts[0], DataPart)
+
+    def test_falls_back_to_text(self):
+        from praisonaiagents.ui.a2a.conversion import create_response_artifact
+        from praisonaiagents.ui.a2a.types import TextPart
+
+        artifact = create_response_artifact("Hello", chat_history=[])
+        assert isinstance(artifact.parts[0], TextPart)
+        assert artifact.parts[0].text == "Hello"

src/praisonai-agents/tests/a2ui/test_a2ui_tools.py

@@ -0,0 +1,51 @@
+"""Tests for send_a2ui_messages minimal arg unwrap."""
+
+import json
+
+import pytest
+
+from praisonaiagents.tools.a2ui_tools import _unwrap_messages
+
+try:
+    import a2ui  # noqa: F401
+
+    HAS_A2UI_SDK = True
+except ImportError:
+    HAS_A2UI_SDK = False
+
+SAMPLE_MSG = {"createSurface": {"surfaceId": "main", "catalogId": "basic"}}
+
+
+class TestUnwrapMessages:
+    def test_list_input(self):
+        assert _unwrap_messages([SAMPLE_MSG]) == [SAMPLE_MSG]
+
+    def test_json_string_list(self):
+        assert _unwrap_messages(json.dumps([SAMPLE_MSG])) == [SAMPLE_MSG]
+
+    def test_wrapped_messages_dict(self):
+        assert _unwrap_messages({"messages": [SAMPLE_MSG]}) == [SAMPLE_MSG]
+
+    def test_json_string_wrapped_object(self):
+        assert _unwrap_messages(json.dumps({"messages": [SAMPLE_MSG]})) == [SAMPLE_MSG]
+
+    def test_single_message_dict(self):
+        assert _unwrap_messages(SAMPLE_MSG) == [SAMPLE_MSG]
+
+    def test_invalid_json_raises(self):
+        with pytest.raises(ValueError, match="not valid JSON"):
+            _unwrap_messages("{not json")
+
+    def test_invalid_shape_raises(self):
+        with pytest.raises(ValueError, match="must be"):
+            _unwrap_messages(123)
+
+
+@pytest.mark.skipif(not HAS_A2UI_SDK, reason="a2ui-agent-sdk not installed")
+class TestSendA2uiMessagesIntegration:
+    def test_full_tool_call(self):
+        from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
+
+        result = send_a2ui_messages(messages=[SAMPLE_MSG])
+        assert result["mime_type"] == "application/json+a2ui"
+        assert len(result["messages"]) == 1

src/praisonai-agents/tests/a2ui/test_adapter.py

@@ -34,3 +34,27 @@ class TestA2UITool:
             )
             assert result["mime_type"] == "application/json+a2ui"
             assert len(result["messages"]) == 1
+
+
+class TestA2UIAdapterSmoke:
+    def test_is_a2ui_part_with_sdk(self):
+        pytest.importorskip("a2ui", reason="a2ui-agent-sdk not installed")
+        from praisonaiagents.ui.a2ui.adapter import create_a2ui_part, is_a2ui_part
+
+        part = create_a2ui_part({"messages": [{"createSurface": {"surfaceId": "m"}}]})
+        assert is_a2ui_part(part) is True
+
+    def test_missing_sdk_import_error_message(self):
+        from unittest.mock import patch
+
+        from praisonaiagents.ui.a2ui import adapter
+
+        def _raise(_name):
+            raise ImportError(
+                "A2UI support requires the optional dependency. "
+                "Install with: pip install praisonaiagents[a2ui]"
+            )
+
+        with patch.object(adapter, "_import", side_effect=_raise):
+            with pytest.raises(ImportError, match="praisonaiagents\\[a2ui\\]"):
+                adapter.create_a2ui_part({"messages": []})

src/praisonai-agents/tests/a2ui/test_protocols.py

@@ -0,0 +1,27 @@
+"""Tests for UI protocol types (zero-dep contracts)."""
+
+from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
+
+
+class TestA2UIProtocols:
+    def test_mime_type_constant(self):
+        assert A2UI_MIME_TYPE == "application/json+a2ui"
+
+    def test_tool_result_protocol_shape(self):
+        result: A2UIToolResultProtocol = {
+            "mime_type": A2UI_MIME_TYPE,
+            "messages": [{"createSurface": {"surfaceId": "main"}}],
+            "a2ui_part": {"messages": []},
+        }
+        assert result["mime_type"] == A2UI_MIME_TYPE
+
+    def test_exported_from_a2ui_package(self):
+        from praisonaiagents.ui.a2ui import A2UI_MIME_TYPE, A2UIToolResultProtocol
+
+        assert A2UI_MIME_TYPE == "application/json+a2ui"
+        assert A2UIToolResultProtocol is not None
+
+    def test_top_level_lazy_a2ui_export(self):
+        from praisonaiagents import A2UI
+
+        assert A2UI is not None

src/praisonai-agents/tests/agui/test_streaming.py

@@ -129,6 +129,29 @@ class TestStreamEventBridge:
         assert agui_events[0].type == "TOOL_CALL_START"
         assert agui_events[0].tool_call_name == "search"
 
+    def test_tool_call_result_a2ui_emits_custom_event(self):
+        from praisonaiagents.streaming.events import StreamEvent, StreamEventType
+        from praisonaiagents.ui.agui.streaming import EventBuffer, stream_event_to_agui_events
+        from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
+
+        buffer = EventBuffer()
+        a2ui_result = {
+            "mime_type": A2UI_MIME_TYPE,
+            "messages": [{"createSurface": {"surfaceId": "panel-1"}}],
+        }
+        event = StreamEvent(
+            type=StreamEventType.TOOL_CALL_RESULT,
+            tool_call={"id": "tc-a2ui", "name": "send_a2ui_messages", "result": a2ui_result},
+        )
+        agui_events = stream_event_to_agui_events(event, "msg-1", buffer)
+
+        assert len(agui_events) == 2
+        assert agui_events[0].type == "TOOL_CALL_RESULT"
+        assert agui_events[1].type == "CUSTOM"
+        assert agui_events[1].name == "a2ui"
+        assert agui_events[1].value["surface_id"] == "panel-1"
+        assert agui_events[1].value["messages"] == a2ui_result["messages"]
+
 
 class TestRunLifecycleEvents:
     """Test run lifecycle events."""

src/praisonai-agents/tests/unit/streaming/test_tool_call_events.py

@@ -225,3 +225,19 @@ class TestToolCallIdPropagation:
         )
         
         assert event.tool_call["id"] == "call_abc123xyz"
+
+
+class TestToolExecutionMixinStreamEmitterLookup:
+    def test_mixin_resolves_name_mangled_stream_emitter_without_agent_class_name(self):
+        from praisonaiagents.agent.tool_execution import ToolExecutionMixin
+        from praisonaiagents.streaming.events import StreamEventEmitter
+
+        class CustomAgent(ToolExecutionMixin):
+            def __init__(self):
+                self.__stream_emitter = StreamEventEmitter()
+
+        agent = CustomAgent()
+        emitter = agent._get_existing_stream_emitter()
+
+        assert emitter is not None
+        assert isinstance(emitter, StreamEventEmitter)

src/praisonai-agents/tests/unit/telemetry/test_token_usage_sink.py

@@ -134,3 +134,37 @@ class TestTokenCollectorSinkIntegration:
         )
         assert sink.records[0]["task_id"] == "task-abc"
         assert sink.records[0]["agent_name"] == "researcher"
+
+
+class TestInMemoryUsageQuery:
+    """Test query adapter backed by in-memory sink records."""
+
+    def test_get_summary_uses_total_tokens_when_present(self):
+        from praisonaiagents.telemetry.protocols import InMemoryTokenUsageSink, InMemoryUsageQuery
+
+        sink = InMemoryTokenUsageSink()
+        sink.records.extend([
+            {
+                "task_id": "t1",
+                "agent_name": "a1",
+                "model": "m1",
+                "input_tokens": 10,
+                "output_tokens": 5,
+                "total_tokens": 50,
+            },
+            {
+                "task_id": "t2",
+                "agent_name": "a1",
+                "model": "m2",
+                "input_tokens": 2,
+                "output_tokens": 3,
+            },
+        ])
+
+        summary = InMemoryUsageQuery(sink).get_summary()
+
+        assert summary["total_input_tokens"] == 12
+        assert summary["total_output_tokens"] == 8
+        assert summary["total_tokens"] == 55
+        assert summary["by_model"]["m1"]["total_tokens"] == 50
+        assert summary["by_model"]["m2"]["total_tokens"] == 5

src/praisonai-agents/uv.lock

@@ -3125,7 +3125,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "1.6.43"
+version = "1.6.45"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai-rust/FEATURE_PARITY_TRACKER.json

@@ -1,6 +1,6 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-05-20",
+  "lastUpdated": "2026-05-21",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "status": "PARITY_ACHIEVED",

src/praisonai-ts/FEATURE_PARITY_TRACKER.json

@@ -1,6 +1,6 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-05-20",
+  "lastUpdated": "2026-05-21",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "summary": {

src/praisonai-ts/PARITY.md

@@ -1,6 +1,6 @@
 # Feature Parity Tracker
 
-> **Version:** 1.5.87 | **Last Updated:** 2026-05-20
+> **Version:** 1.5.87 | **Last Updated:** 2026-05-21
 > **Source of Truth:** Python SDK (praisonaiagents)
 
 ## Summary

src/praisonai/docs/channels-gateway.md

@@ -0,0 +1,31 @@
+# Channels gateway integration
+
+Channel bots (Telegram, Discord, Slack, WhatsApp) require the **`praisonai[bot]`** optional extra:
+
+```bash
+pip install "praisonai[bot]"
+```
+
+## Architecture
+
+- **Pattern B/C host**: `configure_host()` + `AIUIGateway` serve chat UI and REST APIs.
+- **Channels feature** (`praisonaiui.features.channels`): starts platform bots via `praisonai.bots`.
+- **Gateway WebSocket**: `/ws` on the same process when using Pattern C.
+
+## Environment variables
+
+| Platform | Variables |
+|----------|-------------|
+| Telegram | `TELEGRAM_BOT_TOKEN` |
+| Discord | `DISCORD_BOT_TOKEN` |
+| Slack | `SLACK_BOT_TOKEN`, `SLACK_APP_TOKEN` |
+| WhatsApp | `WHATSAPP_ACCESS_TOKEN`, `WHATSAPP_PHONE_NUMBER_ID` |
+
+## CLI
+
+```bash
+praisonai dashboard --aiui          # Pattern B in-process host
+PRAISONAI_HOST_LEGACY=1 praisonai ui  # Legacy @aiui.reply callbacks only
+```
+
+See `praisonai.bots.BotOS` for multi-platform orchestration.

src/praisonai/docs/pattern-d-platform.md

@@ -0,0 +1,26 @@
+# Pattern D — Platform API integration (defer)
+
+Pattern D connects PraisonAIUI dashboard pages to **PraisonAI Cloud** via `PlatformClient` `/api/v1`.
+
+## Scope (P3)
+
+| Item | Status |
+|------|--------|
+| aiui pages → PlatformClient | Documented, optional |
+| Platform JWT auth | Optional `auth.platform_jwt` config |
+| Issues / kanban pages | Optional `@aiui.page` via platform |
+| agent_id linking | See below |
+
+## agent_id linking
+
+Three registries may coexist:
+
+1. **Platform roster** — cloud tenant agents (`/api/v1/agents`)
+2. **Gateway registry** — WebSocket `/ws` agent ids
+3. **aiui SDKAgentRegistry** — YAML CRUD + local `Agent` instances
+
+Use explicit `agent_id` in session metadata (`source=gateway|platform|ui`) to correlate rows in the workflow-runs and sessions tables.
+
+## Optional platform JWT
+
+When `PRAISONAI_PLATFORM_TOKEN` is set, future aiui auth middleware may validate JWT from the platform issuer. Not required for Patterns B/C self-hosted installs.

src/praisonai/praisonai.rb

@@ -3,8 +3,8 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.43.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.43.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.45.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.45.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/__init__.py

@@ -6,6 +6,8 @@ from .version import __version__
 # Define __all__ for lazy loading
 __all__ = [
     'PraisonAI',
+    'run',
+    'arun',
     '__version__',
     'Deploy',
     'DeployConfig',
@@ -92,7 +94,13 @@ def __getattr__(name):
     # Note: Telemetry initialization moved out of lazy hook to avoid side effects
     # It should be called explicitly from cli.PraisonAI.__init__ instead
 
-    if name == 'PraisonAI':
+    if name == 'run':
+        from ._entrypoint import run
+        return run
+    elif name == 'arun':
+        from ._entrypoint import arun
+        return arun
+    elif name == 'PraisonAI':
         from .cli import PraisonAI
         return PraisonAI
     elif name == 'Agent':
@@ -130,9 +138,12 @@ def __getattr__(name):
     elif name == 'EmbeddingResult':
         from praisonaiagents.embedding import EmbeddingResult
         return EmbeddingResult
-    elif name == 'AgentOS':
-        from .app import AgentOS
-        return AgentOS
+    elif name == 'build_host_app':
+        from .integration.host_app import build_host_app
+        return build_host_app
+    elif name == 'configure_host':
+        from .integration.host_app import configure_host
+        return configure_host
     elif name == 'AgentApp':
         # Silent alias for AgentOS (backward compatibility)
         from .app import AgentOS

src/praisonai/praisonai/_dev/api_md.py

@@ -119,7 +119,7 @@ class ApiMdGenerator:
         'Observability': ['obs', 'FlowDisplay', 'track_workflow'],
         'Context': ['FastContext', 'FastContextResult', 'FileMatch', 'LineRange',
                     'ContextManager', 'ManagerConfig'],
-        'UI': ['AGUI', 'A2A'],
+        'UI': ['AGUI', 'A2A', 'A2UI'],
         'Config': ['MemoryConfig', 'KnowledgeConfig', 'PlanningConfig', 'ReflectionConfig',
                    'GuardrailConfig', 'WebConfig', 'OutputConfig', 'ExecutionConfig',
                    'TemplateConfig', 'CachingConfig', 'HooksConfig', 'SkillsConfig',

src/praisonai/praisonai/_dev/parity/python_extractor.py

@@ -82,7 +82,7 @@ class PythonFeatureExtractor:
         'telemetry': ['telemetry'],
         'context': ['context.fast', 'context.manager', 'context.models'],
         'session': ['session'],
-        'ui': ['ui.agui', 'ui.a2a'],
+        'ui': ['ui.agui', 'ui.a2a', 'ui.a2ui'],
         'rag': ['rag'],
         'planning': ['planning'],
         'config': ['config.feature_configs', 'config.param_resolver', 

src/praisonai/praisonai/_entrypoint.py

@@ -0,0 +1,73 @@
+"""Top-level convenience entry points for the Python API.
+
+Keep this file minimal - it is part of the Minimal API surface and must defer
+all heavy imports to call time.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+
+def _build_config_list() -> list[dict[str, Any]]:
+    """Reuse the same env/keyfile resolution the CLI already performs."""
+    # Import LLM endpoint resolution from the same path the CLI uses
+    from praisonai.llm.env import resolve_llm_endpoint
+    ep = resolve_llm_endpoint()
+    
+    return [{
+        'model': ep.model,
+        'base_url': ep.base_url,
+        'api_key': ep.api_key,
+        'api_type': 'openai'        # AutoGen expects this field
+    }]
+
+
+def run(agent_file: str,
+        framework: str | None = None,
+        *,
+        tools: list | None = None,
+        agent_yaml: str | None = None,
+        cli_config: dict | None = None) -> str:
+    """One-line Python entry point. Equivalent to `praisonai <agent_file>`."""
+    from .agents_generator import AgentsGenerator
+    from ._framework_availability import is_available
+
+    if framework is None:
+        # Mirror the CLI default-resolution order
+        for candidate in ("crewai", "praisonaiagents", "autogen", "ag2"):
+            if is_available(candidate):
+                framework = "praisonai" if candidate == "praisonaiagents" else candidate
+                break
+        else:
+            raise RuntimeError(
+                "No supported framework installed. "
+                "Install one of: crewai, praisonaiagents, autogen, ag2."
+            )
+    else:
+        # Validate explicit framework like CLI does
+        from .framework_adapters.validators import assert_framework_available
+        assert_framework_available(framework)
+
+    gen = AgentsGenerator(
+        agent_file=agent_file,
+        framework=framework,
+        config_list=_build_config_list(),
+        tools=tools,
+        agent_yaml=agent_yaml,
+        cli_config=cli_config,
+    )
+    return gen.generate_crew_and_kickoff()
+
+
+async def arun(agent_file: str,
+               framework: str | None = None,
+               *,
+               tools: list | None = None,
+               agent_yaml: str | None = None,
+               cli_config: dict | None = None) -> str:
+    """Async equivalent of `run()` for FastAPI / Jupyter async callers."""
+    import asyncio
+    return await asyncio.to_thread(
+        run, agent_file, framework,
+        tools=tools, agent_yaml=agent_yaml, cli_config=cli_config,
+    )

src/praisonai/praisonai/_framework_availability.py

@@ -30,6 +30,11 @@ _PROBES: dict[str, Callable[[], bool]] = {
     "agentops":          lambda: importlib.util.find_spec("agentops") is not None,
     "litellm":           lambda: importlib.util.find_spec("litellm") is not None,
     "openai":            lambda: importlib.util.find_spec("openai") is not None,
+    # Additional probes for Gap 2 consolidation
+    "acp":               lambda: importlib.util.find_spec("acp") is not None,
+    "rich":              lambda: importlib.util.find_spec("rich") is not None,
+    "gradio":            lambda: importlib.util.find_spec("gradio") is not None,
+    "unsloth":           lambda: importlib.util.find_spec("unsloth") is not None,
 }
 
 def is_available(name: str) -> bool:

src/praisonai/praisonai/acp/__init__.py

@@ -42,7 +42,8 @@ def __getattr__(name: str):
 
 def _check_acp_available() -> bool:
     """Check if agent-client-protocol package is installed."""
-    return importlib.util.find_spec("acp") is not None
+    from .._framework_availability import is_available
+    return is_available("acp")
 
 
 def _get_install_instructions() -> str:

src/praisonai/praisonai/claw/default_app.py

@@ -8,29 +8,9 @@ Custom page: Feature Explorer
 """
 
 import os
+
 import praisonaiui as aiui
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
-
-# ── Set up datastore bridge ─────────────────────────────────
-aiui.set_datastore(PraisonAISessionDataStore())
-
-# ── Dashboard style ─────────────────────────────────────────
-aiui.set_style("dashboard")
-aiui.set_pages([
-    "chat",        # Core AI chat
-    "channels",    # Messaging platforms
-    "agents",      # Agent management
-    "skills",      # Available tools
-    "memory",      # Agent memory
-    "knowledge",   # Knowledge base
-    "cron",        # Scheduled jobs
-    "guardrails",  # Safety rules
-    "sessions",    # Conversation history
-    "usage",       # Token usage & costs
-    "config",      # Runtime configuration
-    "logs",        # Server logs
-    "debug",       # System debug info
-])
+from praisonai.integration.host_app import configure_host, create_host_app
 
 # ── Agent definitions ───────────────────────────────────────
 AGENTS = [
@@ -71,6 +51,34 @@ AGENTS = [
 ]
 
 
+def _build_praison_agents():
+    from praisonaiagents import Agent
+
+    return [
+        Agent(name=a["name"], instructions=a["instructions"], llm=a["model"])
+        for a in AGENTS
+    ]
+
+
+configure_host(
+    pages=[
+        "chat",
+        "channels",
+        "agents",
+        "skills",
+        "memory",
+        "knowledge",
+        "cron",
+        "guardrails",
+        "sessions",
+        "usage",
+        "config",
+        "logs",
+        "debug",
+    ],
+    agents=_build_praison_agents(),
+)
+
 
 # ── Custom page: Feature Explorer ───────────────────────────
 
@@ -110,7 +118,10 @@ def _register_agents_in_dashboard():
         })
         print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
 
+
 try:
     _register_agents_in_dashboard()
 except Exception as e:
     print(f"   ⚠️  Agent registration failed (non-fatal): {e}")
+
+app = create_host_app()

src/praisonai/praisonai/cli/commands/dashboard.py

@@ -199,93 +199,24 @@ def _auto_start_services(console, host: str) -> bool:
 
 
 def _run_aiui_dashboard(port: int, host: str, console) -> bool:
-    """Run the aiui dashboard interface using proper CLI entrypoint."""
-    console.print("[bold green]🦞 Starting aiui Dashboard...[/bold green]")
-    
+    """Run the aiui dashboard in-process via ``build_host_app()`` (Pattern B)."""
+    console.print("[bold green]🦞 Starting aiui Dashboard (Pattern B host)...[/bold green]")
+
     try:
-        import tempfile
-        import os
-        
-        # Check if aiui is available first
-        result = subprocess.run([
-            sys.executable, "-c", "import praisonaiui"
-        ], capture_output=True, text=True)
-        
-        if result.returncode != 0:
-            console.print("[red]Error: aiui package not installed.[/red]")
-            console.print("[yellow]Install with: pip install aiui[/yellow]")
-            return False
-        
-        # Create a temporary script for aiui dashboard based on claw pattern
-        aiui_script = '''import os
-import praisonaiui as aiui
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
-
-# Set up datastore bridge
-aiui.set_datastore(PraisonAISessionDataStore())
-
-# Configure dashboard style
-aiui.set_style("dashboard")
-aiui.set_branding(title="PraisonAI Unified Dashboard", logo="🌟")
-
-# Set up pages for unified dashboard
-aiui.set_pages([
-    "chat", "agents", "memory", "knowledge", 
-    "skills", "sessions", "usage", "config", "logs"
-])
-
-# Register a simple welcome message
-@aiui.welcome
-async def on_welcome():
-    await aiui.say("🌟 Welcome to PraisonAI Unified Dashboard!")
-    await aiui.say("Access all your AI services from one interface.")
-
-# Register basic reply handler
-@aiui.reply
-async def on_reply(message: str, settings: dict | None = None):
-    await aiui.think("Processing...")
-    await aiui.say(f"Unified Dashboard received: {message}")
-    await aiui.say("Use the sidebar to access Flow, Agents, Memory, and more!")
-'''
-
-        # Create temp file safely
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
-            f.write(aiui_script)
-            temp_script = f.name
-        
-        try:
-            console.print(f"[green]✓ Starting aiui dashboard on {host}:{port}[/green]")
-            
-            # Use aiui CLI to run the dashboard (NOT create_app)
-            # Try aiui command first, fallback to python -m
-            try:
-                result = subprocess.run([
-                    "aiui", "run", temp_script, "--port", str(port), "--host", host
-                ], check=True)
-            except (subprocess.CalledProcessError, FileNotFoundError):
-                # Fallback to python -m praisonaiui
-                result = subprocess.run([
-                    sys.executable, "-m", "praisonaiui.cli", "run", 
-                    temp_script, "--port", str(port), "--host", host
-                ], check=True)
-            
-            return True
-            
-        except subprocess.CalledProcessError as e:
-            console.print(f"[red]aiui dashboard exited with code {e.returncode}[/red]")
-            return False
-        finally:
-            # Clean up temp file
-            try:
-                os.unlink(temp_script)
-            except (OSError, FileNotFoundError):
-                pass
-        
-    except subprocess.CalledProcessError as e:
-        console.print(f"[red]aiui dashboard failed with code {e.returncode}[/red]")
+        import uvicorn
+        from praisonai.integration.host_app import build_host_app
+    except ImportError:
+        console.print("[red]Error: aiui integration dependencies missing.[/red]")
+        console.print('[yellow]Install with: pip install "praisonai[ui]"[/yellow]')
         return False
+
+    try:
+        app = build_host_app(pages=["chat", "sessions", "workflows", "hooks", "usage"])
+        console.print(f"[green]✓ Starting in-process host on {host}:{port}[/green]")
+        uvicorn.run(app, host=host, port=port, log_level="info")
+        return True
     except Exception as e:
-        console.print(f"[red]Error running aiui dashboard: {e}[/red]")
+        console.print(f"[red]Error running in-process host: {e}[/red]")
         return False
 
 
@@ -612,6 +543,10 @@ def unified(
     """
     Launch the PraisonAI Unified Dashboard.
     
+    Pattern B (default with --aiui): in-process ``build_host_app()`` host.
+    Pattern C: ``praisonai serve gateway`` or ``run_integrated_gateway()``.
+    Legacy: set ``PRAISONAI_HOST_LEGACY=1`` for callback-only ``@aiui.reply`` mode.
+    
     Provides a single interface at localhost:3000 to access:
     - Flow Visual Builder (Langflow) - port 7860
     - Claw Dashboard (Full UI) - port 8082  

src/praisonai/praisonai/cli/commands/ui.py

@@ -13,7 +13,12 @@ from typing import Optional
 
 import typer
 
-app = typer.Typer(help="🤖 PraisonAI Clean Chat UI")
+app = typer.Typer(
+    help=(
+        "🤖 PraisonAI Clean Chat UI (Pattern B host via bundled default_app). "
+        "Pattern C: AIUIGateway. Legacy: PRAISONAI_HOST_LEGACY=1."
+    )
+)
 
 UI_DIR = Path.home() / ".praisonai" / "ui"
 DEFAULT_APP = UI_DIR / "app.py"

src/praisonai/praisonai/cli/features/serve.py

@@ -127,6 +127,11 @@ Launch PraisonAI servers with unified discovery support.
 
 [bold]Discovery:[/bold]
   All servers expose /__praisonai__/discovery for unified endpoint discovery.
+
+[bold]A2UI / AG-UI rendering:[/bold]
+  For live A2UI surfaces in a web UI, use PraisonAIUI (POST /agui SSE + /api/surfaces):
+    pip install praisonaiui praisonaiagents[a2ui]
+  See docs: Integrate A2UI with Your Frontend.
   Use `praisonai endpoints` to interact with any server.
 """
         self._print_rich(help_text)

src/praisonai/praisonai/cli/main.py

@@ -166,17 +166,19 @@ AUTOGEN_AVAILABLE = False
 PRAISONAI_AVAILABLE = False
 TRAIN_AVAILABLE = False
 
-# Use find_spec for fast availability checks (no actual import)
-import importlib.util
-GRADIO_AVAILABLE = importlib.util.find_spec("gradio") is not None
+# Use centralized availability detection
+from .._framework_availability import is_available
+
+GRADIO_AVAILABLE = is_available("gradio")
 try:
+    import importlib.util
     CALL_MODULE_AVAILABLE = importlib.util.find_spec("praisonai.api.call") is not None
 except (ModuleNotFoundError, AttributeError):
     CALL_MODULE_AVAILABLE = False
-CREWAI_AVAILABLE = importlib.util.find_spec("crewai") is not None
-AUTOGEN_AVAILABLE = importlib.util.find_spec("autogen") is not None
-PRAISONAI_AVAILABLE = importlib.util.find_spec("praisonaiagents") is not None
-TRAIN_AVAILABLE = importlib.util.find_spec("unsloth") is not None
+CREWAI_AVAILABLE = is_available("crewai")
+AUTOGEN_AVAILABLE = is_available("autogen")
+PRAISONAI_AVAILABLE = is_available("praisonaiagents")
+TRAIN_AVAILABLE = is_available("unsloth")
 
 # Lazy import helpers for optional dependencies (defined after availability flags)
 def _get_call_module():

src/praisonai/praisonai/cli/output/console.py

@@ -28,8 +28,8 @@ def _get_rich_available() -> bool:
     global _rich_available
     if _rich_available is None:
         try:
-            import importlib.util
-            _rich_available = importlib.util.find_spec("rich") is not None
+            from ..._framework_availability import is_available
+            _rich_available = is_available("rich")
         except ImportError:
             _rich_available = False
     return _rich_available

src/praisonai/praisonai/deploy.py

@@ -57,7 +57,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==4.6.43 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==4.6.45 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/inbuilt_tools/__init__.py

@@ -1,11 +1,10 @@
 """Lazy access to autogen_tools, with `find_spec` startup probes for crewai/autogen."""
-import importlib.util
 from praisonai.auto import _load_optional
 
 from .._framework_availability import is_available
 CREWAI_AVAILABLE = is_available("crewai")
 AUTOGEN_AVAILABLE = is_available("autogen")
-PRAISONAI_TOOLS_PACKAGE_AVAILABLE = importlib.util.find_spec("praisonai_tools") is not None
+PRAISONAI_TOOLS_PACKAGE_AVAILABLE = is_available("praisonai_tools")
 
 
 def _load_autogen_tools():

src/praisonai/praisonai/integration/__init__.py

@@ -0,0 +1,19 @@
+"""PraisonAI ↔ PraisonAIUI integration layer (Pattern B/C host bootstrap)."""
+
+from .host_app import (
+    build_host_app,
+    configure_host,
+    create_host_app,
+    is_legacy_host,
+    setup_bridges,
+)
+from .gateway_host import run_integrated_gateway
+
+__all__ = [
+    "build_host_app",
+    "configure_host",
+    "create_host_app",
+    "is_legacy_host",
+    "setup_bridges",
+    "run_integrated_gateway",
+]

src/praisonai/praisonai/integration/_legacy_handlers.py

@@ -0,0 +1,18 @@
+"""Shared legacy @aiui.reply handlers for bundled default apps."""
+
+from __future__ import annotations
+
+
+def register_legacy_reply(aiui, *, agent_name: str = "PraisonAI"):
+    """Register callback-only reply handler when ``PRAISONAI_HOST_LEGACY=1``."""
+
+    @aiui.reply
+    async def legacy_reply(message: str, session_id: str = "default"):
+        from praisonaiagents import Agent
+
+        agent = Agent(
+            name=agent_name,
+            instructions="You are a helpful assistant.",
+            llm="gpt-4o-mini",
+        )
+        return agent.run(message)

src/praisonai/praisonai/integration/bridges/__init__.py

@@ -0,0 +1 @@
+"""L1→L2 bridge helpers for PraisonAIUI feature routes."""

src/praisonai/praisonai/integration/bridges/approvals_bridge.py

@@ -0,0 +1,31 @@
+"""ApprovalRegistry CRUD adapter for aiui backends."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+
+def list_pending_approvals() -> List[Dict[str, Any]]:
+    try:
+        from praisonaiagents.approval import get_approval_registry
+
+        registry = get_approval_registry()
+        backend = getattr(registry, "_backend", None)
+        if backend and hasattr(backend, "list_pending"):
+            return backend.list_pending()
+    except ImportError:
+        pass
+    return []
+
+
+def get_approval_policies() -> Dict[str, Any]:
+    try:
+        from praisonaiagents.approval import get_approval_registry
+
+        registry = get_approval_registry()
+        return {
+            "auto_approve_env": registry.is_env_auto_approve(),
+            "requirements": getattr(registry, "_requirements", {}),
+        }
+    except ImportError:
+        return {}

src/praisonai/praisonai/integration/bridges/hooks_query.py

@@ -0,0 +1,15 @@
+"""Expose praisonaiagents HookRegistry for admin/API consumers."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+
+def list_hooks_for_api() -> List[Dict[str, Any]]:
+    """Flat hook list suitable for REST responses."""
+    try:
+        from praisonaiagents.hooks.registry import list_hooks_for_api as core_list
+
+        return core_list()
+    except ImportError:
+        return []

src/praisonai/praisonai/integration/bridges/schedules_runner.py

@@ -0,0 +1,30 @@
+"""Ensure SDK schedule runner + optional ScheduleLoop daemon."""
+
+from __future__ import annotations
+
+import logging
+
+log = logging.getLogger(__name__)
+_runner_started = False
+_loop = None
+
+
+def ensure_schedule_runner() -> None:
+    """Lazy-init FileScheduleStore-backed runner and optional ScheduleLoop."""
+    global _runner_started, _loop
+    if _runner_started:
+        return
+    try:
+        from praisonaiagents.scheduler import FileScheduleStore, ScheduleLoop
+
+        store = FileScheduleStore()
+
+        def on_trigger(job):
+            log.info("Schedule triggered: %s", getattr(job, "name", job))
+
+        _loop = ScheduleLoop(on_trigger=on_trigger, store=store)
+        _loop.start()
+        _runner_started = True
+        log.info("ScheduleLoop started for host integration")
+    except Exception as exc:
+        log.debug("Schedule runner unavailable: %s", exc)

src/praisonai/praisonai/integration/bridges/usage_bridge.py

@@ -0,0 +1,41 @@
+"""Wire core token usage collector to optional persistence sink."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+def register_usage_sink() -> Optional[Any]:
+    """Attach global token collector to in-memory sink when available."""
+    try:
+        from praisonaiagents.telemetry.protocols import (
+            InMemoryTokenUsageSink,
+            InMemoryUsageQuery,
+            TokenCollectorUsageQuery,
+        )
+        from praisonaiagents.telemetry.token_collector import get_token_collector
+
+        sink = InMemoryTokenUsageSink()
+        collector = get_token_collector()
+        collector.set_sink(sink)
+        return sink
+    except ImportError:
+        return None
+
+
+def get_usage_query():
+    """Return UsageQueryProtocol adapter for UI consumption."""
+    try:
+        from praisonaiagents.telemetry.protocols import (
+            InMemoryUsageQuery,
+            TokenCollectorUsageQuery,
+        )
+        from praisonaiagents.telemetry.token_collector import get_token_collector
+
+        collector = get_token_collector()
+        sink = getattr(collector, "_sink", None)
+        if sink is not None and hasattr(sink, "records"):
+            return InMemoryUsageQuery(sink)
+        return TokenCollectorUsageQuery(collector)
+    except ImportError:
+        return None

src/praisonai/praisonai/integration/bridges/workflows_service.py

@@ -0,0 +1,77 @@
+"""Run praisonaiagents AgentFlow workflows with run tracking."""
+
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import yaml
+
+
+def load_workflow_yaml(path: str) -> Dict[str, Any]:
+    """Load workflow definition from a YAML file."""
+    with open(path, encoding="utf-8") as f:
+        data = yaml.safe_load(f) or {}
+    if not isinstance(data, dict):
+        raise ValueError("Workflow YAML must be a mapping")
+    return data
+
+
+def run_workflow(
+    workflow_id: str,
+    *,
+    input_text: str = "",
+    workflow_config: Optional[Dict[str, Any]] = None,
+    yaml_path: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Execute a workflow and return a serialisable run record."""
+    run_id = uuid.uuid4().hex[:12]
+    config = dict(workflow_config or {})
+    if yaml_path:
+        loaded = load_workflow_yaml(yaml_path)
+        config = {**loaded, **config}
+
+    steps = config.get("steps") or []
+    if not steps:
+        return {
+            "id": run_id,
+            "workflow_id": workflow_id,
+            "status": "failed",
+            "input": {"text": input_text},
+            "error": "Workflow has no steps",
+        }
+
+    try:
+        from praisonaiagents import Agent, AgentFlow
+
+        flow_steps = []
+        for i, step in enumerate(steps):
+            if isinstance(step, str):
+                flow_steps.append(
+                    Agent(
+                        name=f"step_{i + 1}",
+                        instructions=step,
+                        llm=config.get("model", "gpt-4o-mini"),
+                    )
+                )
+            else:
+                flow_steps.append(step)
+
+        flow = AgentFlow(steps=flow_steps, name=config.get("name", "Workflow"))
+        result = flow.run(input_text)
+        return {
+            "id": run_id,
+            "workflow_id": workflow_id,
+            "status": "completed",
+            "input": {"text": input_text},
+            "output": result if isinstance(result, dict) else {"result": str(result)},
+        }
+    except Exception as exc:
+        return {
+            "id": run_id,
+            "workflow_id": workflow_id,
+            "status": "failed",
+            "input": {"text": input_text},
+            "error": str(exc),
+        }

src/praisonai/praisonai/integration/context_files.py

@@ -0,0 +1,22 @@
+"""AGENTS.md-style context file injection for host apps."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List, Optional
+
+
+def load_context_files(
+    paths: Optional[List[str]] = None,
+    *,
+    cwd: Optional[Path] = None,
+) -> str:
+    """Load context from AGENTS.md-style files and return combined text."""
+    base = cwd or Path.cwd()
+    candidates = paths or ["AGENTS.md", "agents.md", ".agents/AGENTS.md"]
+    chunks: List[str] = []
+    for name in candidates:
+        path = base / name
+        if path.is_file():
+            chunks.append(path.read_text(encoding="utf-8"))
+    return "\n\n".join(chunks)

src/praisonai/praisonai/integration/gateway_host.py

@@ -0,0 +1,27 @@
+"""Pattern C — integrated AIUIGateway launcher."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+async def run_integrated_gateway(
+    *,
+    port: int = 8080,
+    host: str = "127.0.0.1",
+    static_dir: Optional[str] = None,
+    ui_config: Optional[dict[str, Any]] = None,
+    **configure_kwargs: Any,
+) -> None:
+    """Start ``AIUIGateway`` after ``configure_host()`` (Pattern B/C parity)."""
+    from praisonai.integration.host_app import configure_host
+    from praisonaiui.integration import AIUIGateway
+
+    configure_host(**configure_kwargs)
+    gateway = AIUIGateway(
+        host=host,
+        port=port,
+        static_dir=static_dir,
+        ui_config=ui_config or {},
+    )
+    await gateway.start()

src/praisonai/praisonai/integration/host_app.py

@@ -0,0 +1,163 @@
+"""Official PraisonAI → PraisonAIUI host bootstrap (Pattern B).
+
+Wires ``PraisonAISessionDataStore`` + ``PraisonAIProvider`` before ``create_app()``.
+Set ``PRAISONAI_HOST_LEGACY=1`` to skip provider wiring (callback-only mode).
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, List, Optional, Sequence
+
+_CONFIGURED = False
+
+
+def is_legacy_host() -> bool:
+    """True when callback-only ``@aiui.reply`` mode is requested."""
+    return os.environ.get("PRAISONAI_HOST_LEGACY", "").strip().lower() in (
+        "1",
+        "true",
+        "yes",
+    )
+
+
+def configure_host(
+    *,
+    pages: Optional[Sequence[str]] = None,
+    title: str = "PraisonAI",
+    logo: str = "🤖",
+    sidebar: bool = True,
+    page_header: bool = True,
+    theme: Optional[Dict[str, Any]] = None,
+    agents: Optional[List[Any]] = None,
+    agent_kwargs: Optional[Dict[str, Any]] = None,
+    gateway: Any = None,
+    modules: Optional[Sequence[str]] = None,
+) -> None:
+    """Apply PraisonAIUI host settings and wire L1 backends (unless legacy mode)."""
+    global _CONFIGURED
+
+    import praisonaiui as aiui
+    from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
+
+    aiui.set_datastore(PraisonAISessionDataStore())
+    aiui.set_style("dashboard")
+    aiui.set_branding(title=title, logo=logo)
+
+    if pages is not None:
+        aiui.set_pages(list(pages))
+
+    dashboard_opts: Dict[str, Any] = {"sidebar": sidebar, "page_header": page_header}
+    if modules:
+        dashboard_opts["modules"] = list(modules)
+    aiui.set_dashboard(**dashboard_opts)
+
+    if theme:
+        aiui.set_theme(**theme)
+
+    try:
+        from praisonai.ui._external_agents import aiui_settings_entries
+
+        external = aiui_settings_entries()
+        if external and hasattr(aiui, "set_settings"):
+            aiui.set_settings(external)
+    except ImportError:
+        pass
+
+    if gateway is not None:
+        try:
+            from praisonaiui.features._gateway_ref import set_gateway
+
+            set_gateway(gateway)
+        except ImportError:
+            pass
+
+    if not is_legacy_host():
+        from praisonaiui.providers import PraisonAIProvider
+        from praisonaiui.server import set_provider
+
+        kwargs = dict(agent_kwargs or {})
+        if agents:
+            set_provider(PraisonAIProvider(agents=list(agents), **kwargs))
+        else:
+            set_provider(
+                PraisonAIProvider(
+                    name=kwargs.pop("name", "PraisonAI"),
+                    instructions=kwargs.pop(
+                        "instructions",
+                        "You are a helpful assistant.",
+                    ),
+                    llm=kwargs.pop(
+                        "llm", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")
+                    ),
+                    **kwargs,
+                )
+            )
+        setup_bridges()
+
+    _CONFIGURED = True
+
+
+def setup_bridges() -> None:
+    """Register optional L1→L2 bridge hooks (usage sink, schedules, aiui backends)."""
+    import logging
+
+    log = logging.getLogger(__name__)
+    sink = None
+
+    try:
+        from praisonai.integration.bridges.usage_bridge import register_usage_sink
+
+        sink = register_usage_sink()
+    except Exception as exc:
+        log.debug("usage bridge unavailable: %s", exc)
+
+    try:
+        from praisonai.integration.bridges.schedules_runner import ensure_schedule_runner
+
+        ensure_schedule_runner()
+    except Exception as exc:
+        log.debug("schedule runner unavailable: %s", exc)
+
+    try:
+        import praisonaiui.backends as backends
+        from praisonai.integration.bridges.hooks_query import list_hooks_for_api
+        from praisonai.integration.bridges.workflows_service import run_workflow as svc_run
+
+        def _workflow_backend(wf_id, *, workflow, input_data):
+            text = (input_data or {}).get("text") or (input_data or {}).get("message") or ""
+            return svc_run(wf_id, input_text=text, workflow_config=workflow)
+
+        backends.set_backend("hooks", list_hooks_for_api)
+        backends.set_backend("workflows", _workflow_backend)
+        if sink is not None:
+            backends.set_backend("usage_sink", sink)
+        from praisonai.integration.bridges.usage_bridge import get_usage_query
+
+        query = get_usage_query()
+        if query is not None:
+            backends.set_backend("usage_query", query)
+        from praisonai.integration.bridges.approvals_bridge import (
+            get_approval_policies,
+            list_pending_approvals,
+        )
+
+        backends.set_backend("approvals_pending", list_pending_approvals)
+        backends.set_backend("approvals_policies", get_approval_policies)
+    except Exception as exc:
+        log.debug("aiui backend injection failed: %s", exc)
+
+
+def create_host_app():
+    """Return the Starlette app from PraisonAIUI (call after ``configure_host``)."""
+    from praisonaiui.server import create_app
+
+    if not _CONFIGURED:
+        configure_host()
+    return create_app()
+
+
+def build_host_app(**configure_kwargs):
+    """One-shot: configure host + return ``create_app()``."""
+    configure_host(**configure_kwargs)
+    return create_host_app()

src/praisonai/praisonai/integration/pages/bot_health.py

@@ -0,0 +1,21 @@
+"""Optional L3 page — bot / channel health."""
+
+from __future__ import annotations
+
+import praisonaiui as aiui
+
+
+@aiui.page("bot-health", title="Bot Health", icon="🤖")
+async def bot_health_page():
+    """Dashboard page for channel and gateway status."""
+    status = {"gateway": "unknown", "channels": []}
+    try:
+        from praisonaiui.features._gateway_ref import get_gateway
+
+        gw = get_gateway()
+        if gw is not None:
+            status["gateway"] = "running"
+            status["agents"] = gw.list_agents() if hasattr(gw, "list_agents") else []
+    except ImportError:
+        pass
+    return status

src/praisonai/praisonai/integration/pages/workflow_runs.py

@@ -0,0 +1,16 @@
+"""Optional L3 page — workflow runs table."""
+
+from __future__ import annotations
+
+import praisonaiui as aiui
+
+
+@aiui.page("workflow-runs", title="Workflow Runs", icon="🔄")
+async def workflow_runs_page():
+    """Dashboard page listing recent workflow runs."""
+    try:
+        from praisonai.integration.bridges.workflows_service import run_workflow
+    except ImportError:
+        return {"runs": [], "note": "Workflow bridge unavailable"}
+
+    return {"runs": [], "service": "WorkflowRunService", "status": "ready"}

src/praisonai/praisonai/tool_resolver.py

@@ -163,7 +163,8 @@ class ToolResolver:
         """
         # Cache availability check
         if self._praisonai_tools_available is None:
-            self._praisonai_tools_available = importlib.util.find_spec("praisonai_tools") is not None
+            from ._framework_availability import is_available
+            self._praisonai_tools_available = is_available("praisonai_tools")
         
         if not self._praisonai_tools_available:
             return None
@@ -314,7 +315,8 @@ class ToolResolver:
         
         # 3. Add praisonai-tools (if installed)
         if self._praisonai_tools_available is None:
-            self._praisonai_tools_available = importlib.util.find_spec("praisonai_tools") is not None
+            from ._framework_availability import is_available
+            self._praisonai_tools_available = is_available("praisonai_tools")
         
         if self._praisonai_tools_available:
             try:

src/praisonai/praisonai/ui_agents/default_app.py

@@ -7,22 +7,22 @@ Requires: pip install "praisonai[ui]"
 """
 
 import os
+
 import praisonaiui as aiui
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
+from praisonai.integration.host_app import configure_host, create_host_app, is_legacy_host
 
-# ── Set up datastore bridge ─────────────────────────────────
-aiui.set_datastore(PraisonAISessionDataStore())
+configure_host(
+    title="PraisonAI Agents",
+    logo="🤖",
+    pages=["chat", "agents", "memory", "sessions"],
+    theme={"preset": "blue", "dark_mode": True, "radius": "lg"},
+    agent_kwargs={
+        "name": "AgentsRunner",
+        "instructions": "You are an agent runner that coordinates YAML-defined workflows.",
+        "llm": os.getenv("MODEL_NAME", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")),
+    },
+)
 
-# ── Dashboard style ─────────────────────────────────────────
-aiui.set_style("dashboard")
-aiui.set_branding(title="PraisonAI Agents", logo="🤖")
-aiui.set_theme(preset="blue", dark_mode=True, radius="lg")
-aiui.set_pages([
-    "chat",
-    "agents",
-    "memory",
-    "sessions",
-])
 
 # ── Load agents from YAML ───────────────────────────────────
 def _load_agents_from_yaml():
@@ -30,20 +30,25 @@ def _load_agents_from_yaml():
     agents_file = os.path.join(os.getcwd(), "agents.yaml")
     if not os.path.exists(agents_file):
         return []
-    
+
     try:
         import yaml
-        with open(agents_file, 'r') as f:
+
+        with open(agents_file, "r") as f:
             config = yaml.safe_load(f)
-        
+
         agents = []
-        if 'agents' in config:
-            for agent_def in config['agents']:
+        if "agents" in config:
+            for agent_def in config["agents"]:
                 agents.append({
-                    "name": agent_def.get('name', 'Agent'),
-                    "description": agent_def.get('description', 'YAML-defined agent'),
-                    "instructions": agent_def.get('instructions', agent_def.get('role', '')),
-                    "model": agent_def.get('model', os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")),
+                    "name": agent_def.get("name", "Agent"),
+                    "description": agent_def.get("description", "YAML-defined agent"),
+                    "instructions": agent_def.get(
+                        "instructions", agent_def.get("role", "")
+                    ),
+                    "model": agent_def.get(
+                        "model", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")
+                    ),
                     "icon": "🤖",
                 })
         return agents
@@ -51,13 +56,13 @@ def _load_agents_from_yaml():
         print(f"⚠️  Failed to load agents.yaml: {e}")
         return []
 
-# ── Register YAML agents ────────────────────────────────────
+
 yaml_agents = _load_agents_from_yaml()
 if yaml_agents:
     try:
         from praisonaiui.features.agents import get_agent_registry
+
         registry = get_agent_registry()
-        
         for agent_def in yaml_agents:
             registry.create(agent_def)
             print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
@@ -66,7 +71,6 @@ if yaml_agents:
 else:
     print("   ℹ️  No agents.yaml found, using default agents")
 
-# ── Default agents if no YAML ───────────────────────────────
 if not yaml_agents:
     DEFAULT_AGENTS = [
         {
@@ -77,17 +81,18 @@ if not yaml_agents:
             "icon": "🤖",
         },
     ]
-    
+
     try:
         from praisonaiui.features.agents import get_agent_registry
+
         registry = get_agent_registry()
-        
         for agent_def in DEFAULT_AGENTS:
             registry.create(agent_def)
             print(f"   ✓ Default Agent: {agent_def['icon']} {agent_def['name']}")
     except Exception as e:
         print(f"   ⚠️  Default agent registration failed: {e}")
 
+
 @aiui.starters
 async def get_starters():
     """Suggest conversation starters for agents."""
@@ -97,43 +102,48 @@ async def get_starters():
         {"label": "Agent Status", "message": "Show status of all agents", "icon": "📊"},
     ]
 
+
 @aiui.welcome
 async def on_welcome():
     """Welcome message."""
-    await aiui.say("👋 Welcome to PraisonAI Agents! This interface runs YAML-defined multi-agent workflows.")
+    await aiui.say(
+        "👋 Welcome to PraisonAI Agents! This interface runs YAML-defined multi-agent workflows."
+    )
 
-# Session-scoped agent cache
-_agents_cache = {}
 
-@aiui.reply
-async def on_message(message: str):
-    """Handle messages with agent execution."""
-    await aiui.think("Processing...")
-    
-    try:
-        from praisonaiagents import Agent
-        
-        # Create or get cached agent
-        session_id = getattr(aiui.current_session, 'id', 'default')
-        if session_id not in _agents_cache:
-            _agents_cache[session_id] = Agent(
-                name="AgentsRunner",
-                instructions="You are an agent runner that coordinates YAML-defined workflows.",
-                llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
-            )
-        
-        agent = _agents_cache[session_id]
-        result = await agent.achat(str(message))
-        
-        # Stream the response
-        response_text = str(result) if result else "No response"
-        words = response_text.split(" ")
-        for i, word in enumerate(words):
-            await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
-            
-    except Exception as e:
-        await aiui.say(f"❌ Error: {e}")
+if is_legacy_host():
+    _agents_cache = {}
 
-@aiui.cancel
-async def on_cancel():
-    await aiui.say("⏹️ Stopped.")
+    @aiui.reply
+    async def on_message(message: str):
+        """Handle messages with agent execution."""
+        await aiui.think("Processing...")
+
+        try:
+            from praisonaiagents import Agent
+
+            session_id = getattr(aiui.current_session, "id", "default")
+            if session_id not in _agents_cache:
+                _agents_cache[session_id] = Agent(
+                    name="AgentsRunner",
+                    instructions="You are an agent runner that coordinates YAML-defined workflows.",
+                    llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
+                )
+
+            agent = _agents_cache[session_id]
+            result = await agent.achat(str(message))
+
+            response_text = str(result) if result else "No response"
+            words = response_text.split(" ")
+            for i, word in enumerate(words):
+                await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
+
+        except Exception as e:
+            await aiui.say(f"❌ Error: {e}")
+
+    @aiui.cancel
+    async def on_cancel():
+        await aiui.say("⏹️ Stopped.")
+
+
+app = create_host_app()

src/praisonai/praisonai/ui_bot/default_app.py

@@ -7,20 +7,24 @@ Requires: pip install "praisonai[ui]"
 """
 
 import os
+
 import praisonaiui as aiui
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
+from praisonai.integration.host_app import configure_host, create_host_app, is_legacy_host
 
-# ── Set up datastore bridge ─────────────────────────────────
-aiui.set_datastore(PraisonAISessionDataStore())
+configure_host(
+    title="PraisonAI Bot",
+    logo="🤖",
+    pages=["chat", "sessions"],
+    theme={"preset": "green", "dark_mode": True, "radius": "lg"},
+    agent_kwargs={
+        "name": "PraisonBot",
+        "instructions": (
+            "You are a helpful bot assistant. Break down your reasoning into clear steps."
+        ),
+        "llm": os.getenv("MODEL_NAME", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")),
+    },
+)
 
-# ── Dashboard style ─────────────────────────────────────────
-aiui.set_style("dashboard")
-aiui.set_branding(title="PraisonAI Bot", logo="🤖")
-aiui.set_theme(preset="green", dark_mode=True, radius="lg")
-aiui.set_pages([
-    "chat",
-    "sessions",
-])
 
 @aiui.starters
 async def get_starters():
@@ -32,58 +36,59 @@ async def get_starters():
         {"label": "About", "message": "Tell me about yourself", "icon": "ℹ️"},
     ]
 
+
 @aiui.welcome
 async def on_welcome():
     """Welcome message for bot interface."""
-    await aiui.say("🤖 Hi! I'm your PraisonAI Bot. I can help with various tasks and show you step-by-step reasoning.")
+    await aiui.say(
+        "🤖 Hi! I'm your PraisonAI Bot. I can help with various tasks "
+        "and show you step-by-step reasoning."
+    )
 
-# Session-scoped bot cache
-_bots_cache = {}
 
-@aiui.reply
-async def on_message(message: str):
-    """Handle bot interactions with step visualization."""
-    session_id = getattr(aiui.current_session, 'id', 'default')
-    
-    # Show initial thinking step
-    await aiui.think("🤔 Analyzing request...")
-    
-    try:
-        from praisonaiagents import Agent
-        
-        # Create or get cached bot agent
-        if session_id not in _bots_cache:
-            _bots_cache[session_id] = Agent(
-                name="PraisonBot",
-                instructions="You are a helpful bot assistant. Break down your reasoning into clear steps.",
-                llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
-            )
-        
-        bot = _bots_cache[session_id]
-        
-        # Execute with step visualization
-        await aiui.think("⚙️ Processing...")
-        
-        result = await bot.achat(str(message))
-        
-        # Stream the response with step markers
-        response_text = str(result) if result else "No response from bot"
-        
-        # Add step visualization
-        await aiui.say("**🔄 Step 1: Analysis Complete**")
-        await aiui.think("📤 Generating response...")
-        
-        # Stream the main response
-        words = response_text.split(" ")
-        for i, word in enumerate(words):
-            await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
-        
-        await aiui.say("\n\n**✅ Step 2: Response Generated**")
-            
-    except Exception as e:
-        await aiui.say(f"❌ Bot Error: {e}")
-        await aiui.say("Please check your configuration and try again.")
+if is_legacy_host():
+    _bots_cache = {}
 
-@aiui.cancel
-async def on_cancel():
-    await aiui.say("⏹️ Bot interaction cancelled.")
+    @aiui.reply
+    async def on_message(message: str):
+        """Handle bot interactions with step visualization."""
+        session_id = getattr(aiui.current_session, "id", "default")
+        await aiui.think("🤔 Analyzing request...")
+
+        try:
+            from praisonaiagents import Agent
+
+            if session_id not in _bots_cache:
+                _bots_cache[session_id] = Agent(
+                    name="PraisonBot",
+                    instructions=(
+                        "You are a helpful bot assistant. "
+                        "Break down your reasoning into clear steps."
+                    ),
+                    llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
+                )
+
+            bot = _bots_cache[session_id]
+            await aiui.think("⚙️ Processing...")
+            result = await bot.achat(str(message))
+            response_text = str(result) if result else "No response from bot"
+
+            await aiui.say("**🔄 Step 1: Analysis Complete**")
+            await aiui.think("📤 Generating response...")
+
+            words = response_text.split(" ")
+            for i, word in enumerate(words):
+                await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
+
+            await aiui.say("\n\n**✅ Step 2: Response Generated**")
+
+        except Exception as e:
+            await aiui.say(f"❌ Bot Error: {e}")
+            await aiui.say("Please check your configuration and try again.")
+
+    @aiui.cancel
+    async def on_cancel():
+        await aiui.say("⏹️ Bot interaction cancelled.")
+
+
+app = create_host_app()

src/praisonai/praisonai/ui_chat/default_app.py

@@ -13,27 +13,24 @@ Run:
 import os
 
 import praisonaiui as aiui
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
+from praisonai.integration.host_app import configure_host, create_host_app, is_legacy_host
 
-# ── Set up datastore bridge ─────────────────────────────────
-aiui.set_datastore(PraisonAISessionDataStore())
-
-# ── Dashboard style, but no sidebar navigation ─────────────
-aiui.set_style("dashboard")
-aiui.set_dashboard(sidebar=False, page_header=False)
-aiui.set_branding(title="PraisonAI Chat", logo="🤖")
-aiui.set_theme(preset="blue", dark_mode=True, radius="lg")
-aiui.set_pages(["chat"])
-
-# Add external agent settings
-try:
-    from praisonai.ui._external_agents import aiui_settings_entries
-    external_settings = aiui_settings_entries()
-    if external_settings:
-        aiui.set_settings(external_settings)
-except ImportError:
-    # External agents not available
-    pass
+configure_host(
+    title="PraisonAI Chat",
+    logo="🤖",
+    pages=["chat"],
+    sidebar=False,
+    page_header=False,
+    theme={"preset": "blue", "dark_mode": True, "radius": "lg"},
+    agent_kwargs={
+        "name": "PraisonAI",
+        "instructions": (
+            "You are a helpful assistant. Delegate coding/analysis tasks "
+            "to external subagents when available."
+        ),
+        "llm": os.getenv("MODEL_NAME", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")),
+    },
+)
 
 
 @aiui.starters
@@ -53,90 +50,93 @@ async def on_welcome():
     await aiui.say("👋 Hi! I'm your PraisonAI assistant. Ask me anything!")
 
 
-# Session-scoped agent cache to avoid cross-user state leaks
-_agents_cache = {}
+if is_legacy_host():
+    _agents_cache = {}
 
-@aiui.settings
-async def on_settings(new_settings):
-    # Clear cache for current session when settings change
-    session_id = getattr(aiui.current_session, 'id', 'default')
-    if session_id in _agents_cache:
-        del _agents_cache[session_id]
+    @aiui.settings
+    async def on_settings(new_settings):
+        session_id = getattr(aiui.current_session, "id", "default")
+        _agents_cache.pop(session_id, None)
+
+    def _get_agent(settings: dict | None = None):
+        session_id = getattr(aiui.current_session, "id", "default")
+        settings_key = str(sorted((settings or {}).items()))
+        cache_key = f"{session_id}:{settings_key}"
+
+        if cache_key not in _agents_cache:
+            try:
+                from praisonaiagents import Agent
+                from praisonai.ui._external_agents import external_agent_tools
+
+                tools = external_agent_tools(
+                    settings or {},
+                    workspace=os.environ.get("PRAISONAI_WORKSPACE", "."),
+                )
+                _agents_cache[cache_key] = Agent(
+                    name="PraisonAI",
+                    instructions=(
+                        "You are a helpful assistant. Delegate coding/analysis tasks "
+                        "to external subagents when available."
+                    ),
+                    llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
+                    tools=tools if tools else None,
+                )
+            except ImportError:
+                _agents_cache[cache_key] = None
+        return _agents_cache[cache_key]
+
+    @aiui.reply
+    async def on_message(message: str, settings: dict | None = None):
+        """Stream a response using PraisonAI Agent or fallback to OpenAI."""
+        await aiui.think("Thinking...")
+
+        agent = _get_agent(settings)
+        if agent is not None:
+            try:
+                result = await agent.achat(str(message))
+                response_text = str(result) if result else ""
+                words = response_text.split(" ")
+                for i, word in enumerate(words):
+                    await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
+                return
+            except Exception as e:
+                import logging
+
+                logging.getLogger(__name__).exception("Agent execution failed")
+                await aiui.say(f"⚠️ Agent error: {e}. Falling back to OpenAI...")
 
-def _get_agent(settings: dict | None = None):
-    session_id = getattr(aiui.current_session, 'id', 'default')
-    settings_key = str(sorted((settings or {}).items()))
-    cache_key = f"{session_id}:{settings_key}"
-    
-    if cache_key not in _agents_cache:
         try:
-            from praisonaiagents import Agent
-            from praisonai.ui._external_agents import external_agent_tools
-            
-            # Get external agent tools based on settings
-            tools = external_agent_tools(settings or {}, workspace=os.environ.get("PRAISONAI_WORKSPACE", "."))
-            
-            agent = Agent(
-                name="PraisonAI",
-                instructions="You are a helpful assistant. Delegate coding/analysis tasks to external subagents when available.",
-                llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
-                tools=tools if tools else None,
-            )
-            _agents_cache[cache_key] = agent
+            from openai import AsyncOpenAI
         except ImportError:
-            # Fallback to OpenAI if PraisonAI agents not available
-            _agents_cache[cache_key] = None
-    return _agents_cache[cache_key]
-
-@aiui.reply
-async def on_message(message: str, settings: dict | None = None):
-    """Stream a response using PraisonAI Agent or fallback to OpenAI."""
-    await aiui.think("Thinking...")
-    
-    # Try PraisonAI Agent first
-    agent = _get_agent(settings)
-    if agent is not None:
-        try:
-            # Use async call to avoid blocking the event loop
-            result = await agent.achat(str(message))
-            response_text = str(result) if result else ""
-            words = response_text.split(" ")
-            for i, word in enumerate(words):
-                await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
+            await aiui.say("❌ Please install openai: `pip install openai`")
             return
-        except Exception as e:
-            import logging
-            logging.getLogger(__name__).exception("Agent execution failed")
-            await aiui.say(f"⚠️ Agent error: {e}. Falling back to OpenAI...")
-    
-    # Fallback to direct OpenAI
-    try:
-        from openai import AsyncOpenAI
-    except ImportError:
-        await aiui.say("❌ Please install openai: `pip install openai`")
-        return
 
-    api_key = os.getenv("OPENAI_API_KEY")
-    if not api_key:
-        await aiui.say("❌ Please set OPENAI_API_KEY environment variable.")
-        return
+        api_key = os.getenv("OPENAI_API_KEY")
+        if not api_key:
+            await aiui.say("❌ Please set OPENAI_API_KEY environment variable.")
+            return
 
-    client = AsyncOpenAI(api_key=api_key)
-    stream = await client.chat.completions.create(
-        model=os.getenv("PRAISONAI_MODEL", "gpt-4o-mini"),
-        messages=[
-            {"role": "system", "content": "You are a helpful, concise assistant. Use markdown for formatting."},
-            {"role": "user", "content": str(message)},
-        ],
-        stream=True,
-    )
+        client = AsyncOpenAI(api_key=api_key)
+        stream = await client.chat.completions.create(
+            model=os.getenv("PRAISONAI_MODEL", "gpt-4o-mini"),
+            messages=[
+                {
+                    "role": "system",
+                    "content": "You are a helpful, concise assistant. Use markdown for formatting.",
+                },
+                {"role": "user", "content": str(message)},
+            ],
+            stream=True,
+        )
 
-    async for chunk in stream:
-        delta = chunk.choices[0].delta
-        if delta.content:
-            await aiui.stream_token(delta.content)
+        async for chunk in stream:
+            delta = chunk.choices[0].delta
+            if delta.content:
+                await aiui.stream_token(delta.content)
+
+    @aiui.cancel
+    async def on_cancel():
+        await aiui.say("⏹️ Stopped.")
 
 
-@aiui.cancel
-async def on_cancel():
-    await aiui.say("⏹️ Stopped.")
+app = create_host_app()

src/praisonai/praisonai/ui_dashboard/default_app.py

@@ -0,0 +1,47 @@
+"""PraisonAI Unified Dashboard — single host for all sidebar features.
+
+Loaded by ``praisonai unified`` (aiui mode) via ``build_host_app()``.
+"""
+
+import os
+
+import praisonaiui as aiui
+from praisonai.integration.host_app import configure_host, create_host_app, is_legacy_host
+
+configure_host(
+    title="PraisonAI Unified Dashboard",
+    logo="🌟",
+    pages=[
+        "chat",
+        "agents",
+        "memory",
+        "knowledge",
+        "skills",
+        "sessions",
+        "usage",
+        "config",
+        "logs",
+    ],
+    agent_kwargs={
+        "name": "PraisonAI",
+        "instructions": "You are a helpful assistant for the PraisonAI unified dashboard.",
+        "llm": os.getenv("PRAISONAI_MODEL", "gpt-4o-mini"),
+    },
+)
+
+
+@aiui.welcome
+async def on_welcome():
+    await aiui.say("🌟 Welcome to PraisonAI Unified Dashboard!")
+    await aiui.say("Use the sidebar to access chat, agents, memory, and more.")
+
+
+if is_legacy_host():
+
+    @aiui.reply
+    async def on_reply(message: str, settings: dict | None = None):
+        await aiui.think("Processing...")
+        await aiui.say(f"Unified Dashboard received: {message}")
+
+
+app = create_host_app()

src/praisonai/praisonai/ui_realtime/default_app.py

@@ -9,22 +9,28 @@ Requires: pip install "praisonai[ui]"
 """
 
 import os
-import praisonaiui as aiui
-from praisonaiui.features.realtime import OpenAIRealtimeManager
-from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
 
-# ── Set up datastore bridge and realtime manager ───────────
-aiui.set_datastore(PraisonAISessionDataStore())
+import praisonaiui as aiui
+from praisonai.integration.host_app import configure_host, create_host_app, is_legacy_host
+from praisonaiui.features.realtime import OpenAIRealtimeManager
+
+configure_host(
+    title="PraisonAI Realtime Voice",
+    logo="🎤",
+    pages=["chat", "sessions"],
+    theme={"preset": "red", "dark_mode": True, "radius": "lg"},
+    agent_kwargs={
+        "name": "RealtimeAssistant",
+        "instructions": (
+            "You are a voice-optimized assistant. "
+            "Keep responses conversational and concise for voice interaction."
+        ),
+        "llm": os.getenv("MODEL_NAME", os.getenv("PRAISONAI_MODEL", "gpt-4o-mini")),
+    },
+)
+
 aiui.set_realtime_manager(OpenAIRealtimeManager())
 
-# ── Dashboard style ─────────────────────────────────────────
-aiui.set_style("dashboard")
-aiui.set_branding(title="PraisonAI Realtime Voice", logo="🎤")
-aiui.set_theme(preset="red", dark_mode=True, radius="lg")
-aiui.set_pages([
-    "chat",
-    "sessions",
-])
 
 @aiui.starters
 async def get_starters():
@@ -35,6 +41,7 @@ async def get_starters():
         {"label": "Features", "message": "What realtime features are available?", "icon": "⚡"},
     ]
 
+
 @aiui.welcome
 async def on_welcome():
     """Welcome with realtime status."""
@@ -48,43 +55,43 @@ Welcome to PraisonAI's voice-powered realtime chat! Use the microphone button to
 - Dashboard with chat history and usage logs
 """)
 
-# Session-scoped realtime agent cache
-_realtime_cache = {}
 
-@aiui.reply
-async def on_message(message: str):
-    """Handle realtime interactions via voice or text."""
-    session_id = getattr(aiui.current_session, 'id', 'default')
-    
-    await aiui.think("🎤 Processing realtime request...")
-    
-    try:
-        from praisonaiagents import Agent
-        
-        # Create or get cached realtime agent
-        if session_id not in _realtime_cache:
-            _realtime_cache[session_id] = Agent(
-                name="RealtimeAssistant",
-                instructions="You are a voice-optimized assistant. Keep responses conversational and concise for voice interaction.",
-                llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
-            )
-        
-        agent = _realtime_cache[session_id]
-        
-        # Process the message with the agent
-        result = await agent.achat(str(message))
-        
-        # Stream response naturally (aiui handles voice output via WebRTC)
-        response_text = str(result) if result else "I'm sorry, I couldn't process that."
-        
-        # Stream tokens for smooth output
-        words = response_text.split(" ")
-        for i, word in enumerate(words):
-            await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
-            
-    except Exception as e:
-        await aiui.say(f"❌ Realtime Error: {e}")
+if is_legacy_host():
+    _realtime_cache = {}
 
-@aiui.cancel
-async def on_cancel():
-    await aiui.say("🔇 Realtime interaction stopped.")
+    @aiui.reply
+    async def on_message(message: str):
+        """Handle realtime interactions via voice or text."""
+        session_id = getattr(aiui.current_session, "id", "default")
+        await aiui.think("🎤 Processing realtime request...")
+
+        try:
+            from praisonaiagents import Agent
+
+            if session_id not in _realtime_cache:
+                _realtime_cache[session_id] = Agent(
+                    name="RealtimeAssistant",
+                    instructions=(
+                        "You are a voice-optimized assistant. "
+                        "Keep responses conversational and concise for voice interaction."
+                    ),
+                    llm=os.getenv("MODEL_NAME", "gpt-4o-mini"),
+                )
+
+            agent = _realtime_cache[session_id]
+            result = await agent.achat(str(message))
+            response_text = str(result) if result else "I'm sorry, I couldn't process that."
+
+            words = response_text.split(" ")
+            for i, word in enumerate(words):
+                await aiui.stream_token(word + (" " if i < len(words) - 1 else ""))
+
+        except Exception as e:
+            await aiui.say(f"❌ Realtime Error: {e}")
+
+    @aiui.cancel
+    async def on_cancel():
+        await aiui.say("🔇 Realtime interaction stopped.")
+
+
+app = create_host_app()

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "4.6.43"
+__version__ = "4.6.45"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=1.6.43",
+    "praisonaiagents>=1.6.45",
     "python-dotenv>=0.19.0",
     "litellm>=1.83.14,<2",
     "PyYAML>=6.0",

src/praisonai/tests/integration/test_aiui_backends_injection.py

@@ -0,0 +1,19 @@
+"""Cross-repo test — wrapper build_host_app + aiui backend injection."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip("praisonaiui")
+
+
+def test_build_host_injects_backends():
+    import praisonaiui.backends as backends
+    from praisonai.integration.host_app import build_host_app
+
+    backends.clear_backends()
+    app = build_host_app(pages=["chat"])
+    assert app is not None
+    registered = backends.list_backends()
+    assert "hooks" in registered
+    assert "workflows" in registered

src/praisonai/tests/integration/test_aiui_gateway_parity.py

@@ -0,0 +1,22 @@
+"""Pattern C gateway parity — datastore + backends after bootstrap."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip("praisonaiui")
+
+
+def test_gateway_start_wires_datastore(monkeypatch):
+    monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+    import praisonaiui.backends as backends
+    import praisonaiui.server as srv
+    from praisonai.integration import host_app
+
+    host_app._CONFIGURED = False
+    backends.clear_backends()
+    srv._provider = None
+
+    host_app.configure_host(pages=["chat"])
+    assert srv.get_datastore() is not None
+    assert "hooks" in backends.list_backends() or host_app._CONFIGURED

src/praisonai/tests/integration/test_aiui_host.py

@@ -0,0 +1,110 @@
+"""Integration tests for PraisonAI ↔ PraisonAIUI host bootstrap (Pattern B)."""
+
+from __future__ import annotations
+
+import importlib
+import os
+import time
+
+import pytest
+
+
+pytest.importorskip("praisonaiui")
+
+
+class TestHostAppBootstrap:
+    def test_is_legacy_host_env(self, monkeypatch):
+        from praisonai.integration.host_app import is_legacy_host
+
+        monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+        assert is_legacy_host() is False
+
+        monkeypatch.setenv("PRAISONAI_HOST_LEGACY", "1")
+        assert is_legacy_host() is True
+
+    def test_configure_host_sets_provider(self, monkeypatch):
+        monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+
+        import praisonaiui.server as srv
+        from praisonai.integration import host_app
+        from praisonaiui.providers import PraisonAIProvider
+
+        importlib.reload(host_app)
+        host_app._CONFIGURED = False
+        srv._provider = None
+
+        seen = []
+        monkeypatch.setattr(srv, "set_provider", lambda p: seen.append(p))
+
+        host_app.configure_host(
+            title="Test Host",
+            pages=["chat"],
+            agent_kwargs={"name": "test", "instructions": "test", "llm": "gpt-4o-mini"},
+        )
+
+        assert len(seen) == 1
+        assert isinstance(seen[0], PraisonAIProvider)
+        assert srv.get_datastore() is not None
+
+    def test_legacy_skips_provider(self, monkeypatch):
+        monkeypatch.setenv("PRAISONAI_HOST_LEGACY", "true")
+
+        import praisonaiui.server as srv
+        from praisonai.integration import host_app
+
+        importlib.reload(host_app)
+        host_app._CONFIGURED = False
+
+        seen = []
+
+        def _track(provider):
+            seen.append(provider)
+
+        monkeypatch.setattr(srv, "set_provider", _track)
+
+        host_app.configure_host(pages=["chat"])
+        assert seen == []
+
+    def test_build_host_app_returns_starlette(self, monkeypatch):
+        monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+
+        from praisonai.integration import host_app
+
+        importlib.reload(host_app)
+        host_app._CONFIGURED = False
+
+        app = host_app.build_host_app(pages=["chat"])
+        assert app is not None
+        assert hasattr(app, "routes")
+
+    def test_host_import_performance(self):
+        start = time.perf_counter()
+        import praisonai.integration.host_app  # noqa: F401
+
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        assert elapsed_ms < 2000, f"host_app import took {elapsed_ms:.0f}ms"
+
+    def test_hooks_query_bridge(self):
+        from praisonai.integration.bridges.hooks_query import list_hooks_for_api
+
+        hooks = list_hooks_for_api()
+        assert isinstance(hooks, list)
+
+
+class TestDefaultAppsLoad:
+    @pytest.mark.parametrize(
+        "module_path",
+        [
+            "praisonai.ui_chat.default_app",
+            "praisonai.ui_dashboard.default_app",
+            "praisonai.claw.default_app",
+            "praisonai.ui_bot.default_app",
+            "praisonai.ui_agents.default_app",
+            "praisonai.ui_realtime.default_app",
+        ],
+    )
+    def test_default_app_exports_app(self, module_path, monkeypatch):
+        monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+        mod = importlib.import_module(module_path)
+        assert hasattr(mod, "app")
+        assert mod.app is not None

src/praisonai/tests/integration/test_aiui_host_agentic.py

@@ -0,0 +1,31 @@
+"""Real agentic integration test — LLM via /run (AC-3). Skips without API key."""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+from starlette.testclient import TestClient
+
+pytest.importorskip("praisonaiui")
+
+pytestmark = pytest.mark.skipif(
+    not os.getenv("OPENAI_API_KEY"),
+    reason="OPENAI_API_KEY required for real agentic test",
+)
+
+
+def test_run_creates_session():
+    from praisonai.integration.host_app import build_host_app
+
+    client = TestClient(build_host_app(pages=["chat"]))
+    session_id = "agentic-ac3-test"
+    response = client.post(
+        "/run",
+        json={"message": "Reply with exactly: pong", "session_id": session_id},
+    )
+    assert response.status_code == 200
+
+    sessions = client.get("/sessions").json()
+    ids = [s.get("id") or s.get("session_id") for s in sessions.get("sessions", sessions)]
+    assert session_id in ids or any(session_id in str(i) for i in ids)

src/praisonai/tests/integration/test_aiui_host_isolation.py

@@ -0,0 +1,45 @@
+"""Multi-agent session isolation test (AC-9)."""
+
+from __future__ import annotations
+
+import pytest
+from starlette.testclient import TestClient
+
+pytest.importorskip("praisonaiui")
+
+
+class _SessionProvider:
+    name = "mock"
+
+    def __init__(self):
+        self._sessions = {}
+
+    async def run(self, message, *, session_id=None, agent_name=None, **kwargs):
+        sid = session_id or "default"
+        self._sessions.setdefault(sid, []).append(message)
+        reply = f"echo:{message}"
+        yield {"type": "text", "content": reply}
+        yield {"type": "done", "content": ""}
+
+
+@pytest.fixture
+def isolated_client(monkeypatch):
+    monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+    import praisonaiui.server as srv
+    from praisonai.integration import host_app
+
+    host_app._CONFIGURED = False
+    provider = _SessionProvider()
+    monkeypatch.setattr(srv, "set_provider", lambda p: setattr(srv, "_provider", p))
+    app = host_app.build_host_app(pages=["chat"])
+    srv.set_provider(provider)
+    return TestClient(app), provider
+
+
+def test_two_sessions_no_cross_leak(isolated_client):
+    client, provider = isolated_client
+    client.post("/run", json={"message": "secret-a", "session_id": "sess-a"})
+    client.post("/run", json={"message": "secret-b", "session_id": "sess-b"})
+    assert provider._sessions.get("sess-a") == ["secret-a"]
+    assert provider._sessions.get("sess-b") == ["secret-b"]
+    assert "secret-b" not in provider._sessions.get("sess-a", [])

src/praisonai/tests/integration/test_aiui_host_sse.py

@@ -0,0 +1,45 @@
+"""Integration tests — POST /run SSE RunEvent stream (AC-2)."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from starlette.testclient import TestClient
+
+pytest.importorskip("praisonaiui")
+
+
+class _MockProvider:
+    name = "mock"
+
+    async def run(self, message, *, session_id=None, agent_name=None, **kwargs):
+        yield {"type": "text", "content": "Hello SSE"}
+        yield {"type": "done", "content": ""}
+
+
+@pytest.fixture
+def host_client(monkeypatch):
+    monkeypatch.delenv("PRAISONAI_HOST_LEGACY", raising=False)
+    import praisonaiui.server as srv
+    from praisonai.integration import host_app
+
+    host_app._CONFIGURED = False
+    srv._provider = None
+    monkeypatch.setattr(srv, "set_provider", lambda p: setattr(srv, "_provider", p))
+    app = host_app.build_host_app(pages=["chat"])
+    srv.set_provider(_MockProvider())
+    return TestClient(app)
+
+
+def test_run_sse_stream(host_client):
+    with host_client.stream(
+        "POST",
+        "/run",
+        json={"message": "hi", "session_id": "sse-test"},
+        headers={"Accept": "text/event-stream"},
+    ) as response:
+        assert response.status_code == 200
+        chunks = list(response.iter_text())
+    body = "".join(chunks)
+    assert "data:" in body or len(chunks) > 0

src/praisonai/uv.lock

@@ -4983,7 +4983,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "1.6.43"
+version = "1.6.45"
 source = { directory = "../praisonai-agents" }
 dependencies = [
     { name = "aiohttp" },