fix: set completed status on workflow cache hit

A cache hit reusing a step's result left stale status: a step skipped
on an earlier run kept 'skipped' even after the gate reopened and a
cached result was served. Mark both step.status and step_statuses as
'completed' on cache hits (execution callbacks stay disabled since
nothing runs). Addresses CodeRabbit review on PR #5421.

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/DOCS_PARITY.md

@@ -25,41 +25,41 @@ This report compares **Python SDK feature categories** against **Python document
 | ✅ Audio | 2 | 12 | 1000 |
 | ✅ Auto Generation | 5 | 12 | 3941 |
 | ✅ Autonomy | 3 | 7 | 2805 |
-| ✅ Bots | 7 | 38 | 15809 |
+| ✅ Bots | 7 | 38 | 15546 |
 | ✅ Budget | 1 | 1 | 287 |
-| ✅ CLI | 5 | 124 | 39191 |
+| ✅ CLI | 5 | 124 | 39107 |
 | ✅ Chunking | 2 | 2 | 435 |
 | ✅ Citations | 2 | 1 | 202 |
 | ✅ Code Execution | 2 | 14 | 5022 |
 | ✅ Conditions | 1 | 3 | 1386 |
 | ✅ Configuration | 3 | 8 | 3341 |
-| ✅ Context Management | 16 | 36 | 14384 |
+| ✅ Context Management | 16 | 36 | 14331 |
 | ✅ Deep Research | 8 | 2 | 591 |
 | ✅ Display | 6 | 3 | 844 |
 | ✅ Embeddings | 6 | 23 | 2051 |
-| ✅ Evaluation | 1 | 9 | 3637 |
+| ✅ Evaluation | 1 | 9 | 3537 |
 | ✅ Events | 1 | 2 | 782 |
-| ✅ Execution | 3 | 4 | 1694 |
+| ✅ Execution | 3 | 4 | 1693 |
 | ✅ Failover | 2 | 1 | 537 |
 | ✅ Files | 2 | 7 | 2376 |
 | ✅ Flow | 1 | 3 | 811 |
-| ✅ Gateway | 7 | 97 | 37597 |
-| ✅ Guardrails | 5 | 4 | 2642 |
+| ✅ Gateway | 7 | 93 | 36047 |
+| ✅ Guardrails | 5 | 4 | 2638 |
 | ✅ Handoffs | 11 | 6 | 2748 |
 | ✅ Hooks | 2 | 9 | 5230 |
 | ✅ Image | 1 | 11 | 1295 |
-| ✅ Knowledge | 4 | 20 | 6260 |
+| ✅ Knowledge | 4 | 20 | 6188 |
 | ✅ LLM | 3 | 15 | 5958 |
-| ✅ Loops | 4 | 5 | 1478 |
-| ✅ MCP | 1 | 64 | 15116 |
-| ✅ Memory | 6 | 21 | 8198 |
+| ✅ Loops | 4 | 5 | 1475 |
+| ✅ MCP | 1 | 64 | 15068 |
+| ✅ Memory | 6 | 21 | 8194 |
 | ✅ OCR | 2 | 1 | 237 |
 | ✅ Observability | 2 | 23 | 2859 |
 | ✅ Optimizer | 1 | 2 | 792 |
 | ✅ Output | 3 | 5 | 1289 |
 | ✅ Parallel Execution | 3 | 2 | 528 |
-| ✅ Planning | 6 | 7 | 1827 |
-| ✅ Plugins | 8 | 5 | 4108 |
+| ✅ Planning | 6 | 7 | 1825 |
+| ✅ Plugins | 8 | 5 | 3785 |
 | ✅ Prompts | 2 | 9 | 2043 |
 | ✅ Providers | 1 | 58 | 8751 |
 | ✅ Query | 1 | 2 | 736 |
@@ -68,14 +68,14 @@ This report compares **Python SDK feature categories** against **Python document
 | ✅ Reflection | 3 | 3 | 732 |
 | ✅ Retrieval | 2 | 5 | 1046 |
 | ✅ Routing | 1 | 2 | 430 |
-| ✅ Sandbox | 5 | 7 | 3720 |
+| ✅ Sandbox | 5 | 7 | 3712 |
 | ✅ Security | 1 | 3 | 3642 |
-| ✅ Sessions | 4 | 21 | 8388 |
+| ✅ Sessions | 4 | 21 | 8386 |
 | ✅ Skills | 6 | 16 | 6045 |
 | ✅ Tasks | 2 | 8 | 3266 |
 | ✅ Telemetry | 1 | 2 | 660 |
 | ✅ Templates | 1 | 9 | 1859 |
-| ✅ Tools | 12 | 147 | 45378 |
+| ✅ Tools | 12 | 145 | 44877 |
 | ✅ Tracing | 3 | 2 | 139 |
 | ✅ Vector Store | 1 | 12 | 1219 |
 | ✅ Video | 2 | 10 | 1303 |
@@ -87,7 +87,7 @@ This report compares **Python SDK feature categories** against **Python document
 
 These docs exist but don't match any implemented feature category:
 
-- ℹ️ Database (44 docs, 8206 lines)
+- ℹ️ Database (44 docs, 8056 lines)
 - ℹ️ Documents (1 docs, 810 lines)
 - ℹ️ Teams (1 docs, 214 lines)
 

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

@@ -7,7 +7,10 @@ from praisonaiagents._logging import get_logger
 import asyncio
 import contextlib
 import threading
-from typing import List, Optional, Any, Dict, Union, Literal, TYPE_CHECKING, Callable
+import concurrent.futures
+import random
+import re
+from typing import List, Optional, Any, Dict, Union, Literal, TYPE_CHECKING, Callable, Generator
 from collections import OrderedDict
 import inspect
 
@@ -7123,25 +7126,6 @@ Answer:"""
         """The most recent rejection message a guardrail sent back to the model."""
         return getattr(self, "_last_guardrail_error", None)
 
-    def _guardrail_blocked_message(self, error=None):
-        """Fixed, user-facing string returned when an output guardrail blocks a run.
-
-        A blocked response must never surface the rejected (potentially unsafe)
-        model text, but it must also never be ``None`` - callers expect a
-        ``str`` and would otherwise crash on ``.strip()`` (issue #5596, FR-001).
-
-        The returned string is intentionally **fixed** and carries no diagnostic
-        detail: a guardrail's own rejection reason (especially an LLM validator's
-        free-text) can echo the blocked output, so appending it would leak the
-        very content the guardrail rejected. The reason is logged instead, where
-        it stays auditable without reaching the caller. The rejection is also
-        exposed programmatically via :attr:`last_guardrail_error`.
-        """
-        reason = str(error) if error else getattr(self, "_last_guardrail_error", None)
-        if reason:
-            logging.info(f"Agent {self.name}: guardrail block reason: {reason}")
-        return "I'm sorry, but I can't share that response because it did not pass the safety guardrail."
-
     def _process_guardrail(self, task_output):
         """Process the guardrail validation for a task output.
         

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

@@ -751,43 +751,6 @@ Your Goal: {self.goal}"""
         
         return None
 
-    def _coerce_structured_output(self, response_text, schema_model):
-        """Parse a string response into the requested Pydantic model.
-
-        When ``output_pydantic``/``output_json`` is a Pydantic model, callers
-        expect a validated instance rather than a raw string. This tolerates
-        markdown code fences around the JSON payload.
-
-        Returns the parsed model instance on success, or the original
-        ``response_text`` unchanged if parsing is not possible (preserves
-        backward compatibility for non-Pydantic schemas or malformed output).
-        """
-        if not schema_model or not isinstance(response_text, str):
-            return response_text
-        if not hasattr(schema_model, 'model_validate_json'):
-            return response_text
-
-        candidate = response_text.strip()
-        try:
-            return schema_model.model_validate_json(candidate)
-        except Exception:
-            pass
-
-        # Strip a surrounding markdown code fence (```json ... ```) if present
-        fenced = re.search(r"```(?:json)?\s*(.*?)\s*```", candidate, re.DOTALL)
-        if fenced:
-            try:
-                return schema_model.model_validate_json(fenced.group(1).strip())
-            except Exception:
-                pass
-
-        logging.warning(
-            f"Agent {self.name}: output_pydantic={getattr(schema_model, '__name__', schema_model)} "
-            f"was requested but the response could not be parsed into the model; "
-            f"returning raw string."
-        )
-        return response_text
-
     def _supports_native_structured_output(self):
         """Check if current model supports native structured output via response_format.
         
@@ -3599,13 +3562,6 @@ Your Goal: {self.goal}"""
                     # 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, cancel_token=cancel_token, messages=processed_history)
-                        # Coerce into the requested Pydantic model so callers
-                        # get a validated instance, not a raw JSON string.
-                        # Only output_pydantic changes the return type; output_json
-                        # must stay a string for backward-compatible downstream
-                        # JSON parsing (task parsing, workflow results).
-                        if output_pydantic:
-                            validated_response = self._coerce_structured_output(validated_response, output_pydantic)
                         # Execute callback and display after validation
                         self._execute_callback_and_display(prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
                         return self._trigger_after_agent_hook(prompt, validated_response, start_time)
@@ -3613,7 +3569,7 @@ Your Goal: {self.goal}"""
                         logging.error(f"Agent {self.name}: Guardrail validation failed for custom LLM: {e}")
                         # Rollback chat history on guardrail failure
                         self._rollback_chat_history_to(chat_history_length)
-                        return self._guardrail_blocked_message(e)
+                        return None
                 except ToolExecutionError:
                     raise
                 except Exception as e:
@@ -3737,11 +3693,6 @@ Your Goal: {self.goal}"""
                             # Apply guardrail validation even for JSON output
                             try:
                                 validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id, cancel_token=cancel_token, messages=messages)
-                                # Coerce into the requested Pydantic model so callers
-                                # get a validated instance, not a raw JSON string.
-                                # output_json intentionally stays a string (backward compat).
-                                if output_pydantic:
-                                    validated_response = self._coerce_structured_output(validated_response, output_pydantic)
                                 # Execute callback after validation
                                 self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
                                 return self._trigger_after_agent_hook(original_prompt, validated_response, start_time)
@@ -3749,7 +3700,7 @@ Your Goal: {self.goal}"""
                                 logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
                                 # Rollback chat history on guardrail failure
                                 self._rollback_chat_history_to(chat_history_length)
-                                return self._guardrail_blocked_message(e)
+                                return None
 
                         if not self.self_reflect:
                             # User message already added before LLM call via _build_messages
@@ -3770,7 +3721,7 @@ Your Goal: {self.goal}"""
                                     logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
                                     # Rollback chat history on guardrail failure
                                     self._rollback_chat_history_to(chat_history_length)
-                                    return self._guardrail_blocked_message(e)
+                                    return None
                             # Apply guardrail to regular response
                             try:
                                 validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id, cancel_token=cancel_token, messages=messages)
@@ -3781,7 +3732,7 @@ Your Goal: {self.goal}"""
                                 logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
                                 # Rollback chat history on guardrail failure
                                 self._rollback_chat_history_to(chat_history_length)
-                                return self._guardrail_blocked_message(e)
+                                return None
 
                         reflection_prompt = f"""
 Reflect on your previous response: '{response_text}'.
@@ -3849,7 +3800,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                     # Rollback chat history on guardrail failure
                                     self._rollback_chat_history_to(chat_history_length)
                                     self._end_run(None, "error", {"error": str(e)})
-                                    return self._guardrail_blocked_message(e)
+                                    return None
 
                             # Check if we've hit max reflections
                             if reflection_count >= self.max_reflect - 1:
@@ -3867,7 +3818,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                     logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
                                     # Rollback chat history on guardrail failure
                                     self._rollback_chat_history_to(chat_history_length)
-                                    return self._guardrail_blocked_message(e)
+                                    return None
                             
                             # If not satisfactory and not at max reflections, continue with regeneration
                             logging.debug(f"{self.name} reflection count {reflection_count + 1}, continuing reflection process")
@@ -4277,11 +4228,6 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                     # Apply guardrail validation for custom LLM response
                     try:
                         validated_response = await self._aapply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id, messages=effective_history)
-                        # Coerce into the requested Pydantic model so callers
-                        # get a validated instance, not a raw JSON string.
-                        # output_json intentionally stays a string (backward compat).
-                        if output_pydantic:
-                            validated_response = self._coerce_structured_output(validated_response, output_pydantic)
                         # 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)
@@ -4289,7 +4235,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                         logging.error(f"Agent {self.name}: Guardrail validation failed for custom LLM: {e}")
                         # Rollback chat history on guardrail failure
                         self._rollback_chat_history_to(chat_history_length)
-                        return self._guardrail_blocked_message(e)
+                        return None
                 except ToolExecutionError:
                     raise
                 except Exception as e:
@@ -4427,11 +4373,6 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                             # 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, messages=messages)
-                                # Coerce into the requested Pydantic model so callers
-                                # get a validated instance, not a raw JSON string.
-                                # output_json intentionally stays a string (backward compat).
-                                if output_pydantic:
-                                    validated_response = self._coerce_structured_output(validated_response, output_pydantic)
                                 # 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)
@@ -4439,7 +4380,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                 logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
                                 # Rollback chat history on guardrail failure
                                 self._rollback_chat_history_to(chat_history_length)
-                                return self._guardrail_blocked_message(e)
+                                return None
 
                         # For regular responses (no self-reflection)
                         if not self.self_reflect:
@@ -4462,7 +4403,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                     logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
                                     # Rollback chat history on guardrail failure
                                     self._rollback_chat_history_to(chat_history_length)
-                                    return self._guardrail_blocked_message(e)
+                                    return None
                             else:
                                 # Apply guardrail to regular response content
                                 try:
@@ -4474,7 +4415,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                     logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
                                     # Rollback chat history on guardrail failure
                                     self._rollback_chat_history_to(chat_history_length)
-                                    return self._guardrail_blocked_message(e)
+                                    return None
                         
                         # If self-reflection is enabled, implement reflection logic
                         if self.self_reflect:
@@ -4514,7 +4455,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                             logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
                                             # Rollback chat history on guardrail failure
                                             self._rollback_chat_history_to(chat_history_length)
-                                            return self._guardrail_blocked_message(e)
+                                            return None
                                     
                                     _areflection_kwargs = {}
                                     if temperature is not None:
@@ -4548,7 +4489,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                             logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
                                             # Rollback chat history on guardrail failure
                                             self._rollback_chat_history_to(chat_history_length)
-                                            return self._guardrail_blocked_message(e)
+                                            return None
                                     
                                     # Check if we've hit max reflections
                                     if reflection_count >= self.max_reflect - 1:
@@ -4567,7 +4508,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                             logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
                                             # Rollback chat history on guardrail failure
                                             self._rollback_chat_history_to(chat_history_length)
-                                            return self._guardrail_blocked_message(e)
+                                            return None
                                     
                                     # Regenerate response based on reflection
                                     regenerate_messages = reflection_messages + [
@@ -4617,7 +4558,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                                             logging.error(f"Agent {self.name}: Guardrail validation failed after reflection error: {guard_e}")
                                             # Rollback chat history on guardrail failure
                                             self._rollback_chat_history_to(chat_history_length)
-                                            return self._guardrail_blocked_message(guard_e)
+                                            return None
                                     continue
                         
                         # This should never be reached due to the returns above
@@ -4667,11 +4608,6 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                             if get_logger().getEffectiveLevel() == logging.DEBUG:
                                 total_time = time.time() - start_time
                                 logging.debug(f"Agent.achat completed in {total_time:.2f} seconds")
-                            # Coerce into the requested Pydantic model so callers
-                            # get a validated instance, not a raw JSON string.
-                            # output_json intentionally stays a string (backward compat).
-                            if output_pydantic:
-                                response_text = self._coerce_structured_output(response_text, output_pydantic)
                             # Execute callback after JSON/Pydantic completion
                             self._execute_callback_and_display(original_prompt, response_text, time.time() - start_time, task_name, task_description, task_id)
                             return await self._atrigger_after_agent_hook(original_prompt, response_text, start_time)
@@ -4786,7 +4722,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                             logging.error(f"Agent {self.name}: Guardrail validation failed for OpenAI client: {e}")
                             # Rollback chat history on guardrail failure
                             self._rollback_chat_history_to(chat_history_length)
-                            return self._guardrail_blocked_message(e)
+                            return None
                 except ToolExecutionError:
                     raise
                 except Exception as e:

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

@@ -764,50 +764,6 @@ class MemoryMixin:
         except Exception as e:
             logging.debug(f"Auto-memory extraction failed: {e}")
 
-    def _persist_memory_turn(self, user_message: str, assistant_response: str):
-        """Persist a raw conversation turn to the memory store.
-
-        Called after each response when ``_memory_instance`` is set but
-        ``auto_memory`` extraction is off. Stores the turn to short-term memory
-        so a later Agent sharing the same store (e.g. ``memory={"user_id": uid}``)
-        recalls it via ``get_memory_context()``. Best-effort: never raises.
-
-        Async-safe: a memory store write is blocking file/DB I/O. When this runs
-        inside a live event loop (the async ``achat``/``astart`` after-agent path),
-        the write is offloaded to a worker thread so it never stalls the loop
-        (AGENTS.md §4.5). On the sync path (no running loop) it writes inline.
-        """
-        memory = getattr(self, "_memory_instance", None)
-        if memory is None or not assistant_response:
-            return
-        store = getattr(memory, "store_short_term", None) or getattr(memory, "add_short_term", None)
-        if store is None:
-            return
-        text = f"User: {user_message}\nAssistant: {assistant_response}"
-        metadata = {"agent_id": getattr(self, "agent_id", getattr(self, "name", None))}
-
-        def _do_store():
-            try:
-                store(text, metadata=metadata)
-            except Exception as e:
-                logging.debug(f"Memory turn persistence failed: {e}")
-
-        try:
-            import asyncio
-            asyncio.get_running_loop()
-        except RuntimeError:
-            # No running loop (sync chat/run/start path): write inline.
-            _do_store()
-            return
-        # Inside a live loop: offload the blocking write so it cannot stall the
-        # event loop. Fire-and-forget — persistence is best-effort.
-        try:
-            import asyncio
-            asyncio.get_running_loop().run_in_executor(None, _do_store)
-        except Exception as e:
-            logging.debug(f"Memory turn persistence scheduling failed: {e}")
-            _do_store()
-
     def _process_auto_learning(self):
         """Process auto-learning extraction after agent response.
         

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

@@ -1904,11 +1904,6 @@ class ToolExecutionMixin:
         ``_atrigger_after_agent_hook`` without any async-safety change.
         """
         from ..hooks import HookEvent, AfterAgentInput
-        # AfterAgentInput.response is typed/serialized as a string (its
-        # to_dict truncates via ``response[:500]``). Structured-output turns
-        # (output_pydantic) return a Pydantic instance, so coerce to str here
-        # to keep string-only hooks (e.g. command hooks) from crashing while
-        # the public chat() return value stays the model.
         return AfterAgentInput(
             session_id=getattr(self, '_session_id', 'default'),
             cwd=os.getcwd(),
@@ -1916,7 +1911,7 @@ class ToolExecutionMixin:
             timestamp=str(time.time()),
             agent_name=self.name,
             prompt=prompt if isinstance(prompt, str) else str(prompt),
-            response=response if isinstance(response, str) else (str(response) if response else ""),
+            response=response or "",
             tools_used=tools_used or [],
             total_tokens=0,
             execution_time_ms=(time.time() - start_time) * 1000
@@ -1933,16 +1928,7 @@ class ToolExecutionMixin:
         # Auto-memory extraction (opt-in via MemoryConfig(auto_memory=True))
         if response:
             prompt_str = prompt if isinstance(prompt, str) else str(prompt)
-            if getattr(self, "_auto_memory", False):
-                self._process_auto_memory(prompt_str, str(response))
-            else:
-                # When memory is enabled but auto_memory extraction is off, still
-                # persist the raw turn so a later Agent sharing the same memory
-                # (e.g. memory={"user_id": uid}) can recall it. Without this the
-                # sync chat/run/start path wrote nothing and cross-session recall
-                # silently returned empty (issue #5595). Mirrors the write already
-                # done on the async unified path.
-                self._persist_memory_turn(prompt_str, str(response))
+            self._process_auto_memory(prompt_str, str(response))
 
         # Auto-learning extraction (opt-in via LearnConfig(mode=LearnMode.AGENTIC))
         self._process_auto_learning()

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

@@ -1002,6 +1002,14 @@ Respond with ONLY a valid JSON tool call in this format:
         ]):
             return "idle_timeout"
         
+        # Format errors
+        if any(indicator in error_str for indicator in [
+            "validation error", "invalid format", "parse error",
+            "parsing error", "decode error", "malformed",
+            "invalid json", "schema error"
+        ]):
+            return "format_error"
+        
         # Default fallback
         return "unknown"
     
@@ -2031,6 +2039,98 @@ Respond with ONLY a valid JSON tool call in this format:
                 schemas.append(f"- {name}({', '.join(param_strs)})")
         return '\n'.join(schemas) if schemas else "None"
 
+    def _param_accepts_string(self, function_name, param_name, tools) -> bool:
+        """True when a tool declares this parameter as accepting a string.
+
+        Used to decide whether an argument whose value matches a previous tool's
+        name is a legitimate string or a weak model's way of referring to that
+        tool's result. A declared string parameter is left alone.
+        """
+        try:
+            for tool in tools or []:
+                fn = tool.get("function") if isinstance(tool, dict) else None
+                if not fn or fn.get("name") != function_name:
+                    continue
+                spec = (fn.get("parameters") or {}).get("properties", {}).get(param_name)
+                if not isinstance(spec, dict):
+                    return False
+                return self._schema_accepts_string(spec)
+        except Exception:  # noqa: BLE001 -- never break a tool call on a schema read
+            return False
+        return False
+
+    def _schema_accepts_string(self, spec) -> bool:
+        """True when a JSON-Schema fragment can accept a string value.
+
+        Handles the shapes this repo's own generator emits: a bare
+        ``{"type": "string"}``, a list type ``{"type": ["string", "null"]}``,
+        and the ``anyOf``/``oneOf``/``allOf`` unions produced for ``Optional[str]``
+        and ``Union`` parameters (``tools/schema.py``). Without the union case an
+        ``Optional[str]`` argument was still treated as a result reference and
+        silently overwritten.
+        """
+        if not isinstance(spec, dict):
+            return False
+        declared = spec.get("type")
+        if isinstance(declared, list):
+            if "string" in declared:
+                return True
+        elif declared == "string":
+            return True
+        # enum without an explicit type is string-typed in JSON Schema practice
+        if declared is None and isinstance(spec.get("enum"), list):
+            if any(isinstance(v, str) for v in spec["enum"]):
+                return True
+        for key in ("anyOf", "oneOf", "allOf"):
+            members = spec.get(key)
+            if isinstance(members, list) and any(
+                    self._schema_accepts_string(m) for m in members):
+                return True
+        return False
+
+    def _force_tool_usage_message(self, response_text, tool_calls, formatted_tools,
+                                  iteration_count):
+        """The nudge to send when a model ignored tools it should have used.
+
+        Returns the message content, or None when no nudge is warranted. Shared
+        so every response path applies the same policy -- this lived inline in
+        get_response only, so an agent that was awaited instead of called
+        silently lost a setting it had accepted.
+        """
+        if not self._should_force_tool_usage(
+                response_text, tool_calls, formatted_tools, iteration_count):
+            return None
+        tool_names = self._get_tool_names_for_prompt(formatted_tools)
+        logging.debug(
+            f"[OLLAMA_RELIABILITY] Force tool usage triggered. Adding prompt for tools: {tool_names}")
+        return self.FORCE_TOOL_USAGE_PROMPT.format(tool_names=tool_names)
+
+    def _tool_repair_message(self, tool_calls, formatted_tools):
+        """The correction to send when a model produced an invalid tool call.
+
+        Returns the message content and charges the repair budget, or None when
+        the calls are valid or the budget is spent. Shared for the same reason
+        as _force_tool_usage_message.
+        """
+        repair_attempt_count = getattr(self, '_current_repair_count', 0)
+        max_tool_repairs = getattr(self, 'max_tool_repairs', 0)
+        if not (tool_calls and max_tool_repairs > 0
+                and repair_attempt_count < max_tool_repairs):
+            return None
+        validation_errors = [e for e in
+                             (self._validate_tool_call(tc, formatted_tools) for tc in tool_calls)
+                             if e]
+        if not validation_errors:
+            return None
+        error_msg = "; ".join(validation_errors)
+        tool_schemas = self._get_tool_schemas_for_prompt(formatted_tools)
+        logging.debug(
+            f"[OLLAMA_RELIABILITY] Tool call repair attempt "
+            f"{repair_attempt_count + 1}/{max_tool_repairs}: {error_msg}")
+        self._current_repair_count = repair_attempt_count + 1
+        return self.TOOL_CALL_REPAIR_PROMPT.format(
+            error=error_msg, tool_schemas=tool_schemas)
+
     def _should_force_tool_usage(self, response_text: str, tool_calls: Optional[List], formatted_tools: Optional[List], iteration_count: int) -> bool:
         """
         Determine if we should force tool usage based on settings and context.
@@ -7028,9 +7128,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                     # (input_text / input_image); local image paths become
                     # data URLs. Plain-string content is passed through.
                     from .openai_client import OpenAIClient
-                    # Do not forward persisted metadata/provider extensions as
-                    # Responses message properties.
-                    item = {"role": role}
+                    item = dict(msg)
                     item["content"] = OpenAIClient._build_responses_content(
                         msg.get("content", "")
                     )

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

@@ -781,9 +781,7 @@ class OpenAIClient:
                         "output": msg.get("content", ""),
                     })
                 else:
-                    # Persisted history may carry local metadata/provider fields.
-                    # Responses message items accept role/content, not that record.
-                    item = {"role": role}
+                    item = dict(msg)
                     item["content"] = self._build_responses_content(
                         msg.get("content", "")
                     )

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

@@ -73,7 +73,7 @@ class SqliteMemoryAdapter:
 
     def _get_stm_conn(self):
         """Get thread-local short-term memory connection."""
-        if getattr(self._local, 'stm_conn', None) is None:
+        if not hasattr(self._local, 'stm_conn'):
             self._local.stm_conn = sqlite3.connect(
                 self.short_db,
                 check_same_thread=False,
@@ -97,7 +97,7 @@ class SqliteMemoryAdapter:
     
     def _get_ltm_conn(self):
         """Get thread-local long-term memory connection."""
-        if getattr(self._local, 'ltm_conn', None) is None:
+        if not hasattr(self._local, 'ltm_conn'):
             self._local.ltm_conn = sqlite3.connect(
                 self.long_db,
                 check_same_thread=False,
@@ -272,30 +272,4 @@ class SqliteMemoryAdapter:
                     conn.close()
                 except Exception:
                     pass
-            self._all_connections.clear()
-
-    def close_thread_connections(self):
-        """Close only the calling thread's connections.
-
-        Leaves the shared adapter and other threads' connections intact so the
-        adapter can be reused; the calling thread lazily reopens its own
-        connections on next access.
-
-        A connection is only removed from the registry and cleared from
-        thread-local storage after it closes successfully. If ``close()`` raises,
-        the connection stays tracked so a later ``close_connections()`` /
-        ``close_thread_connections()`` can retry it, and the failure is logged
-        rather than silently swallowed.
-        """
-        for attr in ("stm_conn", "ltm_conn"):
-            conn = getattr(self._local, attr, None)
-            if conn is None:
-                continue
-            try:
-                conn.close()
-            except Exception as e:
-                logger.warning(f"Failed to close {attr}; keeping it tracked for retry: {e}")
-                continue
-            with self._connection_lock:
-                self._all_connections.discard(conn)
-            setattr(self._local, attr, None)
+            self._all_connections.clear()

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

@@ -266,12 +266,11 @@ class FileMemory:
             return True
         except (IOError, OSError) as e:
             self._log(f"Error writing {filepath}: {e}", logging.ERROR)
-            return False
-        finally:
             try:
                 tmp_path.unlink()
             except OSError:
                 pass
+            return False
 
     def commit_memory_batch(
         self,

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

@@ -2532,31 +2532,16 @@ class Memory(SearchMixin, MemoryCoreMixin):
         # Other active threads should manage their own connections
         # Only clear full registry in __del__ or explicit shutdown
         
-        # Close memory adapter if it exists (protocol-driven path).
-        #
-        # SQLite is thread-scoped by contract (#1562/#1563): close only the
-        # calling thread's adapter connections and KEEP the shared adapter so
-        # other threads and subsequent same-thread searches keep using the
-        # adapter's tables. Tearing it down here would make searches fall back
-        # to legacy SQL against short_mem/long_mem (#5486).
+        # Close memory adapter if it exists (protocol-driven path)
         if hasattr(self, 'memory_adapter') and self.memory_adapter:
-            if isinstance(self.memory_adapter, SqliteMemoryAdapter):
-                try:
-                    if hasattr(self.memory_adapter, 'close_thread_connections'):
-                        self.memory_adapter.close_thread_connections()
-                        logger.debug("SQLite adapter thread connections closed")
-                except Exception as e:
-                    logger.warning(f"Error closing SQLite adapter thread connections: {e}")
-                # Intentionally preserve self.memory_adapter for reuse.
-            else:
-                try:
-                    if hasattr(self.memory_adapter, 'close') and callable(self.memory_adapter.close):
-                        self.memory_adapter.close()
-                        logger.debug("Memory adapter closed successfully")
-                except Exception as e:
-                    logger.warning(f"Error closing memory adapter: {e}")
-                finally:
-                    self.memory_adapter = None
+            try:
+                if hasattr(self.memory_adapter, 'close') and callable(self.memory_adapter.close):
+                    self.memory_adapter.close()
+                    logger.debug("Memory adapter closed successfully")
+            except Exception as e:
+                logger.warning(f"Error closing memory adapter: {e}")
+            finally:
+                self.memory_adapter = None
         
         # Close MongoDB client if it exists (legacy path)
         if hasattr(self, 'mongo_client') and self.mongo_client:

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

@@ -13,7 +13,6 @@ import uuid
 import json
 import time
 import logging
-from copy import deepcopy
 from typing import Any, Dict, List, Optional, TYPE_CHECKING
 
 logger = logging.getLogger(__name__)
@@ -503,10 +502,6 @@ class Session:
     ) -> bool:
         """Persist one sub-agent transcript onto this session's own record.
 
-        Built-in stores merge the entry under their session file lock. Older
-        custom stores retain the bounded read/write verification fallback below;
-        that fallback cannot guarantee atomic preservation of concurrent entries.
-
         The store replaces the whole ``AGENT_HISTORY_KEY`` map per write, so a
         naive read-then-replace would let a concurrent writer (another agent
         under the same parent) clobber this agent's entry, or vice versa. We
@@ -514,14 +509,6 @@ class Session:
         *other* key changed underneath us, retrying a bounded number of times.
         Legacy per-agent records are always merged in (they stay on disk).
         """
-        merge_map = getattr(session_store, "merge_session_metadata_map", None)
-        if callable(merge_map):
-            # Built-in stores merge inside their locked read/modify/write path.
-            # A post-write comparison cannot detect entries lost to a stale map.
-            return merge_map(
-                self.session_id, AGENT_HISTORY_KEY, {agent_key: messages},
-                defaults=self._legacy_agent_histories(session_store),
-            )
         update_meta = getattr(session_store, "update_session_metadata", None)
         if not callable(update_meta):
             return False
@@ -577,7 +564,6 @@ class Session:
                         {
                             "role": msg.get("role", "user"),
                             "content": msg.get("content", ""),
-                            **deepcopy(msg),
                         }
                         for msg in chat_history
                         if isinstance(msg, dict)

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

@@ -34,7 +34,7 @@ from typing import Any, Dict, List, Optional
 
 from praisonaiagents._logging import get_logger
 
-from .store import DefaultSessionStore, FileLock, SessionData
+from .store import DefaultSessionStore, SessionData
 
 logger = get_logger(__name__)
 
@@ -370,27 +370,6 @@ class SqliteSessionStore(DefaultSessionStore):
             self._index_session(session)
         return ok
 
-    def _save_imported_session(self, session: SessionData, *, overwrite: bool = True) -> bool:
-        """Refresh recall and routing after a verbatim portable restore."""
-        ok = super()._save_imported_session(session, overwrite=overwrite)
-        if ok:
-            try:
-                filepath = self._get_session_path(session.session_id)
-                with FileLock(filepath, self.lock_timeout):
-                    # Read-only cache fallback is unsafe after an overwrite:
-                    # it can restore routing fields removed by the import.
-                    fresh = self._load_session_from_disk(session.session_id, filepath)
-                    self._reingest_spill(session.session_id, fresh)
-                with self._lock:
-                    self._cache[session.session_id] = fresh
-                self._index_session(fresh)
-            except Exception as exc:
-                # The JSON write succeeded, but the old index is no longer a
-                # trustworthy view. Fail closed until a later refresh/rebuild.
-                self._deindex_session(session.session_id)
-                logger.debug("Post-import index refresh failed for %s: %s", session.session_id, exc)
-        return ok
-
     def _modify_session_locked(self, session_id, mutator, **kwargs) -> bool:
         """Refresh the index after any locked read-modify-write.
 

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

@@ -158,9 +158,7 @@ class SqliteTranscriptStore(DefaultSessionStore):
         if not session_dir or not os.path.isdir(session_dir):
             return
         try:
-            filenames = sorted(
-                f for f in os.listdir(session_dir) if f.endswith(".json")
-            )
+            filenames = [f for f in os.listdir(session_dir) if f.endswith(".json")]
         except OSError:
             return
         if not filenames:
@@ -177,7 +175,7 @@ class SqliteTranscriptStore(DefaultSessionStore):
             try:
                 with open(filepath, "r", encoding="utf-8") as f:
                     data = json.load(f)
-            except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError):
+            except (OSError, json.JSONDecodeError, TypeError):
                 continue
             if not isinstance(data, dict):
                 continue
@@ -216,7 +214,7 @@ class SqliteTranscriptStore(DefaultSessionStore):
         except (json.JSONDecodeError, TypeError):
             return None
 
-    def _write_row(self, session: SessionData, conn=None, *, overwrite: bool = True) -> bool:
+    def _write_row(self, session: SessionData, conn=None) -> bool:
         own_lock = conn is None
         if conn is None:
             conn = self._connect()
@@ -235,21 +233,17 @@ class SqliteTranscriptStore(DefaultSessionStore):
             getattr(session, "updated_at", None),
         )
         sql = (
-            ("INSERT OR REPLACE" if overwrite else "INSERT OR IGNORE") + " INTO sessions "
+            "INSERT OR REPLACE INTO sessions "
             "(session_id, data, agent_name, gateway_session_id, agent_id, updated_at) "
             "VALUES (?, ?, ?, ?, ?, ?)"
         )
         try:
             if own_lock:
                 with self._db_lock:
-                    cursor = conn.execute(sql, params)
+                    conn.execute(sql, params)
             else:
-                cursor = conn.execute(sql, params)
-            if not overwrite and cursor.rowcount == 0:
-                raise FileExistsError(session.session_id)
+                conn.execute(sql, params)
             return True
-        except FileExistsError:
-            raise
         except Exception as exc:
             logger.error("Failed to write session %s: %s", session.session_id, exc)
             return False
@@ -292,52 +286,6 @@ class SqliteTranscriptStore(DefaultSessionStore):
                 self._cache[session.session_id] = session
             return True
 
-    def _save_imported_session(self, session: SessionData, *, overwrite: bool = True) -> bool:
-        """Restore to the database without applying the destination's window."""
-        session.updated_at = datetime.now(timezone.utc).isoformat()
-        with self._db_lock:
-            # INSERT OR IGNORE makes no-clobber atomic across store instances;
-            # checking session_exists before a replace would still race.
-            if not self._write_row(session, overwrite=overwrite):
-                return False
-            with self._lock:
-                self._cache[session.session_id] = session
-            return True
-
-    def export_all(self) -> Dict[str, Any]:
-        """Export durable database rows, not legacy JSON sidecar files."""
-        return {
-            "version": self.PORTABLE_VERSION,
-            "sessions": [data for data in self._all_rows() if isinstance(data, dict)],
-        }
-
-    def _collect_lineage(
-        self, session: SessionData, *, exclude: str
-    ) -> List[Dict[str, Any]]:
-        """Collect persisted database records from the same conversation chain."""
-        lineage = self._lineage_key(session.to_dict())
-        if not lineage:
-            return []
-
-        def record_lineage(raw):
-            # Preserve the writer's Python JSON semantics (including NaN)
-            # and the shared lineage precedence, without retaining other rows.
-            try:
-                data = json.loads(raw)
-                return self._lineage_key(data) if isinstance(data, dict) else None
-            except (json.JSONDecodeError, TypeError, AttributeError):
-                return None
-
-        conn = self._connect()
-        with self._db_lock:
-            conn.create_function("portable_session_lineage", 1, record_lineage)
-            rows = conn.execute(
-                "SELECT data FROM sessions WHERE session_id != ? "
-                "AND portable_session_lineage(data) = ? ORDER BY updated_at DESC",
-                (exclude, lineage),
-            ).fetchall()
-        return [json.loads(row[0]) for row in rows]
-
     def _modify_session_locked(
         self,
         session_id: str,

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

@@ -1622,27 +1622,6 @@ class DefaultSessionStore:
             session_id, _apply, error_label="update session metadata"
         )
 
-    def merge_session_metadata_map(
-        self, session_id: str, key: str, updates: Dict[str, Any],
-        *, defaults: Optional[Dict[str, Any]] = None,
-    ) -> bool:
-        """Merge nested metadata entries using the freshest map under FileLock.
-
-        Defaults fill missing entries; persisted entries win over defaults,
-        and the supplied updates win over both.
-        """
-        def _apply(session: SessionData) -> None:
-            merged = dict(defaults or {})
-            current = session.metadata.get(key)
-            if isinstance(current, dict):
-                merged.update(current)
-            merged.update(updates)
-            session.metadata[key] = merged
-
-        return self._modify_session_locked(
-            session_id, _apply, error_label="merge session metadata map"
-        )
-
     def rename_session(self, session_id: str, title: str) -> bool:
         """Give a session a human-readable title (Issue #3737).
 
@@ -2327,7 +2306,7 @@ class DefaultSessionStore:
                 out.append(data)
         return out
 
-    def _save_imported_session(self, session: SessionData, *, overwrite: bool = True) -> bool:
+    def _save_imported_session(self, session: SessionData) -> bool:
         """Persist a restored session verbatim (no retention/window applied).
 
         Mirrors ``_save_session`` (timestamp + atomic, file-locked write) but
@@ -2338,10 +2317,6 @@ class DefaultSessionStore:
         filepath = self._get_session_path(session.session_id)
         session.updated_at = datetime.now(timezone.utc).isoformat()
         with FileLock(filepath, self.lock_timeout):
-            # The initial import check can race with a peer creating this file.
-            # Check again inside the same lock that protects the replacement.
-            if not overwrite and os.path.exists(filepath):
-                raise FileExistsError(filepath)
             if not self._atomic_write_json(filepath, session.to_dict()):
                 logger.error(f"Failed to save imported session {session.session_id}")
                 return False
@@ -2441,7 +2416,7 @@ class DefaultSessionStore:
                 # Persist the imported record verbatim: an import is a restore,
                 # so the destination's retention/active_window must not truncate
                 # or compact a valid larger export before it lands on disk.
-                if not self._save_imported_session(session, overwrite=overwrite):
+                if not self._save_imported_session(session):
                     report.skipped.append(
                         {"session_id": session_id, "reason": "write failed"}
                     )
@@ -2449,10 +2424,6 @@ class DefaultSessionStore:
                 with self._lock:
                     self._cache[session_id] = session
                 report.imported += 1
-            except FileExistsError:
-                report.skipped.append(
-                    {"session_id": session_id, "reason": "already exists (use overwrite)"}
-                )
             except Exception as e:  # pragma: no cover - defensive; one bad record
                 report.skipped.append(
                     {"session_id": session_id, "reason": f"import error: {e}"}

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

@@ -66,9 +66,6 @@ def generate_title(
         ... )
         >>> # Returns something like: "Python Code Debugging Help"
     """
-    if max_length <= 0:
-        return ""
-
     # Fallback title from user message if generation fails
     fallback_title = _create_fallback_title(user_msg, max_length)
     
@@ -122,9 +119,6 @@ async def generate_title_async(
     Returns:
         Generated title string, or fallback based on user message
     """
-    if max_length <= 0:
-        return ""
-
     fallback_title = _create_fallback_title(user_msg, max_length)
     
     try:
@@ -158,7 +152,8 @@ Return ONLY the title text, no quotes, no explanation."""
                 title = title.replace('\n', ' ').replace('\r', '')
                 
                 # Truncate if too long
-                title = _truncate_title(title, max_length)
+                if len(title) > max_length:
+                    title = title[:max_length-3] + "..."
                 
                 # Return if non-empty
                 if title and len(title.strip()) > 0:
@@ -181,17 +176,6 @@ Return ONLY the title text, no quotes, no explanation."""
     return fallback_title
 
 
-def _truncate_title(title: str, max_length: int) -> str:
-    """Apply the character cap without using negative ellipsis slices."""
-    if max_length <= 0:
-        return ""
-    if len(title) <= max_length:
-        return title
-    if max_length <= 3:
-        return title[:max_length]
-    return title[:max_length - 3] + "..."
-
-
 def _create_fallback_title(user_msg: str, max_length: int) -> str:
     """Create a fallback title from the user message.
     
@@ -203,7 +187,7 @@ def _create_fallback_title(user_msg: str, max_length: int) -> str:
         Simple title based on user message
     """
     if not user_msg or not user_msg.strip():
-        return _truncate_title("Chat Session", max_length)
+        return "Chat Session"
     
     # Clean up the message
     clean_msg = user_msg.strip()
@@ -216,4 +200,8 @@ def _create_fallback_title(user_msg: str, max_length: int) -> str:
     if '.' in clean_msg:
         clean_msg = clean_msg.split('.')[0]
     
-    return _truncate_title(clean_msg or "Chat Session", max_length)
+    # Truncate if too long
+    if len(clean_msg) > max_length:
+        clean_msg = clean_msg[:max_length-3] + "..."
+    
+    return clean_msg if clean_msg else "Chat Session"

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

@@ -69,17 +69,6 @@ class FileBackend:
         # Sanitize key for filesystem
         safe_key = "".join(c if c.isalnum() or c in "-_" else "_" for c in key)
         return self.storage_dir / f"{safe_key}{self.suffix}"
-
-    def _path_to_key(self, file_path: Path) -> Optional[str]:
-        """Recognize records using the full configured suffix."""
-        if not file_path.is_file():
-            return None
-        if self.suffix:
-            if file_path.name.endswith(self.suffix):
-                return file_path.name[:-len(self.suffix)]
-            return None
-        # Preserve the existing no-extension selection for an empty suffix.
-        return file_path.name if not file_path.suffix else None
     
     def save(self, key: str, data: Dict[str, Any]) -> None:
         """Save data with the given key."""
@@ -95,11 +84,11 @@ class FileBackend:
                     delete=False,
                     suffix=".tmp"
                 ) as f:
-                    temp_path = f.name
                     if self.pretty:
                         json.dump(data, f, indent=2, default=str, ensure_ascii=False)
                     else:
                         json.dump(data, f, default=str, ensure_ascii=False)
+                    temp_path = f.name
                 
                 os.replace(temp_path, file_path)
             except Exception as e:
@@ -121,7 +110,7 @@ class FileBackend:
         try:
             with open(file_path, "r", encoding="utf-8") as f:
                 return json.load(f)
-        except (json.JSONDecodeError, UnicodeDecodeError, IOError) as e:
+        except (json.JSONDecodeError, IOError) as e:
             logger.warning(f"Failed to load {key}: {e}")
             return None
     
@@ -144,8 +133,8 @@ class FileBackend:
         keys = []
         
         for file_path in self.storage_dir.iterdir():
-            key = self._path_to_key(file_path)
-            if key is not None:
+            if file_path.is_file() and file_path.suffix == self.suffix:
+                key = file_path.stem
                 if not prefix or key.startswith(prefix):
                     keys.append(key)
         
@@ -160,7 +149,7 @@ class FileBackend:
         count = 0
         with self._lock:
             for file_path in self.storage_dir.iterdir():
-                if self._path_to_key(file_path) is not None:
+                if file_path.is_file() and file_path.suffix == self.suffix:
                     try:
                         file_path.unlink()
                         count += 1
@@ -202,21 +191,13 @@ class SQLiteBackend:
         if not isinstance(table_name, str) or not _re.match(r'^[a-zA-Z0-9_]+$', table_name):
             raise ValueError("table_name must contain only alphanumeric characters and underscores")
         self.table_name = table_name
-        # Quote identifiers so validated names that are SQL keywords or start
-        # with a digit (e.g. "select", "9table") are valid in all statements.
-        self._quoted_table = f'"{table_name}"'
-        self._quoted_index = f'"idx_{table_name}_key"'
         self._local = threading.local()
         
         # Ensure directory exists
         Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
         
         if auto_create:
-            try:
-                self._create_table()
-            except Exception:
-                self.close()
-                raise
+            self._create_table()
     
     def _get_conn(self):
         """Get thread-local connection."""
@@ -239,7 +220,7 @@ class SQLiteBackend:
         cur = conn.cursor()
         
         cur.execute(f"""
-            CREATE TABLE IF NOT EXISTS {self._quoted_table} (
+            CREATE TABLE IF NOT EXISTS {self.table_name} (
                 key TEXT PRIMARY KEY,
                 data TEXT NOT NULL,
                 created_at REAL DEFAULT (strftime('%s', 'now')),
@@ -249,8 +230,8 @@ class SQLiteBackend:
         
         # Index for prefix queries
         cur.execute(f"""
-            CREATE INDEX IF NOT EXISTS {self._quoted_index} 
-            ON {self._quoted_table}(key)
+            CREATE INDEX IF NOT EXISTS idx_{self.table_name}_key 
+            ON {self.table_name}(key)
         """)
         
         conn.commit()
@@ -263,7 +244,7 @@ class SQLiteBackend:
         json_data = json.dumps(data, default=str, ensure_ascii=False)
         
         cur.execute(f"""
-            INSERT INTO {self._quoted_table} (key, data, updated_at)
+            INSERT INTO {self.table_name} (key, data, updated_at)
             VALUES (?, ?, strftime('%s', 'now'))
             ON CONFLICT(key) DO UPDATE SET
                 data = excluded.data,
@@ -278,7 +259,7 @@ class SQLiteBackend:
         cur = conn.cursor()
         
         cur.execute(f"""
-            SELECT data FROM {self._quoted_table} WHERE key = ?
+            SELECT data FROM {self.table_name} WHERE key = ?
         """, (key,))
         
         row = cur.fetchone()
@@ -295,7 +276,7 @@ class SQLiteBackend:
         cur = conn.cursor()
         
         cur.execute(f"""
-            DELETE FROM {self._quoted_table} WHERE key = ?
+            DELETE FROM {self.table_name} WHERE key = ?
         """, (key,))
         
         deleted = cur.rowcount > 0
@@ -309,13 +290,13 @@ class SQLiteBackend:
         
         if prefix:
             cur.execute(f"""
-                SELECT key FROM {self._quoted_table}
+                SELECT key FROM {self.table_name}
                 WHERE key LIKE ?
                 ORDER BY key
             """, (f"{prefix}%",))
         else:
             cur.execute(f"""
-                SELECT key FROM {self._quoted_table}
+                SELECT key FROM {self.table_name}
                 ORDER BY key
             """)
         
@@ -327,7 +308,7 @@ class SQLiteBackend:
         cur = conn.cursor()
         
         cur.execute(f"""
-            SELECT 1 FROM {self._quoted_table} WHERE key = ? LIMIT 1
+            SELECT 1 FROM {self.table_name} WHERE key = ? LIMIT 1
         """, (key,))
         
         return cur.fetchone() is not None
@@ -337,10 +318,10 @@ class SQLiteBackend:
         conn = self._get_conn()
         cur = conn.cursor()
         
-        cur.execute(f"SELECT COUNT(*) as count FROM {self._quoted_table}")
+        cur.execute(f"SELECT COUNT(*) as count FROM {self.table_name}")
         count = cur.fetchone()["count"]
         
-        cur.execute(f"DELETE FROM {self._quoted_table}")
+        cur.execute(f"DELETE FROM {self.table_name}")
         conn.commit()
         
         return count

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

@@ -262,8 +262,8 @@ class BaseJSONStore:
                 delete=False,
                 suffix=".tmp"
             ) as f:
-                temp_path = f.name
                 json.dump(self._data, f, indent=2, default=str, ensure_ascii=False)
+                temp_path = f.name
             
             # Atomic rename
             if self.use_file_lock:
@@ -343,7 +343,7 @@ class BaseJSONStore:
 def list_json_sessions(
     storage_dir: Path,
     suffix: str = ".json",
-    limit: Optional[int] = 50,
+    limit: int = 50,
 ) -> List[BaseSessionInfo]:
     """
     List all sessions in a directory.
@@ -353,7 +353,7 @@ def list_json_sessions(
     Args:
         storage_dir: Directory to search
         suffix: File suffix to filter by
-        limit: Maximum number of sessions to return, or None for all sessions
+        limit: Maximum number of sessions to return
         
     Returns:
         List of BaseSessionInfo, sorted by modification time (newest first)
@@ -370,7 +370,7 @@ def list_json_sessions(
                 # Try to get item count from file
                 if suffix == ".json":
                     try:
-                        with open(file_path, "r", encoding="utf-8") as f:
+                        with open(file_path, "r") as f:
                             data = json.load(f)
                             # Common patterns for item count
                             if "iterations" in data:
@@ -383,7 +383,7 @@ def list_json_sessions(
                         pass
                 elif suffix == ".jsonl":
                     try:
-                        with open(file_path, "r", encoding="utf-8") as f:
+                        with open(file_path, "r") as f:
                             info.item_count = sum(1 for _ in f)
                     except Exception:
                         pass
@@ -424,7 +424,7 @@ def cleanup_old_sessions(
     now = datetime.now()
     
     # Get all sessions sorted by age (oldest first)
-    sessions = list_json_sessions(storage_dir, suffix, limit=None)
+    sessions = list_json_sessions(storage_dir, suffix, limit=10000)
     sessions.sort(key=lambda s: s.modified_at)
     
     # Delete old sessions

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

@@ -1613,40 +1613,6 @@ class AgentFlow:
                 variables=all_variables.copy()
             )
 
-            # Step-result cache (opt-in via cache=). This linear path invokes
-            # handlers INLINE rather than through _execute_single_step_internal,
-            # which the pattern paths use -- so both are wrapped. Caching only
-            # one would make the feature work or not depending on whether a step
-            # happened to sit inside a Parallel or an If, which is worse than
-            # not having it.
-            _step_cache = getattr(self, "_step_cache", None)
-            _cache_key = None
-            if _step_cache is not None:
-                from .step_cache import make_step_key
-                _cache_key = make_step_key(step, previous_output, input, all_variables)
-                _cached = _step_cache.get(_cache_key)
-                if _cached is not None:
-                    if verbose:
-                        print(f"↩︎  cache hit: {step.name}")
-                    previous_output = _cached.get("output")
-                    results.append({"step": step.name, "output": previous_output})
-                    if _cached.get("variables"):
-                        all_variables.update(_cached["variables"])
-                    i += 1
-                    continue
-            
-            # Update step status
-            if hasattr(step, 'status'):
-                step.status = "running"
-            self.step_statuses[step.name] = "running"
-            
-            # Call on_step_start callback
-            if self.on_step_start:
-                try:
-                    self.on_step_start(step.name, context)
-                except Exception as e:
-                    logger.error(f"on_step_start callback failed: {e}")
-            
             # Check should_run condition
             if step.should_run:
                 try:
@@ -1660,7 +1626,45 @@ class AgentFlow:
                         continue
                 except Exception as e:
                     logger.error(f"should_run failed for {step.name}: {e}")
+
+            # Cache only the result, never the decision to run. A condition can
+            # observe external state that is absent from the cache key, so it
+            # must be evaluated even when identical inputs have a cached result.
+            _step_cache = getattr(self, "_step_cache", None)
+            _cache_key = None
+            if _step_cache is not None:
+                from .step_cache import make_step_key
+                _cache_key = make_step_key(step, previous_output, input, all_variables)
+                _cached = _step_cache.get(_cache_key)
+                if _cached is not None:
+                    if verbose:
+                        print(f"↩︎  cache hit: {step.name}")
+                    previous_output = _cached.get("output")
+                    # A cache hit is a successful (re)use of the step's result,
+                    # so record it as completed. Without this, a step that was
+                    # skipped on an earlier run keeps its stale "skipped" status
+                    # even though the gate reopened and a cached result was
+                    # served. Execution callbacks stay disabled: nothing ran.
+                    if hasattr(step, 'status'):
+                        step.status = "completed"
+                    self.step_statuses[step.name] = "completed"
+                    results.append({"step": step.name, "output": previous_output})
+                    if _cached.get("variables"):
+                        all_variables.update(_cached["variables"])
+                    i += 1
+                    continue
             
+            # Mark actual execution only after the condition and cache lookup.
+            if hasattr(step, 'status'):
+                step.status = "running"
+            self.step_statuses[step.name] = "running"
+
+            if self.on_step_start:
+                try:
+                    self.on_step_start(step.name, context)
+                except Exception as e:
+                    logger.error(f"on_step_start callback failed: {e}")
+
             # Gap 3c: Check for cross-step handoff cycles
             self._check_handoff_cycle(step)
             

src/praisonai-agents/tests/integration/test_workflow_cache_condition_real.py

@@ -0,0 +1,38 @@
+"""Opt-in real-agent validation of cached conditional workflow execution."""
+
+import os
+
+import pytest
+
+
+@pytest.mark.skipif(
+    os.getenv("RUN_REAL_KEY_TESTS") != "1" or not os.getenv("OPENAI_API_KEY"),
+    reason="requires an explicitly enabled real test provider",
+)
+def test_real_agent_cached_result_respects_current_condition():
+    from praisonaiagents import Agent, Task
+    from praisonaiagents.workflows.workflows import AgentFlow
+
+    enabled = {"value": True}
+    calls = []
+    agent = Agent(
+        name="conditional", instructions="Answer briefly.",
+        llm=os.getenv("PRAISONAI_TEST_MODEL", "gpt-4o-mini"),
+    )
+
+    def handler(ctx):
+        calls.append("run")
+        response = agent.start("Say hello in one short sentence.")
+        print(response)
+        assert response
+        return response
+
+    flow = AgentFlow(steps=[Task(
+        name="conditional", handler=handler,
+        should_run=lambda ctx: enabled["value"],
+    )], cache=True)
+    flow.run("same", verbose=False)
+    flow.run("same", verbose=False)
+    enabled["value"] = False
+    assert flow.run("same", verbose=False)["steps"] == []
+    assert calls == ["run"]

src/praisonai-agents/tests/test_file_memory.py

@@ -119,54 +119,6 @@ class TestShortTermMemory:
             assert items[0].content == "Persistent memory"
 
 
-class TestAtomicWriteCleanup:
-    """Tests for atomic write staging-file cleanup (regression for #5491)."""
-
-    def test_serialization_error_leaves_no_temp_file(self):
-        """A non-serializable value must not leave a .*.tmp staging file behind."""
-        from pathlib import Path
-        from praisonaiagents.memory.file_memory import FileMemory
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            memory = FileMemory(base_path=tmpdir)
-            target = memory.short_term_file
-
-            with pytest.raises(TypeError):
-                memory._write_json(target, {"bad": object()})
-
-            leftovers = list(Path(target.parent).glob(f".{target.name}.*.tmp"))
-            assert leftovers == [], f"leftover temp files: {leftovers}"
-
-    def test_serialization_error_preserves_existing_file(self):
-        """A failed write must not clobber the previously persisted file."""
-        from praisonaiagents.memory.file_memory import FileMemory
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            memory = FileMemory(base_path=tmpdir)
-            target = memory.short_term_file
-
-            assert memory._write_json(target, [{"content": "keep me"}]) is True
-            original = target.read_text(encoding="utf-8")
-
-            with pytest.raises(TypeError):
-                memory._write_json(target, {"bad": object()})
-
-            assert target.read_text(encoding="utf-8") == original
-
-    def test_successful_write_removes_temp_file(self):
-        """A successful atomic write must leave no staging file behind."""
-        from pathlib import Path
-        from praisonaiagents.memory.file_memory import FileMemory
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            memory = FileMemory(base_path=tmpdir)
-            target = memory.short_term_file
-
-            assert memory._write_json(target, [{"content": "ok"}]) is True
-            leftovers = list(Path(target.parent).glob(f".{target.name}.*.tmp"))
-            assert leftovers == [], f"leftover temp files: {leftovers}"
-
-
 class TestLongTermMemory:
     """Tests for long-term memory functionality."""
     

src/praisonai-agents/tests/test_guardrail_blocked_message.py

@@ -1,140 +0,0 @@
-"""Regression tests for issue #5596.
-
-A guardrail that blocks model output must never make Agent.chat/achat return
-``None`` (callers expect a ``str`` and crash on ``.strip()``). On a block the
-agent returns a safe, non-empty fallback string instead of leaking the rejected
-text.
-"""
-
-from unittest.mock import MagicMock
-
-import pytest
-
-from praisonaiagents import Agent, TaskOutput
-
-
-def _blocking_guardrail(task_output: TaskOutput):
-    """Reject every output, mimicking a forbidden-token guardrail."""
-    return False, "contains forbidden token"
-
-
-def test_guardrail_blocked_message_is_safe_non_empty_string():
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-    )
-    message = agent._guardrail_blocked_message("contains forbidden token")
-    assert isinstance(message, str)
-    assert message.strip()
-    # Fixed safe string (FR-001): the guardrail reason must NOT leak to the
-    # caller, since an LLM validator's reason can echo the rejected output.
-    assert "contains forbidden token" not in message
-
-
-def test_guardrail_blocked_message_without_reason():
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-    )
-    message = agent._guardrail_blocked_message()
-    assert isinstance(message, str)
-    assert message.strip()
-
-
-def test_chat_returns_safe_message_not_none_on_block():
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-    )
-    # retries=0 so the retry loop raises immediately without a network call.
-    agent.max_guardrail_retries = 0
-    # Force the custom-LLM branch and feed it forbidden output offline.
-    agent._using_custom_llm = True
-    agent.llm_instance = MagicMock()
-    agent.llm_instance.get_response.return_value = "LEAKME secret"
-
-    result = agent.chat("say something")
-
-    assert result is not None
-    assert isinstance(result, str)
-    assert result.strip()
-    assert "LEAKME" not in result
-
-
-@pytest.mark.asyncio
-async def test_achat_returns_safe_message_not_none_on_block():
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-    )
-    agent.max_guardrail_retries = 0
-
-    async def _fake_get_response_async(**kwargs):
-        return "LEAKME secret"
-
-    agent._using_custom_llm = True
-    agent.llm_instance = MagicMock()
-    agent.llm_instance.get_response_async = _fake_get_response_async
-
-    result = await agent.achat("say something")
-
-    assert result is not None
-    assert isinstance(result, str)
-    assert result.strip()
-    assert "LEAKME" not in result
-
-
-def test_chat_does_not_leak_guardrail_reason():
-    """The fixed fallback must not echo the guardrail's rejection reason.
-
-    An LLM validator's reason can quote the blocked output, so the reason is
-    logged (and exposed via ``last_guardrail_error``) but never returned.
-    """
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-    )
-    agent.max_guardrail_retries = 0
-    agent._using_custom_llm = True
-    agent.llm_instance = MagicMock()
-    agent.llm_instance.get_response.return_value = "LEAKME secret"
-
-    result = agent.chat("say something")
-
-    assert isinstance(result, str) and result.strip()
-    assert "LEAKME" not in result
-    assert "contains forbidden token" not in result
-
-
-def test_chat_returns_safe_message_with_reflection_enabled():
-    """A guardrail block with self-reflection on still yields a safe string."""
-    from praisonaiagents import ReflectionConfig
-
-    agent = Agent(
-        name="GuardAgent",
-        instructions="You are a test agent",
-        llm="gpt-4o-mini",
-        guardrails=_blocking_guardrail,
-        reflection=ReflectionConfig(min_iterations=1, max_iterations=1),
-    )
-    agent.max_guardrail_retries = 0
-    agent._using_custom_llm = True
-    agent.llm_instance = MagicMock()
-    agent.llm_instance.get_response.return_value = "LEAKME secret"
-
-    result = agent.chat("say something")
-
-    assert result is not None
-    assert isinstance(result, str)
-    assert result.strip()
-    assert "LEAKME" not in result

src/praisonai-agents/tests/test_memory_cross_session.py

@@ -1,136 +0,0 @@
-"""
-Regression tests for cross-session memory recall (issue #5595).
-
-Two Agent instances sharing the same memory store (e.g. the same user_id)
-must be able to recall facts written during an earlier turn. Previously the
-synchronous chat/run/start path never wrote the interaction to the memory
-store, so the second agent recalled nothing.
-
-These tests drive the after-agent side-effect pipeline directly so they do
-not require an LLM or OPENAI_API_KEY.
-"""
-
-import asyncio
-import os
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from praisonaiagents import Agent
-from praisonaiagents.memory import FileMemory
-
-
-class TestCrossSessionMemoryRecall(unittest.TestCase):
-    """MEM-001/002/003 from issue #5595."""
-
-    def _agent(self, user_id, base_path):
-        memory = FileMemory(user_id=user_id, base_path=base_path)
-        return Agent(
-            name="Assistant",
-            instructions="You are a helpful assistant.",
-            memory=memory,
-        )
-
-    def test_turn_persisted_and_recalled_across_instances(self):
-        """MEM-001: a fact stored in session A is recalled by session B."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            base = f"{tmpdir}/memory"
-
-            # Session A: simulate a completed turn storing a codename.
-            agent_a = self._agent("e2e-memory-test", base)
-            agent_a._after_agent_side_effects(
-                "My codename is ORANGE-PANDA. ACK.",
-                "Acknowledged. Your codename is ORANGE-PANDA.",
-            )
-
-            # Session B: a fresh agent sharing the same user_id must recall it.
-            agent_b = self._agent("e2e-memory-test", base)
-            context = agent_b.get_memory_context(query="What is my codename?")
-            self.assertIn("ORANGE-PANDA", context)
-
-    def test_different_user_id_is_isolated(self):
-        """MEM-002: a different user_id must not see another user's memory."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            base = f"{tmpdir}/memory"
-
-            agent_a = self._agent("user-a", base)
-            agent_a._after_agent_side_effects(
-                "My codename is ORANGE-PANDA.",
-                "Acknowledged. Your codename is ORANGE-PANDA.",
-            )
-
-            agent_b = self._agent("user-b", base)
-            context = agent_b.get_memory_context(query="What is my codename?")
-            self.assertNotIn("ORANGE-PANDA", context)
-
-    def test_empty_response_is_not_persisted(self):
-        """A turn with no response must not write an empty entry."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            base = f"{tmpdir}/memory"
-            agent = self._agent("user-empty", base)
-            agent._after_agent_side_effects("hello", "")
-            self.assertEqual(
-                agent._memory_instance.get_stats()["short_term_count"], 0
-            )
-
-    def test_dict_memory_config_persists_and_recalls(self):
-        """MEM-004: the advertised ``memory={'user_id': uid}`` dict config —
-        resolved by the Agent itself, not a preconstructed FileMemory — persists
-        a turn that a second Agent with the same user_id recalls.
-
-        Guards the configuration-resolution + hook wiring that the direct-call
-        tests above cannot (Greptile #3). Runs in a temp cwd so the default
-        FileMemory base path is hermetic; no LLM/API key needed.
-        """
-        cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            os.chdir(tmpdir)
-            try:
-                uid = "dict-config-user"
-                agent_a = Agent(name="A", instructions="x", memory={"user_id": uid})
-                self.assertIsInstance(agent_a._memory_instance, FileMemory)
-                agent_a._after_agent_side_effects(
-                    "My codename is BLUE-FOX.",
-                    "Acknowledged. Your codename is BLUE-FOX.",
-                )
-
-                agent_b = Agent(name="B", instructions="x", memory={"user_id": uid})
-                context = agent_b.get_memory_context(query="What is my codename?")
-                self.assertIn("BLUE-FOX", context)
-            finally:
-                os.chdir(cwd)
-
-    def test_async_path_persists_without_blocking_loop(self):
-        """MEM-005: on the async after-agent path the blocking store write is
-        offloaded to a thread (AGENTS.md §4.5), yet the turn is still persisted
-        and recalled by a later Agent. Exercises the running-loop branch of
-        ``_persist_memory_turn``.
-        """
-        with tempfile.TemporaryDirectory() as tmpdir:
-            base = f"{tmpdir}/memory"
-
-            async def write_turn():
-                agent_a = self._agent("async-user", base)
-                agent_a._after_agent_side_effects(
-                    "My codename is GREEN-HAWK.",
-                    "Acknowledged. Your codename is GREEN-HAWK.",
-                )
-                # The write is scheduled via run_in_executor; yield until the
-                # short-term store reflects it (bounded) rather than racing it.
-                for _ in range(500):
-                    if agent_a._memory_instance.get_stats()["short_term_count"] > 0:
-                        break
-                    await asyncio.sleep(0.01)
-
-            asyncio.run(write_turn())
-
-            agent_b = self._agent("async-user", base)
-            context = agent_b.get_memory_context(query="What is my codename?")
-            self.assertIn("GREEN-HAWK", context)
-
-
-if __name__ == "__main__":
-    unittest.main()

src/praisonai-agents/tests/unit/agent/test_guardrail_retry_feedback.py

@@ -270,11 +270,9 @@ class TestAlwaysFailingValidatorIsBounded:
         except _CallCap as exc:
             pytest.fail(str(exc))
 
-        # chat() reports guardrail exhaustion with a safe, non-empty fallback
-        # string after rolling back history - never None (issue #5596) and
-        # never the rejected answer. The reason is also on the agent to read.
-        assert isinstance(result, str) and result.strip()
-        assert "10001" not in result
+        # chat() reports guardrail exhaustion by returning None after rolling
+        # back history; the reason is on the agent for the caller to read.
+        assert result is None
         assert agent.guardrail_retry_count == 2, \
             f"must retry exactly max_guardrail_retries times, got {agent.guardrail_retry_count}"
         assert t.calls == 1 + 2, f"1 initial call + 2 retries, got {t.calls}"
@@ -302,8 +300,7 @@ class TestAlwaysFailingValidatorIsBounded:
     def test_max_retries_zero_never_retries(self):
         agent, t = _agent(["10001", "SW1A 1AA"], rejects_first_answer, max_retries=0)
         result = agent.chat("what is the postcode?", stream=False)
-        assert isinstance(result, str) and result.strip()
-        assert "10001" not in result
+        assert result is None
         assert t.calls == 1, f"max_guardrail_retries=0 must not call the model again (got {t.calls})"
         assert agent.guardrail_retry_count == 0
 

src/praisonai-agents/tests/unit/agents/test_chat_output_pydantic_coercion.py

@@ -1,116 +0,0 @@
-"""Unit tests for Agent.chat structured-output coercion.
-
-Covers the fix for issue #5594: ``Agent.chat(output_pydantic=Model)`` must
-return a validated Pydantic instance (not a raw JSON string) on the live
-OpenAI/LiteLLM path. These tests exercise the ``_coerce_structured_output``
-helper directly so no live LLM is required.
-"""
-from types import SimpleNamespace
-
-from pydantic import BaseModel
-
-from praisonaiagents.agent.chat_mixin import ChatMixin
-
-
-class Pair(BaseModel):
-    city: str
-    country: str
-
-
-def _fake_completion(content):
-    """Build a minimal OpenAI-style response object for mocking."""
-    message = SimpleNamespace(content=content, reasoning_content=None, tool_calls=None)
-    return SimpleNamespace(choices=[SimpleNamespace(message=message)])
-
-
-def _mixin():
-    m = ChatMixin()
-    m.name = "TestAgent"
-    return m
-
-
-def test_valid_json_string_returns_model():
-    out = _mixin()._coerce_structured_output(
-        '{"city": "Paris", "country": "France"}', Pair
-    )
-    assert isinstance(out, Pair)
-    assert out.city == "Paris"
-    assert out.country == "France"
-
-
-def test_markdown_fenced_json_returns_model():
-    fenced = '```json\n{"city": "Paris", "country": "France"}\n```'
-    out = _mixin()._coerce_structured_output(fenced, Pair)
-    assert isinstance(out, Pair)
-    assert out.city == "Paris"
-
-
-def test_prose_falls_back_to_raw_string():
-    out = _mixin()._coerce_structured_output(
-        "The capital of France is Paris.", Pair
-    )
-    assert isinstance(out, str)
-    assert out == "The capital of France is Paris."
-
-
-def test_no_schema_returns_input_unchanged():
-    out = _mixin()._coerce_structured_output("hello", None)
-    assert out == "hello"
-
-
-def test_non_string_input_returns_unchanged():
-    instance = Pair(city="Paris", country="France")
-    out = _mixin()._coerce_structured_output(instance, Pair)
-    assert out is instance
-
-
-def test_dict_schema_without_model_validate_json_returns_raw():
-    # dict schemas (inline JSON schema) have no model_validate_json; must not crash
-    out = _mixin()._coerce_structured_output('{"a": 1}', {"type": "object"})
-    assert out == '{"a": 1}'
-
-
-# --- Public-path tests: chat()/achat() wiring and backward-compat guards ---
-
-
-def _agent(monkeypatch, content):
-    """Build a real Agent whose LLM call is mocked to return ``content``.
-
-    A placeholder API key lets the lazy OpenAI client construct; the network is
-    never touched because ``_chat_completion`` is mocked.
-    """
-    monkeypatch.setenv("OPENAI_API_KEY", "not-needed")
-    from praisonaiagents import Agent
-
-    agent = Agent(name="TestAgent", instructions="test")
-    monkeypatch.setattr(agent, "_chat_completion", lambda *a, **k: _fake_completion(content))
-    return agent
-
-
-def test_chat_output_pydantic_returns_model(monkeypatch):
-    # Public chat() path must return a validated Pydantic instance (issue #5594).
-    agent = _agent(monkeypatch, '{"city": "Paris", "country": "France"}')
-    out = agent.chat("capital of France?", output_pydantic=Pair)
-    assert isinstance(out, Pair)
-    assert out.city == "Paris"
-
-
-def test_chat_output_json_stays_string(monkeypatch):
-    # Backward compat: output_json must keep returning a raw JSON string so that
-    # downstream task/workflow JSON parsing (json.loads) keeps working.
-    agent = _agent(monkeypatch, '{"city": "Paris", "country": "France"}')
-    out = agent.chat("capital of France?", output_json=Pair)
-    assert isinstance(out, str)
-    assert '"Paris"' in out
-
-
-def test_after_agent_hook_receives_string_for_pydantic(monkeypatch):
-    # A Pydantic response must not crash command/string hooks: the after-agent
-    # hook payload truncates via response[:500], which requires a string.
-    agent = _agent(monkeypatch, '{"city": "Paris", "country": "France"}')
-    payload = agent._build_after_agent_input(
-        prompt="p", response=Pair(city="Paris", country="France"), start_time=0.0
-    )
-    assert isinstance(payload.response, str)
-    # to_dict() performs the [:500] slice that would otherwise raise on a model
-    assert isinstance(payload.to_dict()["response"], str)

src/praisonai-agents/tests/unit/memory/test_memory_thread_cleanup.py

@@ -1,164 +0,0 @@
-"""Thread-scoped SQLite cleanup must preserve the shared adapter (#5486).
-
-`Memory.close_connections()` is documented as thread-scoped: it closes only the
-calling thread's SQLite connections and leaves the shared `SqliteMemoryAdapter`
-(and other threads' connections) intact. A previous adapter-shutdown block
-cleared `self.memory_adapter` unconditionally, so subsequent searches fell back
-to legacy SQL against `short_mem` / `long_mem` and raised
-``sqlite3.OperationalError: no such table``.
-
-These tests pin the contract: after another thread (or the same thread) calls
-`close_connections()`, the shared adapter survives and searches keep working by
-lazily reopening the calling thread's connection.
-"""
-from concurrent.futures import ThreadPoolExecutor
-
-import pytest
-
-from praisonaiagents.memory.memory import Memory
-from praisonaiagents.memory.adapters.sqlite_adapter import SqliteMemoryAdapter
-
-
-@pytest.fixture
-def store(tmp_path, monkeypatch):
-    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
-    monkeypatch.chdir(tmp_path)
-    return Memory(config={"provider": "sqlite"})
-
-
-def _texts(results):
-    items = results if isinstance(results, list) else results.get("results", [])
-    return [r.get("text") or r.get("memory") or "" for r in items]
-
-
-def test_adapter_is_sqlite(store):
-    """Control: these tests only mean something for the SQLite adapter."""
-    assert isinstance(store.memory_adapter, SqliteMemoryAdapter)
-
-
-def test_search_before_cleanup(store):
-    """Control: searching works before any cleanup happens."""
-    store.store_short_term("shared record")
-    assert "shared record" in _texts(store.search_short_term("shared record"))
-
-
-def test_short_term_search_survives_idle_worker_cleanup(store):
-    """An idle worker that never opened a connection must not break searches."""
-    store.store_short_term("shared record")
-    with ThreadPoolExecutor(max_workers=1) as pool:
-        pool.submit(store.close_connections).result()
-
-    assert store.memory_adapter is not None
-    assert "shared record" in _texts(store.search_short_term("shared record"))
-
-
-def test_long_term_search_survives_idle_worker_cleanup(store):
-    store.store_long_term("shared long record")
-    with ThreadPoolExecutor(max_workers=1) as pool:
-        pool.submit(store.close_connections).result()
-
-    assert store.memory_adapter is not None
-    assert "shared long record" in _texts(store.search_long_term("shared long record"))
-
-
-def test_short_term_search_survives_initialized_worker_cleanup(store):
-    """A worker that has opened its own connection cleans up only its own."""
-    store.store_short_term("shared record")
-
-    def worker():
-        store.store_short_term("worker record")
-        store.close_connections()
-
-    with ThreadPoolExecutor(max_workers=1) as pool:
-        pool.submit(worker).result()
-
-    assert store.memory_adapter is not None
-    assert "shared record" in _texts(store.search_short_term("shared record"))
-    assert "worker record" in _texts(store.search_short_term("worker record"))
-
-
-def test_same_thread_repeated_cleanup_and_reopen(store):
-    """Repeated same-thread cleanup must keep the adapter usable (lazy reopen)."""
-    store.store_short_term("shared record")
-    store.close_connections()
-    store.close_connections()
-
-    assert store.memory_adapter is not None
-    assert "shared record" in _texts(store.search_short_term("shared record"))
-
-
-def test_same_thread_long_term_cleanup_and_reopen(store):
-    store.store_long_term("shared long record")
-    store.close_connections()
-
-    assert store.memory_adapter is not None
-    assert "shared long record" in _texts(store.search_long_term("shared long record"))
-
-
-def test_adapter_thread_close_is_idempotent(store):
-    """Adapter-level thread close tolerates being called with no open conns."""
-    store.memory_adapter.close_thread_connections()
-    store.memory_adapter.close_thread_connections()
-    store.store_short_term("shared record")
-    assert "shared record" in _texts(store.search_short_term("shared record"))
-
-
-def test_adapter_thread_close_closes_open_connection(store):
-    """Opening a connection then closing it directly must close the live conn
-    and drop it from the shared registry (not just the no-op path)."""
-    adapter = store.memory_adapter
-    adapter.store_short_term("shared record")
-    adapter.store_long_term("shared long record")
-
-    stm = adapter._local.stm_conn
-    ltm = adapter._local.ltm_conn
-    assert stm in adapter._all_connections
-    assert ltm in adapter._all_connections
-
-    adapter.close_thread_connections()
-
-    # Thread-local refs cleared and registry no longer tracks the closed conns.
-    assert getattr(adapter._local, "stm_conn", None) is None
-    assert getattr(adapter._local, "ltm_conn", None) is None
-    assert stm not in adapter._all_connections
-    assert ltm not in adapter._all_connections
-
-    # The physical connections are closed (operations raise ProgrammingError).
-    with pytest.raises(Exception):
-        stm.execute("SELECT 1")
-    with pytest.raises(Exception):
-        ltm.execute("SELECT 1")
-
-    # Adapter remains usable via lazy reopen.
-    assert "shared record" in _texts(adapter.search_short_term("shared record"))
-
-
-def test_adapter_thread_close_keeps_connection_tracked_on_failure(store):
-    """If a connection fails to close, it stays tracked for a later retry and
-    the thread-local ref is preserved (the open-conn failure path)."""
-    adapter = store.memory_adapter
-    adapter.store_short_term("shared record")
-    real = adapter._local.stm_conn
-
-    class _FlakyConn:
-        def __init__(self, inner):
-            self._inner = inner
-            self.close_calls = 0
-
-        def close(self):
-            self.close_calls += 1
-            raise RuntimeError("boom")
-
-    flaky = _FlakyConn(real)
-    # Swap the tracked real conn for a flaky wrapper in both registry and tls.
-    with adapter._connection_lock:
-        adapter._all_connections.discard(real)
-        adapter._all_connections.add(flaky)
-    adapter._local.stm_conn = flaky
-
-    adapter.close_thread_connections()
-
-    assert flaky.close_calls == 1
-    # Failed close: connection stays tracked and the ref is NOT cleared.
-    assert flaky in adapter._all_connections
-    assert adapter._local.stm_conn is flaky

src/praisonai-agents/tests/unit/session/test_session_agent_history_fields.py

@@ -1,123 +0,0 @@
-"""Closing and reopening a Session must preserve complete agent messages."""
-
-from copy import deepcopy
-from types import SimpleNamespace
-
-import pytest
-
-from praisonaiagents.session.api import AGENT_HISTORY_KEY, Session
-from praisonaiagents.session.store import DefaultSessionStore
-
-
-@pytest.mark.parametrize("history", [
-    [{"role": "user", "content": "hello"}],
-    [
-        {"role": "assistant", "content": None, "tool_calls": [
-            {"id": "call_1", "type": "function", "function": {
-                "name": "lookup", "arguments": '{"query":"hello"}'}}
-        ]},
-        {"role": "tool", "content": "found", "tool_call_id": "call_1",
-         "name": "lookup"},
-        {"role": "assistant", "content": "answer"},
-    ],
-    [{"role": "assistant", "content": "answer", "reasoning_content": "why",
-      "metadata": {"citations": [{"source": "local"}]}, "custom": False}],
-    [{"role": "user", "content": [
-        {"type": "text", "text": "describe"},
-        {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}},
-    ], "name": "reader"}],
-], ids=["plain-control", "tool-turn", "reasoning-and-metadata", "multimodal"])
-def test_close_reopen_preserves_message_fields(tmp_path, monkeypatch, history):
-    import praisonaiagents.agent as agent_module
-    import praisonaiagents.session.store as store_module
-
-    class LocalAgent:
-        def __init__(self, **kwargs):
-            self.chat_history = []
-
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
-    monkeypatch.setattr(agent_module, "Agent", LocalAgent)
-    session_dir = str(tmp_path / "sessions")
-    store = DefaultSessionStore(session_dir=session_dir)
-    monkeypatch.setattr(store_module, "_default_store", store)
-    session = Session(session_id="parent")
-    session._memory = SimpleNamespace(search_short_term=lambda **kwargs: [])
-    agent = session.Agent(name="helper", memory=False)
-    agent.chat_history = deepcopy(history)
-    session.close()
-
-    fresh = DefaultSessionStore(session_dir=session_dir)
-    monkeypatch.setattr(store_module, "_default_store", fresh)
-    assert fresh.get_session("parent").metadata[AGENT_HISTORY_KEY][
-        "helper:Assistant"
-    ] == history
-    resumed = Session(session_id="parent").Agent(name="helper", memory=False)
-    assert resumed.chat_history == history
-
-
-@pytest.mark.parametrize("builder", ["openai", "litellm"])
-def test_restored_extensions_do_not_reach_responses_input(tmp_path, monkeypatch, builder):
-    import praisonaiagents.session.store as store_module
-    from praisonaiagents.llm.openai_client import OpenAIClient
-    from praisonaiagents.llm.llm import LLM
-
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
-    store = DefaultSessionStore(session_dir=str(tmp_path / "sessions"))
-    monkeypatch.setattr(store_module, "_default_store", store)
-    history = [
-        {"role": "user", "content": "question", "name": "reader",
-         "metadata": {"origin": "disk"}},
-        {"role": "assistant", "content": "answer", "reasoning_content": "why",
-         "metadata": {"citation": "local"}, "extension": False},
-        {"role": "assistant", "content": None, "tool_calls": [{
-            "id": "call_1", "type": "function",
-            "function": {"name": "lookup", "arguments": "{}"},
-        }]},
-        {"role": "tool", "tool_call_id": "call_1", "content": "found"},
-    ]
-    session = Session(session_id="parent")
-    session._agents["helper:Assistant"] = {"agent": None, "chat_history": history}
-    session.close()
-    resumed = Session(session_id="parent")._restore_agent_chat_history("helper:Assistant")
-    if builder == "openai":
-        client = object.__new__(OpenAIClient)
-        params = client._build_responses_input(resumed, model="gpt-4o")
-    else:
-        llm = object.__new__(LLM)
-        llm.max_tokens = None
-        llm.top_p = None
-        llm.base_url = None
-        llm.api_key = None
-        llm.timeout = None
-        llm._resolve_subscription_creds = lambda: None
-        llm._resolve_openai_compatible_model = lambda: "gpt-4o"
-        params = llm._build_responses_params(resumed)
-    assert params["input"] == [
-        {"role": "user", "content": "question"},
-        {"role": "assistant", "content": "answer"},
-        {"type": "function_call", "call_id": "call_1", "name": "lookup",
-         "arguments": "{}"},
-        {"type": "function_call_output", "call_id": "call_1", "output": "found"},
-    ]
-    assert store.get_session("parent").metadata[AGENT_HISTORY_KEY][
-        "helper:Assistant"
-    ] == history
-
-
-def test_legacy_defaults_and_non_message_filter(tmp_path, monkeypatch):
-    import praisonaiagents.session.store as store_module
-
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
-    store = DefaultSessionStore(session_dir=str(tmp_path / "sessions"))
-    monkeypatch.setattr(store_module, "_default_store", store)
-    session = Session(session_id="parent")
-    session._agents["helper:Assistant"] = {
-        "agent": None, "chat_history": [None, {"name": "helper"}],
-    }
-    session.close()
-    assert store.get_session("parent").metadata[AGENT_HISTORY_KEY][
-        "helper:Assistant"
-    ] == [{"role": "user", "content": "", "name": "helper"}]

src/praisonai-agents/tests/unit/session/test_session_atomic_agent_history.py

@@ -1,91 +0,0 @@
-"""Concurrent Session closes must merge different agent histories under lock."""
-
-from concurrent.futures import ThreadPoolExecutor
-from threading import Event
-from pathlib import Path
-
-import pytest
-
-from praisonaiagents.session.api import AGENT_HISTORY_KEY, Session
-from praisonaiagents.session.store import DefaultSessionStore
-
-
-@pytest.mark.parametrize("existing", [{}, {"peer:Assistant": [
-    {"role": "user", "content": "peer"},
-]}])
-def test_two_closes_preserve_both_agent_histories(tmp_path, monkeypatch, existing):
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
-    directory = str(tmp_path / "sessions")
-    first = DefaultSessionStore(session_dir=directory)
-    second = DefaultSessionStore(session_dir=directory)
-    first.update_session_metadata("parent", **{AGENT_HISTORY_KEY: existing})
-    a, b = Session(session_id="parent"), Session(session_id="parent")
-    monkeypatch.setattr(a, "_get_session_store", lambda: first)
-    monkeypatch.setattr(b, "_get_session_store", lambda: second)
-    ah = [{"role": "user", "content": "a"}]
-    bh = [{"role": "user", "content": "b"}]
-    a._agents["a:Assistant"] = {"agent": None, "chat_history": ah}
-    b._agents["b:Assistant"] = {"agent": None, "chat_history": bh}
-    ready, completed = Event(), Event()
-    original = first._modify_session_locked
-
-    def pause_before_lock(*args, **kwargs):
-        ready.set()
-        assert completed.wait(10), "second writer did not complete"
-        return original(*args, **kwargs)
-
-    monkeypatch.setattr(first, "_modify_session_locked", pause_before_lock)
-
-    def close_second():
-        assert ready.wait(10), "first writer did not reach write boundary"
-        try:
-            b.close()
-        finally:
-            completed.set()
-
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        fa, fb = pool.submit(a.close), pool.submit(close_second)
-        fa.result(timeout=15)
-        fb.result(timeout=15)
-    fresh = DefaultSessionStore(session_dir=directory)
-    assert fresh.get_session("parent").metadata[AGENT_HISTORY_KEY] == {
-        **existing, "a:Assistant": ah, "b:Assistant": bh,
-    }
-
-
-def test_metadata_map_defaults_do_not_overwrite_current_entries(tmp_path):
-    store = DefaultSessionStore(session_dir=str(tmp_path))
-    store.update_session_metadata("parent", marker="keep", **{AGENT_HISTORY_KEY: {
-        "current": [], "peer": ["present"],
-    }})
-    assert store.merge_session_metadata_map(
-        "parent", AGENT_HISTORY_KEY, {"new": ["new"]},
-        defaults={"current": ["legacy"], "legacy-only": ["old"]},
-    )
-    data = DefaultSessionStore(session_dir=str(tmp_path)).get_session("parent")
-    assert data.metadata[AGENT_HISTORY_KEY] == {
-        "current": [], "peer": ["present"], "new": ["new"],
-        "legacy-only": ["old"],
-    }
-    assert data.metadata["marker"] == "keep"
-
-
-@pytest.mark.parametrize("failure", ["read", "write"])
-def test_metadata_map_failure_preserves_disk(tmp_path, monkeypatch, failure):
-    import praisonaiagents.session.store as store_module
-
-    store = DefaultSessionStore(session_dir=str(tmp_path))
-    store.update_session_metadata("parent", **{AGENT_HISTORY_KEY: {"peer": ["old"]}})
-    path = Path(store._get_session_path("parent"))
-    before = path.read_bytes()
-
-    def fail(*args, **kwargs):
-        raise OSError("injected I/O failure")
-
-    if failure == "read":
-        monkeypatch.setattr(store, "_load_session_from_disk", fail)
-    else:
-        monkeypatch.setattr(store_module.os, "replace", fail)
-    assert not store.merge_session_metadata_map("parent", AGENT_HISTORY_KEY, {"new": []})
-    assert path.read_bytes() == before

src/praisonai-agents/tests/unit/session/test_session_import_no_clobber.py

@@ -1,52 +0,0 @@
-"""Default imports must not overwrite a session created after the first check."""
-
-from concurrent.futures import ThreadPoolExecutor
-from threading import Event
-
-import pytest
-
-from praisonaiagents.session.store import DefaultSessionStore
-
-
-@pytest.mark.parametrize("overwrite", [False, True])
-def test_import_rechecks_new_peer_session_under_write_lock(tmp_path, monkeypatch, overwrite):
-    directory = str(tmp_path / "sessions")
-    importer = DefaultSessionStore(session_dir=directory)
-    peer = DefaultSessionStore(session_dir=directory)
-    ready, completed = Event(), Event()
-    original = importer._save_imported_session
-
-    def pause_before_write(*args, **kwargs):
-        ready.set()
-        assert completed.wait(10), "peer did not finish creating its session"
-        return original(*args, **kwargs)
-
-    monkeypatch.setattr(importer, "_save_imported_session", pause_before_write)
-    payload = {"version": 1, "sessions": [{
-        "session_id": "shared", "messages": [{
-            "role": "user", "content": "imported", "timestamp": 1,
-        }],
-    }]}
-
-    def create_peer():
-        assert ready.wait(10), "import did not reach write boundary"
-        try:
-            assert peer.add_message("shared", "user", "peer-created")
-        finally:
-            completed.set()
-
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        future = pool.submit(create_peer)
-        report = importer.import_sessions(payload, overwrite=overwrite)
-        future.result(timeout=15)
-    fresh = DefaultSessionStore(session_dir=directory)
-    if overwrite:
-        assert report.imported == 1
-        assert not report.skipped
-        assert fresh.get_chat_history("shared") == [{"role": "user", "content": "imported"}]
-    else:
-        assert report.imported == 0
-        assert report.skipped == [{
-            "session_id": "shared", "reason": "already exists (use overwrite)",
-        }]
-        assert fresh.get_chat_history("shared") == [{"role": "user", "content": "peer-created"}]

src/praisonai-agents/tests/unit/session/test_sqlite_import_index.py

@@ -1,187 +0,0 @@
-"""Portable imports refresh the indexed store's content and routing views."""
-
-import pytest
-import threading
-from concurrent.futures import ThreadPoolExecutor
-
-from praisonaiagents.session.sqlite_store import SqliteSessionStore
-from praisonaiagents.session.store import DefaultSessionStore
-
-
-@pytest.fixture
-def make_store(tmp_path):
-    stores = []
-
-    def create(kind, **kwargs):
-        store = kind(session_dir=str(tmp_path / str(len(stores))), **kwargs)
-        stores.append(store)
-        return store
-
-    yield create
-    for store in stores:
-        conn = getattr(store, "_conn", None)
-        if conn is not None:
-            conn.close()
-
-
-@pytest.mark.parametrize("kind", [DefaultSessionStore, SqliteSessionStore])
-@pytest.mark.parametrize("warm", [False, True])
-@pytest.mark.parametrize("overwrite", [False, True])
-@pytest.mark.parametrize("reset_live_fields", [False, True])
-def test_import_refreshes_content_and_routes(make_store, kind, warm, overwrite, reset_live_fields):
-    source = make_store(DefaultSessionStore)
-    messages = [{"role": "user", "content": f"narwhal imported turn {index}"} for index in range(5)]
-    assert source.set_chat_history("session", messages)
-    assert source.set_gateway_info("session", gateway_session_id="new-gateway", agent_id="new-agent")
-    payload = source.export_all()
-
-    destination = make_store(kind, active_window=2)
-    if overwrite:
-        assert destination.add_message("session", "user", "pelican old turn")
-        assert destination.set_gateway_info("session", gateway_session_id="old-gateway", agent_id="old-agent")
-    if warm:
-        destination.search("pelican")
-        destination.get_by_gateway_session("old-gateway")
-
-    report = destination.import_sessions(payload, overwrite=overwrite, reset_live_fields=reset_live_fields)
-    assert report.imported == 1
-    assert report.skipped == []
-    session = destination.get_session("session")
-    assert [message.content for message in session.messages] == [message["content"] for message in messages]
-    assert destination.get_by_gateway_session("old-gateway") is None
-    assert destination.list_sessions_by_gateway_agent("old-agent") == []
-    found = destination.get_by_gateway_session("new-gateway")
-    if reset_live_fields:
-        assert found is None
-        assert destination.list_sessions_by_gateway_agent("new-agent") == []
-    else:
-        assert found is not None and found.session_id == "session"
-        assert destination.list_sessions_by_gateway_agent("new-agent") == ["session"]
-    assert destination.search("pelican") == []
-    assert [hit.session_id for hit in destination.search("narwhal")] == ["session"]
-
-
-@pytest.mark.parametrize("write_failure", [False, True])
-def test_unsaved_import_keeps_previous_index(make_store, monkeypatch, write_failure):
-    source = make_store(DefaultSessionStore)
-    assert source.add_message("session", "user", "narwhal replacement")
-    destination = make_store(SqliteSessionStore)
-    assert destination.add_message("session", "user", "pelican original")
-    assert destination.set_gateway_info("session", gateway_session_id="gateway", agent_id="agent")
-    assert destination.search("pelican")
-    if write_failure:
-        monkeypatch.setattr(destination, "_atomic_write_json", lambda *args: False)
-
-    report = destination.import_sessions(source.export_all(), overwrite=write_failure)
-    assert report.imported == 0
-    assert len(report.skipped) == 1
-    assert report.skipped[0]["reason"] == ("write failed" if write_failure else "already exists (use overwrite)")
-    assert [hit.session_id for hit in destination.search("pelican")] == ["session"]
-    assert destination.search("narwhal") == []
-    assert destination.get_by_gateway_session("gateway").session_id == "session"
-    assert destination.list_sessions_by_gateway_agent("agent") == ["session"]
-
-
-def test_warm_import_indexes_archived_turns(make_store):
-    source = make_store(DefaultSessionStore, active_window=3)
-    assert source.add_message("session", "user", "pelican archived marker")
-    for index in range(8):
-        assert source.add_message("session", "user", f"later turn {index}")
-    assert any("pelican" in message.content for message in source.get_session("session").archived_messages)
-    destination = make_store(SqliteSessionStore, active_window=2)
-    assert destination.search("pelican") == []
-    assert destination.import_sessions(source.export_all()).imported == 1
-    hits = destination.search("pelican")
-    assert [hit.session_id for hit in hits] == ["session"]
-    assert any(message["archived"] for message in hits[0].messages)
-
-
-@pytest.mark.parametrize("separate_store", [False, True])
-def test_write_between_import_save_and_index_keeps_newer_content(make_store, monkeypatch, separate_store):
-    source = make_store(DefaultSessionStore)
-    assert source.add_message("session", "user", "pelican imported turn")
-    destination = make_store(SqliteSessionStore)
-    destination.search("pelican")
-    writer = destination
-    if separate_store:
-        writer = SqliteSessionStore(session_dir=destination.session_dir, db_path=destination.db_path)
-        writer.search("pelican")
-
-    saved = threading.Event()
-    resume = threading.Event()
-    original = DefaultSessionStore._save_imported_session
-
-    def delayed_save(store, session, **kwargs):
-        result = original(store, session, **kwargs)
-        saved.set()
-        assert resume.wait(5), "concurrent writer did not finish"
-        return result
-
-    monkeypatch.setattr(DefaultSessionStore, "_save_imported_session", delayed_save)
-    try:
-        with ThreadPoolExecutor(max_workers=1) as executor:
-            importing = executor.submit(destination.import_sessions, source.export_all())
-            try:
-                assert saved.wait(5), "import did not persist the JSON"
-                assert writer.set_chat_history("session", [{"role": "user", "content": "narwhal newer turn"}])
-                assert writer.set_gateway_info("session", gateway_session_id="new-gateway", agent_id="new-agent")
-                assert writer.search("narwhal")
-            finally:
-                resume.set()
-            assert importing.result(timeout=5).imported == 1
-        destination.invalidate_cache()
-        assert destination.get_session("session").messages[0].content == "narwhal newer turn"
-        assert [hit.session_id for hit in destination.search("narwhal")] == ["session"]
-        assert destination.search("pelican") == []
-        assert destination.get_by_gateway_session("new-gateway").session_id == "session"
-        assert destination.list_sessions_by_gateway_agent("new-agent") == ["session"]
-    finally:
-        if separate_store and writer._conn is not None:
-            writer._conn.close()
-
-
-def test_post_import_refresh_failure_does_not_report_durable_write_failure(make_store, monkeypatch):
-    source = make_store(DefaultSessionStore)
-    assert source.add_message("session", "user", "durable imported note")
-    destination = make_store(SqliteSessionStore)
-
-    def unavailable(*args):
-        raise OSError("injected index refresh read failure")
-
-    monkeypatch.setattr(destination, "_load_session_from_disk", unavailable)
-    report = destination.import_sessions(source.export_all())
-    assert report.imported == 1
-    assert report.skipped == []
-    reader = DefaultSessionStore(session_dir=destination.session_dir)
-    assert reader.get_session("session").messages[0].content == "durable imported note"
-
-
-def test_post_import_read_failure_does_not_restore_old_routes(make_store, monkeypatch):
-    source = make_store(DefaultSessionStore)
-    assert source.add_message("session", "user", "narwhal replacement")
-    destination = make_store(SqliteSessionStore)
-    assert destination.add_message("session", "user", "pelican original")
-    assert destination.set_gateway_info("session", gateway_session_id="old-gateway", agent_id="old-agent")
-    assert destination.get_by_gateway_session("old-gateway").session_id == "session"
-    assert destination.search("pelican")
-    original = destination._load_session_from_disk
-
-    def unavailable(*args):
-        raise OSError("injected post-save read failure")
-
-    monkeypatch.setattr(destination, "_load_session_from_disk", unavailable)
-    report = destination.import_sessions(source.export_all(), overwrite=True)
-    assert report.imported == 1
-    assert report.skipped == []
-    assert destination.get_by_gateway_session("old-gateway") is None
-    assert destination.list_sessions_by_gateway_agent("old-agent") == []
-    monkeypatch.setattr(destination, "_load_session_from_disk", original)
-    destination.invalidate_cache()
-    saved = destination.get_session("session")
-    assert saved.gateway_session_id is None
-    assert saved.agent_id is None
-    assert saved.messages[0].content == "narwhal replacement"
-    assert destination.search("narwhal") == []
-    assert destination.add_message("session", "user", "later update")
-    assert [hit.session_id for hit in destination.search("narwhal")] == ["session"]
-    assert destination.get_by_gateway_session("old-gateway") is None

src/praisonai-agents/tests/unit/session/test_sqlite_transcript_portability.py

@@ -1,179 +0,0 @@
-"""Portable session operations use the SQLite transcript persistence backend."""
-
-import json
-import tracemalloc
-
-import pytest
-
-from praisonaiagents.session.store import DefaultSessionStore
-from praisonaiagents.session.sqlite_transcript_store import SqliteTranscriptStore
-
-
-@pytest.fixture
-def stores(tmp_path):
-    opened = []
-
-    def make(name, backend=SqliteTranscriptStore, **kwargs):
-        store = backend(session_dir=str(tmp_path / name), **kwargs)
-        opened.append(store)
-        return store
-
-    yield make
-    for store in opened:
-        conn = getattr(store, "_conn", None)
-        if conn is not None:
-            conn.close()
-
-
-def test_export_all_reads_database_not_json_sidecars(stores):
-    store = stores("source")
-    assert store.add_message("one", "user", "First")
-    assert store.add_message("two", "assistant", "Second")
-    from pathlib import Path
-    (Path(store.session_dir) / "unrelated.json").write_text(
-        json.dumps({"session_id": "sidecar", "messages": []}), encoding="utf-8"
-    )
-    payload = store.export_all()
-    assert payload["version"] == store.PORTABLE_VERSION
-    assert {record["session_id"] for record in payload["sessions"]} == {"one", "two"}
-
-
-@pytest.mark.parametrize("key", ["lineage_id", "root_session_id", "thread_id"])
-def test_single_export_includes_database_lineage(stores, key):
-    store = stores("source")
-    for session_id in ("first", "continuation", "unrelated"):
-        assert store.add_message(session_id, "user", session_id)
-        assert store.update_session_metadata(session_id, **{key: "shared" if session_id != "unrelated" else "other"})
-    payload = store.export_session("continuation")
-    assert {record["session_id"] for record in payload["sessions"]} == {"first", "continuation"}
-    assert [record["session_id"] for record in store.export_session("continuation", include_lineage=False)["sessions"]] == ["continuation"]
-
-
-@pytest.mark.parametrize("source_backend", [DefaultSessionStore, SqliteTranscriptStore])
-@pytest.mark.parametrize("target_backend", [DefaultSessionStore, SqliteTranscriptStore])
-def test_cross_backend_restore_survives_reopen_without_window_truncation(stores, source_backend, target_backend):
-    source = stores("source", source_backend)
-    for index in range(6):
-        assert source.add_message("reference", "user", f"message-{index}")
-    assert source.set_gateway_info("reference", gateway_session_id="live", agent_id="agent")
-    payload = source.export_all()
-    assert len(payload["sessions"]) == 1
-    target = stores("target", target_backend, active_window=2, retention="truncate")
-    # An existing SQLite row prevents legacy JSON migration from masking a
-    # restore incorrectly written to a sidecar file instead of the database.
-    assert target.add_message("existing", "user", "Existing history")
-    assert target.import_sessions(payload).imported == 1
-    assert target.session_exists("reference")
-    target.invalidate_cache()
-    reopened = stores("target", target_backend, active_window=2, retention="truncate")
-    for current in (target, reopened):
-        assert [message["content"] for message in current.get_chat_history("reference")] == [f"message-{index}" for index in range(6)]
-        session = current.get_session("reference")
-        assert session.gateway_session_id is None
-        assert session.agent_id is None
-    if target_backend is SqliteTranscriptStore:
-        from pathlib import Path
-        assert list(Path(target.session_dir).glob("*.json")) == []
-
-
-def test_import_sqlite_write_failure_is_reported_and_preserves_existing_row(stores):
-    source = stores("source", DefaultSessionStore)
-    assert source.add_message("reference", "user", "New history")
-    target = stores("target")
-    assert target.add_message("reference", "user", "Original history")
-    target._connect().execute("PRAGMA query_only=ON")
-    report = target.import_sessions(source.export_all(), overwrite=True)
-    assert report.imported == 0
-    assert report.skipped_count == 1
-    assert "write failed" in report.skipped[0]["reason"]
-    target.invalidate_cache()
-    assert target.get_chat_history("reference")[0]["content"] == "Original history"
-
-
-@pytest.mark.parametrize("reset_live_fields", [True, False])
-def test_restore_preserves_archived_history_and_tool_calls(stores, reset_live_fields):
-    tool_calls = [{"id": "call-1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}]
-    payload = {"sessions": [{
-        "session_id": "reference",
-        "gateway_session_id": "live",
-        "agent_id": "agent",
-        "metadata": {"gateway_session_id": "live", "agent_id": "agent", "lineage_id": "shared"},
-        "archived_messages": [{"role": "user", "content": "Archived", "timestamp": 1}],
-        "messages": [{"role": "assistant", "content": "Tool call", "tool_calls": tool_calls}],
-    }]}
-    target = stores("target", active_window=1, retention="truncate")
-    assert target.add_message("existing", "user", "Existing")
-    assert target.import_sessions(payload, reset_live_fields=reset_live_fields).imported == 1
-    record = stores("target").export_session("reference")["sessions"][0]
-    assert record["messages"][0]["tool_calls"] == tool_calls
-    assert record["archived_messages"][0]["content"] == "Archived"
-    assert record["metadata"]["lineage_id"] == "shared"
-    if reset_live_fields:
-        assert record.get("gateway_session_id") is None
-        assert record.get("agent_id") is None
-        assert "gateway_session_id" not in record["metadata"]
-        assert "agent_id" not in record["metadata"]
-    else:
-        assert record["gateway_session_id"] == "live"
-        assert record["agent_id"] == "agent"
-        assert record["metadata"]["gateway_session_id"] == "live"
-        assert record["metadata"]["agent_id"] == "agent"
-
-
-def test_sqlite_restore_keeps_import_guards(stores):
-    source = stores("source", DefaultSessionStore)
-    assert source.add_message("reference", "user", "Original")
-    payload = source.export_all()
-    payload["sessions"].append(dict(payload["sessions"][0]))
-    target = stores("target")
-    report = target.import_sessions(payload)
-    assert report.imported == 1
-    assert "duplicate" in report.skipped[0]["reason"]
-    assert target.import_sessions(payload).imported == 0
-    assert target.import_sessions(payload, overwrite=True).imported == 1
-    assert target.export_session("missing")["sessions"] == []
-    payload["version"] = target.PORTABLE_VERSION + 1
-    assert target.import_sessions(payload, overwrite=True).imported == 0
-    assert target.get_chat_history("reference")[0]["content"] == "Original"
-
-
-@pytest.mark.parametrize("overwrite", [False, True])
-def test_import_does_not_clobber_peer_created_after_initial_check(stores, monkeypatch, overwrite):
-    target = stores("target")
-    peer = stores("target")
-    original = target._save_imported_session
-
-    def create_peer_before_write(*args, **kwargs):
-        assert peer.add_message("reference", "user", "peer-created")
-        return original(*args, **kwargs)
-
-    monkeypatch.setattr(target, "_save_imported_session", create_peer_before_write)
-    payload = {"sessions": [{"session_id": "reference", "messages": [{"role": "user", "content": "imported"}]}]}
-    report = target.import_sessions(payload, overwrite=overwrite)
-    assert report.imported == int(overwrite)
-    assert target.get_chat_history("reference")[0]["content"] == ("imported" if overwrite else "peer-created")
-    if overwrite:
-        assert target._cache["reference"].messages[0].content == "imported"
-    else:
-        assert report.skipped[0]["reason"] == "already exists (use overwrite)"
-
-
-def test_lineage_export_does_not_materialize_unrelated_transcripts(stores):
-    store = stores("source")
-    for sid in ("root", "continuation"):
-        assert store.add_message(sid, "user", sid)
-        assert store.update_session_metadata(sid, lineage_id="shared")
-    for index in range(24):
-        sid = f"unrelated-{index}"
-        assert store.add_message(sid, "user", "x" * 256_000)
-        assert store.update_session_metadata(sid, lineage_id="other", value=float("nan"))
-    tracemalloc.start()
-    try:
-        payload = store.export_session("continuation")
-        _, peak = tracemalloc.get_traced_memory()
-    finally:
-        tracemalloc.stop()
-    assert {record["session_id"] for record in payload["sessions"]} == {"root", "continuation"}
-    # The unrelated transcripts total over 6 MB. A single-record parse can
-    # allocate its input and decoded content, but must not retain all rows.
-    assert peak < 2_000_000

src/praisonai-agents/tests/unit/session/test_sqlite_transcript_store.py

@@ -204,57 +204,6 @@ class TestSqliteTranscriptStore:
             {"role": "user", "content": "legacy transcript"}
         ]
 
-    def test_migration_skips_invalid_utf8_legacy_file(self, tmp_dir):
-        """A legacy file with invalid UTF-8 bytes is skipped (not fatal), and
-        valid transcripts still migrate; the bad bytes are preserved."""
-        legacy = DefaultSessionStore(session_dir=tmp_dir)
-        legacy.add_message("old", "user", "legacy transcript")
-
-        bad_path = os.path.join(tmp_dir, "invalid.json")
-        with open(bad_path, "wb") as handle:
-            handle.write(b"\xff\xfe")
-
-        store = SqliteTranscriptStore(session_dir=tmp_dir)
-        assert store.session_exists("old")
-        assert store.get_chat_history("old") == [
-            {"role": "user", "content": "legacy transcript"}
-        ]
-
-        with open(bad_path, "rb") as handle:
-            assert handle.read() == b"\xff\xfe"
-
-    def test_migration_continues_after_invalid_utf8_ordered_first(self, tmp_dir):
-        """A bad file processed before valid ones must not abort later migrations.
-
-        Migration sorts filenames, so ``aaa_invalid.json`` is deterministically
-        processed before the ``zeta`` session's JSON file — guaranteeing the
-        bad-first scenario is actually exercised.
-        """
-        legacy = DefaultSessionStore(session_dir=tmp_dir)
-        legacy.add_message("zeta", "user", "valid transcript")
-
-        bad_path = os.path.join(tmp_dir, "aaa_invalid.json")
-        with open(bad_path, "wb") as handle:
-            handle.write(b"\xff\xfe")
-
-        import json as _json
-
-        json_files = sorted(f for f in os.listdir(tmp_dir) if f.endswith(".json"))
-        assert json_files[0] == "aaa_invalid.json"
-        valid_file = next(
-            f for f in json_files
-            if f != "aaa_invalid.json" and "zeta" in _json.dumps(
-                _json.load(open(os.path.join(tmp_dir, f)))
-            )
-        )
-        assert "aaa_invalid.json" < valid_file
-
-        store = SqliteTranscriptStore(session_dir=tmp_dir)
-        assert store.session_exists("zeta")
-        assert store.get_chat_history("zeta") == [
-            {"role": "user", "content": "valid transcript"}
-        ]
-
 
 class TestTranscriptArchivedRecall:
     """Issue #5031: transcript store must recall compacted (archived) turns."""

src/praisonai-agents/tests/unit/session/test_title_length_boundaries.py

@@ -1,76 +0,0 @@
-"""The public title length cap applies to generated and fallback titles."""
-
-import pytest
-
-from praisonaiagents.session.title import generate_title, generate_title_async
-
-
-@pytest.mark.parametrize("max_length", [0, 1, 2, 3, 4, 60])
-@pytest.mark.parametrize("user_msg", ["", "?!.", "A long user request about debugging code"])
-@pytest.mark.asyncio
-async def test_sync_fallback_in_running_loop_respects_length(user_msg, max_length):
-    title = generate_title(user_msg, "Response", max_length=max_length)
-    assert len(title) <= max_length
-    if max_length == 60:
-        assert title
-
-
-@pytest.mark.parametrize("max_length", [0, 1, 2, 3, 4, 60])
-@pytest.mark.parametrize("response", ["A long generated title about Python", None])
-@pytest.mark.asyncio
-async def test_async_generated_and_fallback_titles_respect_length(monkeypatch, max_length, response):
-    import praisonaiagents.llm as llm_module
-
-    class LocalLLM:
-        def __init__(self, **kwargs):
-            pass
-
-        async def get_response_async(self, **kwargs):
-            return response
-
-    monkeypatch.setattr(llm_module, "LLM", LocalLLM)
-    title = await generate_title_async("", "Response", llm_model="test", max_length=max_length)
-    assert len(title) <= max_length
-    if max_length == 60:
-        assert title == (response or "Chat Session")
-
-
-@pytest.mark.parametrize("max_length", [0, -1])
-@pytest.mark.asyncio
-async def test_nonpositive_cap_skips_model_initialization(monkeypatch, max_length):
-    import praisonaiagents.llm as llm_module
-
-    calls = []
-
-    class UnexpectedLLM:
-        def __init__(self, **kwargs):
-            calls.append(kwargs)
-
-        async def get_response_async(self, **kwargs):
-            return "Unneeded title"
-
-    monkeypatch.setattr(llm_module, "LLM", UnexpectedLLM)
-    assert await generate_title_async("Question", "Response", max_length=max_length) == ""
-    assert generate_title("Question", "Response", max_length=max_length) == ""
-    assert calls == []
-
-
-@pytest.mark.parametrize("user_msg", ["", "?!."])
-@pytest.mark.asyncio
-async def test_three_character_fallback_keeps_readable_prefix(user_msg):
-    assert generate_title(user_msg, "Response", max_length=3) == "Cha"
-
-
-@pytest.mark.asyncio
-async def test_three_character_generated_title_keeps_readable_prefix(monkeypatch):
-    import praisonaiagents.llm as llm_module
-
-    class LocalLLM:
-        def __init__(self, **kwargs):
-            pass
-
-        async def get_response_async(self, **kwargs):
-            return "Python debugging"
-
-    monkeypatch.setattr(llm_module, "LLM", LocalLLM)
-    assert await generate_title_async("Question", "Response", max_length=3) == "Pyt"

src/praisonai-agents/tests/unit/storage/test_backends.py

@@ -11,8 +11,6 @@ Tests cover:
 import threading
 from pathlib import Path
 
-import pytest
-
 from praisonaiagents.storage.backends import FileBackend, SQLiteBackend, get_backend
 from praisonaiagents.storage.protocols import StorageBackendProtocol
 
@@ -35,33 +33,7 @@ class TestFileBackend:
         
         loaded = backend.load("test_key")
         assert loaded == data
-
-    def test_file_backend_load_invalid_utf8_returns_none(self, tmp_path, caplog):
-        """Undecodable bytes should warn and return None, not raise (Issue #5566)."""
-        import logging
-
-        backend = FileBackend(storage_dir=str(tmp_path))
-
-        for raw in (b"\xff", b"\xe4\xb8", b"\xed\xa0\x80"):
-            backend.save("bad", {"message": "original"})
-            file_path = Path(tmp_path) / "bad.json"
-            file_path.write_bytes(raw)
-
-            with caplog.at_level(logging.WARNING, logger="praisonaiagents.storage.backends"):
-                caplog.clear()
-                assert backend.load("bad") is None
-                # A warning is emitted identifying the failed key
-                assert any(
-                    rec.levelno == logging.WARNING and "Failed to load bad" in rec.getMessage()
-                    for rec in caplog.records
-                )
-            # Original bytes are preserved (load is non-destructive)
-            assert file_path.read_bytes() == raw
-
-        # Valid neighboring keys still load
-        backend.save("good", {"key": "value"})
-        assert backend.load("good") == {"key": "value"}
-
+    
     def test_file_backend_exists(self, tmp_path):
         """Test exists check."""
         backend = FileBackend(storage_dir=str(tmp_path))
@@ -143,29 +115,6 @@ class TestFileBackend:
         backend = FileBackend(storage_dir=str(tmp_path))
         assert isinstance(backend, StorageBackendProtocol)
 
-    @pytest.mark.parametrize("pretty", [True, False])
-    def test_file_backend_save_cleans_tmp_on_serialization_error(self, tmp_path, pretty):
-        """Serialization failures must not leave partial .tmp staging files."""
-        backend = FileBackend(storage_dir=str(tmp_path), pretty=pretty)
-        backend.save("good", {"value": 1})
-
-        circular = {}
-        circular["self"] = circular
-        with pytest.raises(ValueError):
-            backend.save("circular", circular)
-
-        with pytest.raises(TypeError):
-            backend.save("tuple_key", {("a", "b"): 1})
-
-        with pytest.raises((ValueError, TypeError)):
-            backend.save("good", circular)
-
-        tmp_files = list(tmp_path.glob("*.tmp"))
-        assert tmp_files == []
-        assert backend.load("good") == {"value": 1}
-        assert not backend.exists("circular")
-        assert not backend.exists("tuple_key")
-
 
 class TestSQLiteBackend:
     """Tests for SQLiteBackend."""
@@ -299,68 +248,6 @@ class TestSQLiteBackend:
         loaded = backend2.load("key1")
         assert loaded["data"] == 1
 
-    @pytest.mark.parametrize("table_name", ["select", "123", "9table", "praison_storage"])
-    def test_sqlite_backend_keyword_and_digit_table_names(self, tmp_path, table_name):
-        """Validator-accepted names that are SQL keywords or start with a digit
-        must work across schema creation and all CRUD operations (Issue #5569)."""
-        db_path = tmp_path / "test.db"
-        backend = SQLiteBackend(db_path=str(db_path), table_name=table_name)
-
-        # save / upsert
-        backend.save("key1", {"version": 1})
-        backend.save("key1", {"version": 2})
-        assert backend.load("key1")["version"] == 2
-
-        # exists
-        assert backend.exists("key1")
-        assert not backend.exists("missing")
-
-        # list_keys (full + prefix)
-        backend.save("key2", {"data": 2})
-        backend.save("other", {"data": 3})
-        assert backend.list_keys() == ["key1", "key2", "other"]
-        assert backend.list_keys(prefix="key") == ["key1", "key2"]
-
-        # delete
-        assert backend.delete("other") is True
-        assert not backend.exists("other")
-
-        # clear
-        assert backend.clear() == 2
-        assert backend.list_keys() == []
-
-    def test_sqlite_backend_reopen_with_auto_create_false(self, tmp_path):
-        """A keyword table name persists and reopens with auto_create=False."""
-        db_path = tmp_path / "test.db"
-        backend = SQLiteBackend(db_path=str(db_path), table_name="select")
-        backend.save("key1", {"data": 1})
-        backend.close()
-
-        backend2 = SQLiteBackend(
-            db_path=str(db_path), table_name="select", auto_create=False
-        )
-        assert backend2.load("key1")["data"] == 1
-
-    def test_sqlite_backend_separate_tables_same_db(self, tmp_path):
-        """Quoted identifiers keep distinct tables isolated in one DB file."""
-        db_path = tmp_path / "test.db"
-        t1 = SQLiteBackend(db_path=str(db_path), table_name="select")
-        t2 = SQLiteBackend(db_path=str(db_path), table_name="praison_storage")
-
-        t1.save("shared", {"from": "select"})
-        t2.save("shared", {"from": "praison_storage"})
-
-        assert t1.load("shared")["from"] == "select"
-        assert t2.load("shared")["from"] == "praison_storage"
-
-    def test_sqlite_backend_invalid_table_name_rejected(self, tmp_path):
-        """Names with disallowed characters are still rejected by the validator."""
-        db_path = tmp_path / "test.db"
-        with pytest.raises(ValueError):
-            SQLiteBackend(db_path=str(db_path), table_name="bad name")
-        with pytest.raises(ValueError):
-            SQLiteBackend(db_path=str(db_path), table_name='drop";')
-
 
 class TestGetBackend:
     """Tests for get_backend factory function."""

src/praisonai-agents/tests/unit/storage/test_file_backend_compound_suffix.py

@@ -1,30 +0,0 @@
-"""Enumeration and clearing honor the complete configured filename suffix."""
-
-import pytest
-
-from praisonaiagents.storage.backends import FileBackend
-
-
-@pytest.mark.parametrize("suffix", [".data.json", "record", ".json", ""])
-@pytest.mark.parametrize("operation", ["list", "clear"])
-def test_configured_suffix_selects_saved_records(tmp_path, suffix, operation):
-    store = FileBackend(storage_dir=str(tmp_path), suffix=suffix)
-    for key in ("session_one", "session_two", "other"):
-        store.save(key, {"key": key})
-    unrelated = tmp_path / "unrelated.txt"
-    unrelated.write_text("keep", encoding="utf-8")
-    directory = tmp_path / ("directory" + suffix)
-    directory.mkdir()
-    assert store.exists("session_one")
-    assert store.load("session_one") == {"key": "session_one"}
-    if operation == "list":
-        assert store.list_keys() == ["other", "session_one", "session_two"]
-        assert store.list_keys("session_") == ["session_one", "session_two"]
-        for key in store.list_keys():
-            assert store.load(key) == {"key": key}
-    else:
-        assert store.clear() == 3
-        assert not store.exists("session_one")
-        assert store.list_keys() == []
-    assert unrelated.read_text(encoding="utf-8") == "keep"
-    assert directory.is_dir()

src/praisonai-agents/tests/unit/storage/test_session_cleanup_full_enumeration.py

@@ -1,41 +0,0 @@
-"""Retention must include sessions beyond the listing display limit."""
-
-import os
-import time
-
-import pytest
-
-from praisonaiagents.storage.base import cleanup_old_sessions, list_json_sessions
-
-
-@pytest.mark.parametrize("boundary", ["age", "size"])
-@pytest.mark.parametrize("suffix", [".json", ".jsonl"])
-def test_cleanup_includes_oldest_session_after_ten_thousand_newer_files(tmp_path, boundary, suffix):
-    oldest = tmp_path / ("oldest" + suffix)
-    oldest.write_bytes(b"x")
-    now = time.time()
-    old_time = now - 90 * 24 * 60 * 60
-    os.utime(oldest, (old_time, old_time))
-    for index in range(10000):
-        (tmp_path / f"new-{index:05d}{suffix}").write_bytes(b"")
-    neighbor = tmp_path / "unrelated.txt"
-    neighbor.write_bytes(b"keep")
-
-    if boundary == "age":
-        deleted = cleanup_old_sessions(tmp_path, suffix=suffix, max_age_days=30, max_size_mb=100)
-    else:
-        # Only the oldest file contributes bytes to the quota; none exceed age.
-        deleted = cleanup_old_sessions(tmp_path, suffix=suffix, max_age_days=365, max_size_mb=0)
-
-    assert deleted == 1
-    assert not oldest.exists()
-    assert len(list(tmp_path.glob("new-*" + suffix))) == 10000
-    assert neighbor.read_bytes() == b"keep"
-
-
-def test_listing_keeps_default_limit_and_supports_unbounded_listing(tmp_path):
-    for index in range(51):
-        (tmp_path / f"session-{index:02d}.json").write_text("{}", encoding="utf-8")
-    assert len(list_json_sessions(tmp_path)) == 50
-    assert len(list_json_sessions(tmp_path, limit=3)) == 3
-    assert len(list_json_sessions(tmp_path, limit=None)) == 51

src/praisonai-agents/tests/unit/storage/test_session_list_utf8_counts.py

@@ -1,54 +0,0 @@
-"""Session counts use the UTF-8 writer format, independent of host locale."""
-
-import builtins
-import json
-from pathlib import Path
-
-import pytest
-
-from praisonaiagents.storage.base import BaseJSONStore, list_json_sessions
-
-
-@pytest.fixture
-def legacy_text_locale(monkeypatch, tmp_path):
-    original_open = builtins.open
-
-    def locale_open(file, mode="r", *args, **kwargs):
-        if Path(file).parent == tmp_path and mode == "r" and not kwargs.get("encoding"):
-            kwargs["encoding"] = "cp1252"
-        return original_open(file, mode, *args, **kwargs)
-
-    # Exercise real decoding with a deterministic Windows-style default locale.
-    monkeypatch.setattr(builtins, "open", locale_open)
-
-
-@pytest.mark.parametrize("field", ["iterations", "messages", "items", "jsonl"])
-def test_utf8_session_counts_survive_legacy_locale(tmp_path, legacy_text_locale, field):
-    values = ["丁 café 😀", "second"]
-    suffix = ".jsonl" if field == "jsonl" else ".json"
-    path = tmp_path / ("session" + suffix)
-    if field == "jsonl":
-        path.write_text("".join(json.dumps({"value": value}, ensure_ascii=False) + "\n" for value in values), encoding="utf-8")
-    else:
-        BaseJSONStore(path).save({field: values})
-    before = path.read_bytes()
-
-    sessions = list_json_sessions(tmp_path, suffix=suffix)
-
-    assert len(sessions) == 1
-    assert sessions[0].item_count == 2
-    assert sessions[0].size_bytes == len(before)
-    assert path.read_bytes() == before
-
-
-@pytest.mark.parametrize("suffix", [".json", ".jsonl"])
-@pytest.mark.parametrize("valid", [True, False])
-def test_ascii_and_unreadable_files_keep_existing_behavior(tmp_path, legacy_text_locale, suffix, valid):
-    path = tmp_path / ("session" + suffix)
-    content = b'{"items":[1,2]}' if suffix == ".json" else b'{"value":1}\n{"value":2}\n'
-    path.write_bytes(content if valid else b"\x81")
-    before = path.read_bytes()
-    sessions = list_json_sessions(tmp_path, suffix=suffix)
-    assert len(sessions) == 1
-    assert sessions[0].item_count == (2 if valid else 0)
-    assert path.read_bytes() == before

src/praisonai-agents/tests/unit/storage/test_sqlite_backend_init_cleanup.py

@@ -1,52 +0,0 @@
-"""A failed backend constructor releases its real SQLite connection."""
-
-import sqlite3
-
-import pytest
-
-from praisonaiagents.storage import sqlite as connector
-from praisonaiagents.storage.backends import SQLiteBackend
-
-
-@pytest.mark.parametrize("failure", ["reserved_name", "incompatible_schema"])
-def test_constructor_closes_connection_after_schema_failure(tmp_path, monkeypatch, failure):
-    path = str(tmp_path / "store.db")
-    with sqlite3.connect(path) as seed:
-        seed.execute("CREATE TABLE preserved (value TEXT)")
-        seed.execute("INSERT INTO preserved VALUES ('original')")
-        if failure == "incompatible_schema":
-            seed.execute("CREATE TABLE praison_storage (wrong_column TEXT)")
-    seed.close()
-    opened = []
-    original = connector.connect
-
-    def connect(*args, **kwargs):
-        conn = original(*args, **kwargs)
-        opened.append(conn)
-        return conn
-
-    monkeypatch.setattr(connector, "connect", connect)
-    try:
-        name = "sqlite_private" if failure == "reserved_name" else "praison_storage"
-        with pytest.raises(sqlite3.OperationalError) as error:
-            SQLiteBackend(db_path=path, table_name=name)
-        expected = "reserved" if failure == "reserved_name" else "no such column"
-        assert expected in str(error.value)
-        assert len(opened) == 1
-        with pytest.raises(sqlite3.ProgrammingError, match="closed database"):
-            opened[0].execute("SELECT 1")
-        with sqlite3.connect(path) as check:
-            assert check.execute("SELECT value FROM preserved").fetchall() == [("original",)]
-        check.close()
-    finally:
-        for conn in opened:
-            conn.close()
-
-
-def test_successful_constructor_keeps_connection_usable(tmp_path):
-    backend = SQLiteBackend(db_path=str(tmp_path / "store.db"))
-    try:
-        backend.save("key", {"value": 1})
-        assert backend.load("key") == {"value": 1}
-    finally:
-        backend.close()

src/praisonai-agents/tests/unit/test_compaction_wiring.py

@@ -97,10 +97,8 @@ class TestCompactionMaxTokensResolution:
         agent = self._make_agent("gpt-4o")
         cfg = ExecutionConfig(context_compaction=True)
         resolved = agent._resolve_compaction_max_tokens(cfg)
-        # Well beyond the working budget must trigger compaction. Repeated
-        # characters compress heavily under tiktoken (~8 chars/token), so use
-        # 8× not the heuristic 4× when sizing synthetic overflow content.
-        msgs = [{"role": "user", "content": "A" * 8 * (resolved + 50000)}]
+        # Well beyond the working budget must trigger compaction.
+        msgs = [{"role": "user", "content": "A" * 4 * (resolved + 50000)}]
         compactor = ContextCompactor(max_tokens=resolved)
         assert compactor.needs_compaction(msgs) is True
 

src/praisonai-agents/tests/unit/test_run_on_docker.py

@@ -308,7 +308,6 @@ def test_a_hosted_backend_reclaims_its_instance_when_collected():
     instance kept billing until the provider's own idle timer noticed, which
     docker and flyio do not have."""
     import gc
-    import time
 
     from praisonai.integrations.compute_managed_agent import ComputeManagedAgent
 
@@ -339,12 +338,6 @@ def test_a_hosted_backend_reclaims_its_instance_when_collected():
 
     del backend
     gc.collect()
-    # During live GC the finalizer reclaims fire-and-forget on the shared
-    # AsyncBridge's background loop, so the shutdown lands a beat after
-    # gc.collect() returns. Wait (bounded) for it rather than racing it.
-    deadline = time.monotonic() + 5.0
-    while not released and time.monotonic() < deadline:
-        time.sleep(0.01)
     assert released == ["inst-1"], "the instance outlived the backend that owned it"
 
 
@@ -368,7 +361,6 @@ def test_release_reclaims_even_on_an_event_loop_thread():
     ever runs, silently leaking the instance -- so _release must bridge the loop
     the way SharedCompute does, not call asyncio.run() directly."""
     import asyncio
-    import time
 
     from praisonai.integrations.compute_managed_agent import _release
 
@@ -382,11 +374,4 @@ def test_release_reclaims_even_on_an_event_loop_thread():
         _release(Fake(), "inst-loop", "docker")
 
     asyncio.run(collect_on_the_loop())
-    # _release routes through the shared AsyncBridge's background loop
-    # fire-and-forget, so the shutdown lands a beat after asyncio.run()
-    # returns. Wait (bounded) for it rather than racing it — the assertion
-    # still fails if _release ever leaks (never schedules the shutdown).
-    deadline = time.monotonic() + 5.0
-    while not released and time.monotonic() < deadline:
-        time.sleep(0.01)
     assert released == ["inst-loop"], "an instance leaked when GC ran on a live loop"

src/praisonai-agents/tests/unit/workflows/test_cache_run_condition.py

@@ -0,0 +1,55 @@
+"""A cached step result must still respect its current execution gate."""
+
+from praisonaiagents import Task
+from praisonaiagents.workflows.workflows import AgentFlow
+
+
+def test_cached_step_rechecks_gate_without_rerunning_handler():
+    enabled = {"value": True}
+    checks = []
+    calls = []
+
+    def should_run(ctx):
+        checks.append(enabled["value"])
+        return enabled["value"]
+
+    def handler(ctx):
+        calls.append("run")
+        return "done"
+
+    step = Task(name="conditional", handler=handler, should_run=should_run)
+    flow = AgentFlow(steps=[step], cache=True)
+    flow.run("same", verbose=False)
+    flow.run("same", verbose=False)
+    enabled["value"] = False
+    skipped = flow.run("same", verbose=False)
+
+    assert checks == [True, True, False]
+    assert calls == ["run"]
+    assert skipped["steps"] == []
+    assert flow.step_statuses["conditional"] == "skipped"
+    assert "conditional_output" not in skipped["variables"]
+
+    enabled["value"] = True
+    resumed = flow.run("same", verbose=False)
+    assert checks == [True, True, False, True]
+    assert calls == ["run"]
+    assert resumed["output"] == "done"
+    # The reopened gate serves a cached result: the step's status must reflect
+    # that terminal success, not the "skipped" state left by the closed-gate run.
+    assert flow.step_statuses["conditional"] == "completed"
+    assert step.status == "completed"
+
+
+def test_skipped_step_is_not_cached_when_gate_later_opens():
+    enabled = {"value": False}
+    calls = []
+    step = Task(
+        name="conditional", should_run=lambda ctx: enabled["value"],
+        handler=lambda ctx: calls.append("run") or "done",
+    )
+    flow = AgentFlow(steps=[step], cache=True)
+    assert flow.run("same", verbose=False)["steps"] == []
+    enabled["value"] = True
+    assert flow.run("same", verbose=False)["output"] == "done"
+    assert calls == ["run"]

src/praisonai-bot/praisonai_bot/bots/_http_approval.py

@@ -23,18 +23,11 @@ from __future__ import annotations
 
 import asyncio
 import logging
-import os
-import secrets
 import time
 import uuid
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Optional
 
-from ._approval_base import (
-    DEFAULT_APPROVAL_TIMEOUT,
-    DurableApprovalMixin,
-    is_authorized_actor,
-    normalize_approvers,
-)
+from ._approval_base import DEFAULT_APPROVAL_TIMEOUT, DurableApprovalMixin
 
 logger = logging.getLogger(__name__)
 
@@ -52,14 +45,6 @@ class HTTPApproval(DurableApprovalMixin):
         host: Bind address (default ``127.0.0.1``).
         port: Port to listen on (default ``8899``).
         timeout: Max seconds to wait for a response (default 300).
-        allowed_approvers: Optional allowlist of approver IDs permitted to
-            resolve an approval. When provided, each pending approval mints a
-            single-use token that is carried only in the per-approval URL; the
-            resolver must present that token (and, when the request supplies an
-            ``approver``, be in the allowlist) or the decision is rejected. When
-            ``None`` (default) any visitor may decide (legacy behaviour,
-            backward compatible). Falls back to a comma-separated
-            ``HTTP_APPROVERS`` env var when not passed.
 
     Example::
 
@@ -72,17 +57,11 @@ class HTTPApproval(DurableApprovalMixin):
         host: str = "127.0.0.1",
         port: int = 8899,
         timeout: float = DEFAULT_APPROVAL_TIMEOUT,
-        allowed_approvers: Optional[Iterable[str]] = None,
         store: Optional[Any] = None,
     ):
         self._host = host
         self._port = port
         self._timeout = timeout
-        if allowed_approvers is None:
-            _env = os.environ.get("HTTP_APPROVERS", "").strip()
-            if _env:
-                allowed_approvers = [a.strip() for a in _env.split(",") if a.strip()]
-        self._allowed_approvers = normalize_approvers(allowed_approvers)
         self._pending: Dict[str, Optional[Dict[str, Any]]] = {}
         self._server_started = False
         self._runner: Optional[Any] = None
@@ -112,20 +91,6 @@ class HTTPApproval(DurableApprovalMixin):
         self._server_started = True
         logger.info(f"HTTPApproval server started on http://{self._host}:{self._port}")
 
-    def _token_ok(self, pending: Dict[str, Any], request) -> bool:
-        """Return whether *request* carries the per-approval token.
-
-        The token is minted per pending approval and embedded only in the URL
-        handed to the authorised approver. A constant-time comparison is used so
-        the endpoint does not leak the token via timing. When no allowlist is
-        configured there is no token, so this is a no-op (legacy behaviour).
-        """
-        expected = pending.get("token")
-        if not expected:
-            return True
-        supplied = request.query.get("token") or request.headers.get("X-Approval-Token", "")
-        return secrets.compare_digest(str(supplied), str(expected))
-
     async def _handle_page(self, request) -> Any:
         """Serve the approval page HTML."""
         from aiohttp import web
@@ -136,18 +101,11 @@ class HTTPApproval(DurableApprovalMixin):
         if pending is None:
             return web.Response(text="Approval request not found or already decided.", status=404)
 
-        if not self._token_ok(pending, request):
-            logger.warning(
-                "Unauthorised HTTP approval page access for %s", request_id,
-            )
-            return web.Response(text="Not authorized to view this approval.", status=403)
-
         if pending.get("decided"):
             return web.Response(text="This approval has already been decided.", status=200)
 
         info = pending.get("info", {})
-        token = pending.get("token")
-        html = self._build_html(request_id, info, token=token)
+        html = self._build_html(request_id, info)
         return web.Response(text=html, content_type="text/html")
 
     async def _handle_decide(self, request) -> Any:
@@ -159,13 +117,6 @@ class HTTPApproval(DurableApprovalMixin):
 
         if pending is None:
             return web.Response(text="Not found", status=404)
-
-        if not self._token_ok(pending, request):
-            logger.warning(
-                "Unauthorised HTTP approval decision attempt for %s", request_id,
-            )
-            return web.json_response({"ok": False, "error": "unauthorized"}, status=403)
-
         if pending.get("decided"):
             return web.Response(text="Already decided", status=200)
 
@@ -174,32 +125,18 @@ class HTTPApproval(DurableApprovalMixin):
         except Exception:
             body = {}
 
-        approver = body.get("approver", "http_user")
-
-        # Authorisation boundary: when an allowlist is configured a free-text
-        # ``approver`` must be one of the authorised approvers. The per-approval
-        # token already proves the resolver received the private URL; the
-        # allowlist additionally pins the recorded approver identity.
-        if not is_authorized_actor(approver, self._allowed_approvers):
-            logger.warning(
-                "Unauthorised HTTP approver %r for %s", approver, request_id,
-            )
-            return web.json_response({"ok": False, "error": "unauthorized"}, status=403)
-
         decision = body.get("decision", "deny")
         pending["decided"] = True
         pending["approved"] = decision == "approve"
         pending["reason"] = body.get("reason", f"{'Approved' if pending['approved'] else 'Denied'} via HTTP dashboard")
-        pending["approver"] = approver
+        pending["approver"] = body.get("approver", "http_user")
 
         return web.json_response({"ok": True, "decision": decision})
 
-    def _build_html(self, request_id: str, info: Dict[str, Any], token: Optional[str] = None) -> str:
+    def _build_html(self, request_id: str, info: Dict[str, Any]) -> str:
         """Build a minimal approval page."""
         import html as html_module
 
-        token_qs = f"?token={html_module.escape(str(token))}" if token else ""
-
         tool_name = html_module.escape(str(info.get("tool_name", "unknown")))
         risk_level = html_module.escape(str(info.get("risk_level", "unknown")))
         agent_name = html_module.escape(str(info.get("agent_name", "")))
@@ -244,7 +181,7 @@ td {{ padding: 8px; border-bottom: 1px solid #333; }} code {{ background: #2a2a4
 <div class="done" id="result" style="display:none"></div>
 <script>
 async function decide(d) {{
-  const res = await fetch('/approve/{request_id}/decide{token_qs}', {{
+  const res = await fetch('/approve/{request_id}/decide', {{
     method: 'POST', headers: {{'Content-Type': 'application/json'}},
     body: JSON.stringify({{decision: d}})
   }});
@@ -265,13 +202,9 @@ async function decide(d) {{
         await self._ensure_server()
 
         request_id = str(uuid.uuid4())
-        # Mint a single-use token bound to this approval when an allowlist is
-        # configured, so only the holder of the private URL can resolve it.
-        token = secrets.token_urlsafe(32) if self._allowed_approvers is not None else None
         self._pending[request_id] = {
             "decided": False,
             "approved": False,
-            "token": token,
             "info": {
                 "tool_name": request.tool_name,
                 "arguments": request.arguments,
@@ -281,8 +214,6 @@ async function decide(d) {{
         }
 
         url = f"http://{self._host}:{self._port}/approve/{request_id}"
-        if token:
-            url = f"{url}?token={token}"
         logger.info(f"HTTPApproval: Waiting for decision at {url}")
         print(f"\n🔗 Open this URL to approve/deny:\n   {url}\n")
 

src/praisonai-bot/praisonai_bot/bots/_webhook_approval.py

@@ -27,14 +27,9 @@ import logging
 import os
 import time
 import uuid
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Optional
 
-from ._approval_base import (
-    DEFAULT_APPROVAL_TIMEOUT,
-    DurableApprovalMixin,
-    is_authorized_actor,
-    normalize_approvers,
-)
+from ._approval_base import DEFAULT_APPROVAL_TIMEOUT, DurableApprovalMixin
 
 logger = logging.getLogger(__name__)
 
@@ -56,13 +51,6 @@ class WebhookApproval(DurableApprovalMixin):
         headers: Extra HTTP headers (e.g. auth tokens).
         timeout: Max seconds to wait for a response (default 300).
         poll_interval: Seconds between status polls (default 5.0).
-        allowed_approvers: Optional allowlist of approver IDs permitted to
-            resolve an approval. When provided, a decision whose ``approver``
-            is not in the allowlist is dropped (polling continues) so an
-            unauthorised responder to the status endpoint cannot resolve a
-            gated tool. When ``None`` (default) any responder may decide
-            (legacy behaviour, backward compatible). Falls back to a
-            comma-separated ``WEBHOOK_APPROVERS`` env var when not passed.
 
     Example::
 
@@ -83,7 +71,6 @@ class WebhookApproval(DurableApprovalMixin):
         headers: Optional[Dict[str, str]] = None,
         timeout: float = DEFAULT_APPROVAL_TIMEOUT,
         poll_interval: float = 5.0,
-        allowed_approvers: Optional[Iterable[str]] = None,
         store: Optional[Any] = None,
     ):
         self._webhook_url = webhook_url or os.environ.get("APPROVAL_WEBHOOK_URL", "")
@@ -95,29 +82,11 @@ class WebhookApproval(DurableApprovalMixin):
         self._headers = headers or {}
         self._timeout = timeout
         self._poll_interval = poll_interval
-        if allowed_approvers is None:
-            _env = os.environ.get("WEBHOOK_APPROVERS", "").strip()
-            if _env:
-                allowed_approvers = [a.strip() for a in _env.split(",") if a.strip()]
-        self._allowed_approvers = normalize_approvers(allowed_approvers)
         self._init_store(store)
 
     def __repr__(self) -> str:
         return f"WebhookApproval(webhook_url={self._webhook_url!r})"
 
-    def _approver_authorized(self, approver: Any) -> bool:
-        """Return whether *approver* may resolve an approval.
-
-        No-op (always ``True``) when no allowlist is configured (legacy
-        behaviour). Otherwise the responder's ``approver`` must be in the
-        allowlist, so an unauthorised party that can reach the status endpoint
-        cannot resolve a gated tool.
-        """
-        return is_authorized_actor(
-            None if approver is None else str(approver),
-            self._allowed_approvers,
-        )
-
     # ── Internal HTTP helper ───────────────────────────────────────────
 
     async def _http_request(
@@ -189,9 +158,7 @@ class WebhookApproval(DurableApprovalMixin):
 
                 # Check for immediate decision
                 if isinstance(post_data, dict):
-                    if "approved" in post_data and self._approver_authorized(
-                        post_data.get("approver")
-                    ):
+                    if "approved" in post_data:
                         decision = ApprovalDecision(
                             approved=bool(post_data["approved"]),
                             reason=post_data.get("reason", "Webhook immediate response"),
@@ -200,11 +167,6 @@ class WebhookApproval(DurableApprovalMixin):
                         )
                         await self._resolve_pending(request, decision)
                         return decision
-                    if "approved" in post_data:
-                        logger.warning(
-                            "Dropping webhook immediate decision from unauthorised "
-                            "approver %r", post_data.get("approver"),
-                        )
 
                 # 2. Poll for decision
                 status_url = self._status_url or f"{self._webhook_url}/{request_id}"
@@ -265,17 +227,6 @@ class WebhookApproval(DurableApprovalMixin):
                 if status == "pending":
                     continue
 
-                has_decision = (
-                    "approved" in data
-                    or status in ("approved", "approve", "yes", "denied", "deny", "rejected", "no")
-                )
-                if has_decision and not self._approver_authorized(data.get("approver")):
-                    logger.warning(
-                        "Dropping webhook decision from unauthorised approver %r "
-                        "for %s", data.get("approver"), request_id,
-                    )
-                    continue
-
                 if "approved" in data:
                     return ApprovalDecision(
                         approved=bool(data["approved"]),

src/praisonai-code/praisonai_code/cli/commands/run.py

@@ -20,10 +20,6 @@ app = typer.Typer(help="Run agents")
 _FRAMEWORK_HELP = "Framework: praisonai, crewai, autogen"
 
 _ALLOW_LOCAL_TOOLS_ENV = "PRAISONAI_ALLOW_LOCAL_TOOLS"
-# Set while a modern ``run <file>.yaml`` is delegating to the legacy
-# ``PraisonAI`` executor. The legacy ``run`` dispatch branch reads it to avoid
-# re-invoking this same Typer ``run`` app (which would recurse infinitely).
-_IN_MODERN_RUN_ENV = "PRAISONAI_IN_MODERN_RUN"
 DEFAULT_MAX_TOKENS = 16000
 
 
@@ -1902,17 +1898,7 @@ def _run_from_file(
     try:
         # Use existing PraisonAI class
         from praisonai_code.cli.main import PraisonAI
-
-        # Re-entrancy guard: the legacy ``PraisonAI.main()`` re-parses ``sys.argv``
-        # and, when it sees ``run <file>.yaml``, dispatches back into *this* modern
-        # Typer ``run`` app. Since we reach here from that same app, an unguarded
-        # delegation loops forever ("maximum recursion depth exceeded"). Setting
-        # this sentinel tells the legacy ``run`` branch we are already inside a
-        # modern run so it executes the YAML directly instead of bouncing back.
-        import os as _os_guard
-        _prev_in_run = _os_guard.environ.get(_IN_MODERN_RUN_ENV)
-        _os_guard.environ[_IN_MODERN_RUN_ENV] = "1"
-
+        
         praison = PraisonAI(
             agent_file=file_path,
             framework=framework or "praisonai",
@@ -2026,14 +2012,6 @@ def _run_from_file(
         output.emit_error(message=str(e))
         output.print_error(str(e))
         raise typer.Exit(1)
-    finally:
-        # Restore the re-entrancy sentinel so suppression never leaks into a
-        # later in-process invocation (embedded/notebook/test reuse).
-        import os as _os_guard
-        if _prev_in_run is None:
-            _os_guard.environ.pop(_IN_MODERN_RUN_ENV, None)
-        else:
-            _os_guard.environ[_IN_MODERN_RUN_ENV] = _prev_in_run
 
 
 def _run_prompt(

src/praisonai-code/praisonai_code/cli/legacy/praison_ai.py

@@ -1405,39 +1405,16 @@ class PraisonAI:
                         or os.path.exists(first)
                     )
                 )
-                if is_yaml_target and os.environ.get("PRAISONAI_IN_MODERN_RUN"):
-                    # We are already *inside* a modern ``run`` invocation
-                    # (``commands/run.py`` delegated the YAML to this legacy
-                    # executor). Re-dispatching to ``run_app`` would re-enter the
-                    # same Typer app and recurse forever ("maximum recursion depth
-                    # exceeded", issue #5597). Execute the YAML directly instead:
-                    # set it as the agent file and fall through to the normal
-                    # legacy execution path below.
-                    #
-                    # Prefer the constructor-provided ``agent_file`` over the
-                    # relative ``sys.argv`` token: the modern runner resolves the
-                    # target to an absolute path *before* a ``--worktree`` chdir
-                    # and passes it here (``self.agent_file`` is still pristine at
-                    # this point), so an untracked/ignored YAML absent from the
-                    # fresh worktree still loads. Fall back to ``first`` only when
-                    # no explicit path was supplied (default ``agents.yaml``
-                    # sentinel from ``__init__``).
-                    if self.agent_file and self.agent_file != "agents.yaml":
-                        pass  # keep the resolved constructor path
-                    else:
-                        self.agent_file = first
-                    args.command = None
-                elif is_yaml_target:
+                if is_yaml_target:
                     # Modern engine (``commands/run.py`` run_main); the first
                     # arg is passed as the Typer ``target`` (YAML file/prompt).
                     from ..commands.run import app as run_app
                     run_app(unknown_args)
                     sys.exit(0)
-                else:
-                    # Run command - async jobs API for long-running tasks
-                    from ..features.jobs import handle_run_command
-                    handle_run_command(unknown_args, verbose=getattr(args, 'verbose', False))
-                    sys.exit(0)
+                # Run command - async jobs API for long-running tasks
+                from ..features.jobs import handle_run_command
+                handle_run_command(unknown_args, verbose=getattr(args, 'verbose', False))
+                sys.exit(0)
             
             elif args.command == 'managed':
                 # Managed agents — delegate to Typer app

src/praisonai-code/tests/unit/test_run_yaml_routing.py

@@ -132,122 +132,5 @@ def test_run_submit_wins_over_same_named_path(monkeypatch, tmp_path):
     assert code == 0
 
 
-def test_modern_run_delegation_does_not_reenter_run_app(monkeypatch, tmp_path):
-    """Issue #5597: ``python -m praisonai <file>.yaml`` must not recurse.
-
-    The modern Typer ``run`` command delegates YAML execution to the legacy
-    ``PraisonAI`` class. That class re-parses ``sys.argv`` (``run <file>.yaml``)
-    and, before this fix, dispatched the ``run`` command straight back into the
-    modern ``run_app`` — which calls the legacy executor again, looping forever
-    until ``maximum recursion depth exceeded``.
-
-    With the ``PRAISONAI_IN_MODERN_RUN`` re-entrancy sentinel set (as the modern
-    runner does while delegating), the legacy ``run`` branch must execute the
-    YAML directly instead of bouncing back to ``run_app``.
-    """
-    pa = _load_module()
-    _require_wrapper_argparse()
-
-    yaml_path = tmp_path / "agents.yaml"
-    yaml_path.write_text("framework: praisonai\nroles: {}\n")
-    monkeypatch.chdir(tmp_path)
-
-    calls = {"run_app": None, "jobs": None}
-    ran = {"direct": False}
-
-    import praisonai_code.cli.commands.run as run_mod
-    import praisonai_code.cli.features.jobs as jobs_mod
-
-    def fake_run_app(args=None, *a, **k):
-        calls["run_app"] = list(args) if args is not None else []
-
-    def fake_handle_run_command(unknown_args, *a, **k):
-        calls["jobs"] = list(unknown_args)
-
-    # Stub the agents generator so the direct YAML execution path is observed
-    # without spinning up a real team / LLM call.
-    class _FakeGenerator:
-        def __init__(self, *a, **k):
-            ran["direct"] = True
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *exc):
-            return False
-
-        def generate_crew_and_kickoff(self):
-            return "READY"
-
-    monkeypatch.setattr(run_mod, "app", fake_run_app)
-    monkeypatch.setattr(jobs_mod, "handle_run_command", fake_handle_run_command)
-    monkeypatch.setattr(pa, "_get_agents_generator", lambda: _FakeGenerator)
-    monkeypatch.setenv("PRAISONAI_IN_MODERN_RUN", "1")
-
-    import sys as _sys
-    monkeypatch.setattr(_sys, "argv", ["praisonai", "run", str(yaml_path)])
-    monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
-
-    result = pa.PraisonAI(agent_file=str(yaml_path)).main()
-
-    # No re-entry into the modern runner (which caused the infinite recursion),
-    # no mis-routing to jobs, and the YAML was executed directly instead.
-    assert calls["run_app"] is None
-    assert calls["jobs"] is None
-    assert ran["direct"] is True
-    assert result == "READY"
-
-
-def test_modern_run_delegation_preserves_resolved_agent_file(monkeypatch, tmp_path):
-    """Greptile #5606: the delegated legacy ``run`` branch must keep the
-    constructor-provided (resolved, possibly absolute) ``agent_file`` instead of
-    overwriting it with the relative ``sys.argv`` token.
-
-    For ``run <file>.yaml --worktree`` the modern runner resolves the target to
-    an absolute path *before* chdir'ing into the isolated worktree, then passes
-    that path to ``PraisonAI(agent_file=...)``. If the legacy branch replaced it
-    with the relative CLI arg, an untracked/ignored YAML (absent from the fresh
-    worktree) would fail to load. This asserts the resolved path survives.
-    """
-    pa = _load_module()
-    _require_wrapper_argparse()
-
-    yaml_path = tmp_path / "agents.yaml"
-    yaml_path.write_text("framework: praisonai\nroles: {}\n")
-    resolved = str(yaml_path)  # absolute path, as the worktree runner passes
-
-    seen = {"agent_file": None}
-
-    class _FakeGenerator:
-        def __init__(self, agent_file, *a, **k):
-            seen["agent_file"] = agent_file
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *exc):
-            return False
-
-        def generate_crew_and_kickoff(self):
-            return "READY"
-
-    monkeypatch.setattr(pa, "_get_agents_generator", lambda: _FakeGenerator)
-    monkeypatch.setenv("PRAISONAI_IN_MODERN_RUN", "1")
-
-    import sys as _sys
-
-    # sys.argv carries the *relative* CLI token; a bare "agents.yaml" that is not
-    # present in the (simulated) new cwd. The resolved absolute path handed to the
-    # constructor must win.
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setattr(_sys, "argv", ["praisonai", "run", "agents.yaml"])
-    monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
-
-    result = pa.PraisonAI(agent_file=resolved).main()
-
-    assert seen["agent_file"] == resolved
-    assert result == "READY"
-
-
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])

src/praisonai-ts/SIGNATURE_PARITY.md

@@ -46,7 +46,7 @@ This complements `PARITY.md`, which only tracks whether an export exists.
 
 ### `Agent.__init__`
 
-- Python: `src/praisonai-agents/praisonaiagents/agent/agent.py:618`
+- Python: `src/praisonai-agents/praisonaiagents/agent/agent.py:621`
 - TypeScript: `src/praisonai-ts/src/agent/simple.ts:196` (ctor `src/praisonai-ts/src/agent/simple.ts:1018`)
 - Counts: 42 python params: 29 exact, 8 camelCase, 2 alias, 3 flattened, 0 missing; 8 mismatches; 8 waived; 14 TS-only of 60
 
@@ -219,7 +219,7 @@ TS-only members: `previousResult`?, `onToken`?, `signal`?, `onEvent`?, `options`
 
 ### `Agent.chat`
 
-- Python: `src/praisonai-agents/praisonaiagents/agent/chat_mixin.py:3133`
+- Python: `src/praisonai-agents/praisonaiagents/agent/chat_mixin.py:3096`
 - TypeScript: `src/praisonai-ts/src/agent/simple.ts:3588`
 - Counts: 17 python params: 7 exact, 9 camelCase, 1 alias, 0 flattened, 0 missing; 0 mismatches; 0 waived; 2 TS-only of 19
 
@@ -315,7 +315,7 @@ TS-only members: `config`
 
 ### `Session.__init__`
 
-- Python: `src/praisonai-agents/praisonaiagents/session/api.py:67`
+- Python: `src/praisonai-agents/praisonaiagents/session/api.py:66`
 - TypeScript: `src/praisonai-ts/src/session/session.ts:29` (ctor `src/praisonai-ts/src/session/session.ts:132`)
 - Counts: 7 python params: 1 exact, 6 camelCase, 0 alias, 0 flattened, 0 missing; 1 mismatches; 1 waived; 5 TS-only of 12
 

src/praisonai-ts/signature-parity.json

@@ -1185,7 +1185,7 @@
         }
       ],
       "python": {
-        "location": "src/praisonai-agents/praisonaiagents/agent/agent.py:618",
+        "location": "src/praisonai-agents/praisonaiagents/agent/agent.py:621",
         "resolved_class": "Agent"
       },
       "ts_only": [
@@ -1797,7 +1797,7 @@
         }
       ],
       "python": {
-        "location": "src/praisonai-agents/praisonaiagents/agent/chat_mixin.py:3133",
+        "location": "src/praisonai-agents/praisonaiagents/agent/chat_mixin.py:3096",
         "resolved_class": "ChatMixin"
       },
       "ts_only": [
@@ -5082,7 +5082,7 @@
         }
       ],
       "python": {
-        "location": "src/praisonai-agents/praisonaiagents/session/api.py:67",
+        "location": "src/praisonai-agents/praisonaiagents/session/api.py:66",
         "resolved_class": "Session"
       },
       "ts_only": [

src/praisonai/praisonai/_api_auth.py

@@ -15,11 +15,8 @@ def build_api_key_middleware(api_key: str, public_paths: Optional[Iterable[str]]
         api_key: The expected API key; requests must present it via a
             ``Bearer`` ``Authorization`` header or ``X-API-Key`` header.
         public_paths: Paths that bypass authentication. Any request whose path
-            is in this set, or is exactly ``/__praisonai__/discovery`` (the
-            unauthenticated capability document), is allowed through without a
-            token. The discovery path is matched exactly -- never by prefix --
-            so a server whose ``api_prefix`` happens to sit under
-            ``/__praisonai__`` does not silently expose its routes.
+            is in this set, or starts with ``/__praisonai__/``, is allowed
+            through without a token.
 
     Imports of ``hmac``/``starlette`` are kept inside this function so callers
     stay lazy-import friendly.
@@ -29,12 +26,11 @@ def build_api_key_middleware(api_key: str, public_paths: Optional[Iterable[str]]
     from starlette.responses import JSONResponse
 
     public = set(public_paths or ())
-    public.add("/__praisonai__/discovery")
 
     class _APIKeyMiddleware(BaseHTTPMiddleware):
         async def dispatch(self, request, call_next):
             path = request.url.path
-            if path in public:
+            if path in public or path.startswith("/__praisonai__/"):
                 return await call_next(request)
             auth = request.headers.get("Authorization", "")
             header_key = request.headers.get("X-API-Key", "")

src/praisonai/praisonai/api/agent_invoke.py

@@ -697,11 +697,7 @@ if FASTAPI_AVAILABLE and APIRouter is not None:
         if hasattr(agent, 'name'):
             info["name"] = agent.name
         if hasattr(agent, 'instructions'):
-            instructions = agent.instructions
-            if isinstance(instructions, str) and len(instructions) > 200:
-                info["instructions"] = instructions[:200] + "..."
-            else:
-                info["instructions"] = instructions
+            info["instructions"] = agent.instructions[:200] + "..." if len(agent.instructions) > 200 else agent.instructions
         if hasattr(agent, 'tools') and agent.tools:
             info["tools"] = [getattr(tool, 'name', str(tool)) for tool in agent.tools[:5]]  # First 5 tools
             if len(agent.tools) > 5:

src/praisonai/praisonai/app/agentos.py

@@ -128,8 +128,22 @@ class AgentOS:
 
         launch_token = self.config.api_key or os.environ.get("PRAISONAI_AGENTOS_API_KEY")
         if launch_token:
-            from .._api_auth import build_api_key_middleware
-            app.add_middleware(build_api_key_middleware(launch_token, {"/health", "/"}))
+            from starlette.middleware.base import BaseHTTPMiddleware
+            from starlette.responses import JSONResponse
+
+            expected = launch_token
+
+            class _AgentOSAuthMiddleware(BaseHTTPMiddleware):
+                async def dispatch(self, request, call_next):
+                    if request.url.path in ("/health", "/"):
+                        return await call_next(request)
+                    auth = request.headers.get("Authorization", "")
+                    token = auth[7:] if auth.startswith("Bearer ") else request.headers.get("X-API-Key", "")
+                    if token != expected:
+                        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+                    return await call_next(request)
+
+            app.add_middleware(_AgentOSAuthMiddleware)
         
         return app
     

src/praisonai/praisonai/auto.py

@@ -16,7 +16,6 @@ import json
 import re
 import asyncio
 import threading
-import warnings
 from praisonai._logging import get_logger
 
 # Type variable for Pydantic models - will be bound at runtime
@@ -1047,9 +1046,7 @@ class AutoGenerator(BaseAutoGenerator):
                 generate → run round-trip agrees on defaults.
             config_list: Optional LLM configuration
             pattern: Workflow pattern (sequential, parallel, routing, orchestrator-workers, evaluator-optimizer)
-            single_agent: Deprecated and inert. Single-agent generation is not
-                implemented; a team structure is always generated. Passing a
-                truthy value emits a ``DeprecationWarning``.
+            single_agent: If True, generate a single agent instead of a team
         
         Note: autogen framework is different from this AutoGenerator class.
         """
@@ -1080,12 +1077,6 @@ class AutoGenerator(BaseAutoGenerator):
         # Authoritative framework name comes from the resolved adapter.
         self.framework = adapter.name
         self.pattern = pattern
-        if single_agent:
-            warnings.warn(
-                "single_agent generation is not implemented; a team structure "
-                "is generated instead. This parameter is deprecated and inert.",
-                DeprecationWarning, stacklevel=2,
-            )
         self.single_agent = single_agent
     
 
@@ -1647,9 +1638,7 @@ class WorkflowAutoGenerator(BaseAutoGenerator):
                 the generate → run round-trip agrees on defaults. The adapter
                 is validated here rather than deferring the failure to a later
                 ``praisonai run`` invocation.
-            single_agent: Deprecated and inert. Single-agent workflow generation
-                is not implemented; a workflow structure is always generated.
-                Passing a truthy value emits a ``DeprecationWarning``.
+            single_agent: If True, generate a single agent workflow
             workspace: Optional workspace root; when set, ``workflow_file`` is
                 contained to it to reject path traversal (see BaseAutoGenerator).
         """
@@ -1668,13 +1657,6 @@ class WorkflowAutoGenerator(BaseAutoGenerator):
         self.topic = topic
         self.workflow_file = self._safe_join(workflow_file)
         self.framework = adapter.name
-        if single_agent:
-            warnings.warn(
-                "single_agent generation is not implemented; a workflow "
-                "structure is generated instead. This parameter is deprecated "
-                "and inert.",
-                DeprecationWarning, stacklevel=2,
-            )
         self.single_agent = single_agent
     
     def recommend_pattern_llm(self, topic: Optional[str] = None) -> Any:

src/praisonai/praisonai/framework_adapters/autogen_adapter.py

@@ -97,31 +97,6 @@ class AutoGenAdapter(BaseFrameworkAdapter):
             f"Install with 'pip install praisonai-frameworks[autogen]' for v0.2 or 'pip install praisonai-frameworks[autogen-v4]' for v0.4"
         )
 
-    @staticmethod
-    def _fire_task_callback(task_callback: Callable, payload: Dict[str, Any]) -> None:
-        """Invoke ``task_callback`` safely, honouring sync *and* async callbacks.
-
-        AutoGen v0.2's ``run()`` is sync-only, but the adapter contract lets a
-        caller hand us an ``async def`` callback (e.g. an observability sink that
-        awaits a network write). A bare ``task_callback(payload)`` would return
-        an un-awaited coroutine that never runs, so route the result through
-        :func:`dispatch_maybe_awaitable` which drains a returned coroutine (via
-        the shared bridge when inside a loop, or blocking when not). Callback
-        exceptions are caught and logged, never propagated into the chat.
-        """
-        from .._async_bridge import dispatch_maybe_awaitable, DispatchKind
-
-        try:
-            result = task_callback(payload)
-            dispatch_maybe_awaitable(
-                result, kind=DispatchKind.WRITE, op_name="autogen-task-callback"
-            )
-        except Exception as e:  # noqa: BLE001 - callbacks must never break the chat
-            logger.warning(
-                "task_callback raised for event %r: %s",
-                payload.get("event"), e,
-            )
-
     @staticmethod
     def _wrap_tool_for_execution(tool: Callable, tool_name: str) -> Callable:
         """Translate a per-call tool timeout into an LLM-readable string.
@@ -326,31 +301,10 @@ class AutoGenAdapter(BaseFrameworkAdapter):
                     "message": task_spec.description,
                     "summary_method": "last_msg",
                 }
-                # Honour the adapter contract's task_callback hook. AutoGen v0.2
-                # has no first-class per-step task hook, so fire a prepared event
-                # here (mirroring the retrofitted agent_callback above) and a
-                # completed event after initiate_chats below.
-                if task_callback:
-                    self._fire_task_callback(task_callback, {
-                        "event": "task_prepared",
-                        "task_spec": task_spec,
-                        "agent": agents[spec.key],
-                    })
                 tasks.append(chat_task)
         
         # Execute tasks
         response = user_proxy.initiate_chats(tasks)
-
-        # Fan out task completion events now that AutoGen has run the chats.
-        if task_callback:
-            for chat_task, resp in zip(tasks, response or []):
-                self._fire_task_callback(task_callback, {
-                    "event": "task_completed",
-                    "task": chat_task,
-                    "response": resp,
-                    "summary": getattr(resp, "summary", None),
-                })
-
         result = "### AutoGen v0.2 Output ###\n" + (response[-1].summary if hasattr(response[-1], 'summary') else "")
         
         logger.info("AutoGen v0.2 execution completed")

src/praisonai/praisonai/integration/__init__.py

@@ -7,6 +7,7 @@ from .host_app import (
     is_legacy_host,
     setup_bridges,
 )
+from .gateway_host import run_integrated_gateway
 
 __all__ = [
     "build_host_app",
@@ -16,11 +17,3 @@ __all__ = [
     "setup_bridges",
     "run_integrated_gateway",
 ]
-
-
-def __getattr__(name):
-    if name == "run_integrated_gateway":
-        from .gateway_host import run_integrated_gateway
-
-        return run_integrated_gateway
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/praisonai/praisonai/integrations/registry.py

@@ -115,17 +115,12 @@ class ExternalAgentRegistry(PluginRegistry[BaseCLIIntegration]):
         import inspect
         availability = {}
         
-        # Snapshot the registered loader names (the authoritative set). The
-        # resolved-item cache (_items) is only populated as a side effect of
-        # resolve(), so iterating it would return {} on a fresh registry.
+        # Get snapshot of all items from parent class
         with self._lock:
-            names = list(self._loaders.keys())
+            snapshot = list(self._items.items())
         
-        for name in names:
+        for name, integration_class in snapshot:
             try:
-                # Resolve lazily; this also populates the _items cache.
-                integration_class = self.resolve(name)
-                
                 # Check if constructor requires parameters beyond self
                 sig = inspect.signature(integration_class.__init__)
                 params = [p for p_name, p in sig.parameters.items() if p_name != 'self' and p.default is inspect.Parameter.empty]

src/praisonai/praisonai/persistence/__init__.py

@@ -22,10 +22,8 @@ __all__ = [
     "ConversationSession",
     "ConversationMessage",
     "KnowledgeStore",
-    "AsyncKnowledgeStore",
     "KnowledgeDocument",
     "StateStore",
-    "AsyncStateStore",
     # Orchestrator
     "PersistenceOrchestrator",
     # Factory functions
@@ -50,13 +48,13 @@ def __getattr__(name: str):
         from .conversation.base import ConversationStore, ConversationSession, ConversationMessage
         return locals()[name]
     
-    if name in ("KnowledgeStore", "AsyncKnowledgeStore", "KnowledgeDocument"):
-        from .knowledge.base import KnowledgeStore, AsyncKnowledgeStore, KnowledgeDocument
+    if name in ("KnowledgeStore", "KnowledgeDocument"):
+        from .knowledge.base import KnowledgeStore, KnowledgeDocument
         return locals()[name]
     
-    if name in ("StateStore", "AsyncStateStore"):
-        from .state.base import StateStore, AsyncStateStore
-        return locals()[name]
+    if name == "StateStore":
+        from .state.base import StateStore
+        return StateStore
     
     if name == "PersistenceOrchestrator":
         from .orchestrator import PersistenceOrchestrator

src/praisonai/praisonai/persistence/knowledge/__init__.py

@@ -16,12 +16,11 @@ Supported backends:
 
 __all__ = [
     "KnowledgeStore",
-    "AsyncKnowledgeStore",
     "KnowledgeDocument",
 ]
 
 def __getattr__(name: str):
-    if name in ("KnowledgeStore", "AsyncKnowledgeStore", "KnowledgeDocument"):
-        from .base import KnowledgeStore, AsyncKnowledgeStore, KnowledgeDocument
+    if name in ("KnowledgeStore", "KnowledgeDocument"):
+        from .base import KnowledgeStore, KnowledgeDocument
         return locals()[name]
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/praisonai/praisonai/persistence/knowledge/base.py

@@ -174,108 +174,3 @@ class KnowledgeStore(ABC):
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
         return False
-
-
-class AsyncKnowledgeStore(ABC):
-    """
-    Abstract base class for async vector/knowledge persistence.
-
-    Async sibling of :class:`KnowledgeStore`, matching the
-    :class:`~praisonai.persistence.conversation.base.AsyncConversationStore`
-    parity so third-party async vector backends have a formal interface to
-    subclass instead of implementing the sync ABC behind ``run_sync`` shims.
-    Async stores should implement this instead of ``KnowledgeStore`` when they
-    are natively async.
-    """
-
-    @abstractmethod
-    async def create_collection(
-        self,
-        name: str,
-        dimension: int,
-        distance: str = "cosine",
-        metadata: Optional[Dict[str, Any]] = None,
-    ) -> None:
-        """Create a new collection/index for vectors."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def delete_collection(self, name: str) -> bool:
-        """Delete a collection and all its documents."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def collection_exists(self, name: str) -> bool:
-        """Check if a collection exists."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def list_collections(self) -> List[str]:
-        """List all collections."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def insert(
-        self,
-        collection: str,
-        documents: List[KnowledgeDocument],
-    ) -> List[str]:
-        """Insert documents into a collection. Returns list of IDs."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def upsert(
-        self,
-        collection: str,
-        documents: List[KnowledgeDocument],
-    ) -> List[str]:
-        """Insert or update documents. Returns list of IDs."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def search(
-        self,
-        collection: str,
-        query_embedding: List[float],
-        limit: int = 5,
-        filters: Optional[Dict[str, Any]] = None,
-        score_threshold: Optional[float] = None,
-    ) -> List[KnowledgeDocument]:
-        """Search for similar documents by embedding."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def get(
-        self,
-        collection: str,
-        ids: List[str],
-    ) -> List[KnowledgeDocument]:
-        """Get documents by IDs."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def delete(
-        self,
-        collection: str,
-        ids: Optional[List[str]] = None,
-        filters: Optional[Dict[str, Any]] = None,
-    ) -> int:
-        """Delete documents by IDs or filters. Returns count deleted."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def count(self, collection: str) -> int:
-        """Count documents in a collection."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def close(self) -> None:
-        """Close the store and release resources."""
-        raise NotImplementedError
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self.close()
-        return False

src/praisonai/praisonai/persistence/state/__init__.py

@@ -12,11 +12,10 @@ Supported backends:
 
 __all__ = [
     "StateStore",
-    "AsyncStateStore",
 ]
 
 def __getattr__(name: str):
-    if name in ("StateStore", "AsyncStateStore"):
-        from .base import StateStore, AsyncStateStore
-        return locals()[name]
+    if name == "StateStore":
+        from .base import StateStore
+        return StateStore
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/praisonai/praisonai/persistence/state/base.py

@@ -115,101 +115,3 @@ class StateStore(ABC):
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
         return False
-
-
-class AsyncStateStore(ABC):
-    """
-    Abstract base class for async key-value state persistence.
-
-    Async sibling of :class:`StateStore`, matching the
-    :class:`~praisonai.persistence.conversation.base.AsyncConversationStore`
-    parity so third-party async backends (e.g. motor-backed MongoDB) have a
-    formal interface to subclass instead of implementing the sync ABC behind
-    ``run_sync`` shims. Async stores should implement this instead of
-    ``StateStore`` when they are natively async.
-    """
-
-    @abstractmethod
-    async def get(self, key: str) -> Optional[Any]:
-        """Get a value by key. Returns None if not found."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def set(
-        self,
-        key: str,
-        value: Any,
-        ttl: Optional[int] = None,
-    ) -> None:
-        """Set a value. TTL is in seconds."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def delete(self, key: str) -> bool:
-        """Delete a key. Returns True if key existed."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def exists(self, key: str) -> bool:
-        """Check if a key exists."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def keys(self, pattern: str = "*") -> List[str]:
-        """List keys matching pattern."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def ttl(self, key: str) -> Optional[int]:
-        """Get remaining TTL in seconds. Returns None if no TTL or key doesn't exist."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def expire(self, key: str, ttl: int) -> bool:
-        """Set TTL on existing key. Returns True if key exists."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def hget(self, key: str, field: str) -> Optional[Any]:
-        """Get a field from a hash."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def hset(self, key: str, field: str, value: Any) -> None:
-        """Set a field in a hash."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def hgetall(self, key: str) -> Dict[str, Any]:
-        """Get all fields from a hash."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def hdel(self, key: str, *fields: str) -> int:
-        """Delete fields from a hash. Returns count deleted."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def close(self) -> None:
-        """Close the store and release resources."""
-        raise NotImplementedError
-
-    async def get_json(self, key: str) -> Optional[Any]:
-        """Get and deserialize JSON value."""
-        value = await self.get(key)
-        if value is None:
-            return None
-        if isinstance(value, str):
-            return json.loads(value)
-        return value
-
-    async def set_json(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
-        """Serialize and set JSON value."""
-        await self.set(key, json.dumps(value), ttl)
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self.close()
-        return False

src/praisonai/praisonai/security/__init__.py

@@ -22,7 +22,7 @@ Usage (selective):
     enable_audit_log()
 """
 
-from typing import TYPE_CHECKING, Dict, Optional, List
+from typing import TYPE_CHECKING, Optional, List
 
 if TYPE_CHECKING:
     from .injection import InjectionDefense, ScanResult, ThreatLevel
@@ -86,7 +86,7 @@ def enable_injection_defense(
     extra_patterns: Optional[List[str]] = None,
     block_threshold: Optional[int] = None,
     trusted_sources: Optional[List[str]] = None,
-) -> Dict[str, str]:
+) -> str:
     """
     Enable prompt injection defense globally for all agents.
 

src/praisonai/tests/unit/integrations/test_integration_lazy_gateway.py

@@ -1,99 +0,0 @@
-"""Regression tests for lazy ``gateway_host`` loading in ``praisonai.integration``.
-
-Importing ``praisonai.integration`` must NOT force-load the
-``praisonai.integration.gateway_host`` stack — only ``host_app`` symbols are
-eager (what UI/dashboard presets need). ``run_integrated_gateway`` resolves
-lazily on first attribute access via the package ``__getattr__``.
-
-Guards against a future eager import silently undoing the startup improvement.
-"""
-import importlib
-import sys
-
-import pytest
-
-_GATEWAY_HOST = "praisonai.integration.gateway_host"
-_PACKAGE = "praisonai.integration"
-_HOST_APP = "praisonai.integration.host_app"
-
-
-@pytest.fixture
-def reload_integration():
-    """Reload ``praisonai.integration`` cleanly, restoring global state after.
-
-    The tests here poke at import-time behaviour by dropping cached modules,
-    so we snapshot and restore the affected ``sys.modules`` entries to avoid
-    leaking a partially-loaded package into unrelated tests (which would break
-    ``monkeypatch.setattr`` path resolution elsewhere).
-    """
-    tracked = (_GATEWAY_HOST, _PACKAGE, _HOST_APP)
-    saved = {name: sys.modules.get(name) for name in tracked}
-
-    def _reload():
-        for name in (_GATEWAY_HOST, _HOST_APP, _PACKAGE):
-            sys.modules.pop(name, None)
-        return importlib.import_module(_PACKAGE)
-
-    try:
-        yield _reload
-    finally:
-        for name, module in saved.items():
-            if module is not None:
-                sys.modules[name] = module
-            else:
-                sys.modules.pop(name, None)
-        importlib.import_module(_PACKAGE)
-
-
-def test_import_does_not_load_gateway_host(reload_integration):
-    """Importing the package leaves gateway_host unloaded."""
-    reload_integration()
-    assert _GATEWAY_HOST not in sys.modules
-
-
-def test_host_app_symbols_are_eager(reload_integration):
-    """host_app exports remain importable without touching gateway_host."""
-    pkg = reload_integration()
-    for symbol in (
-        "build_host_app",
-        "configure_host",
-        "create_host_app",
-        "is_legacy_host",
-        "setup_bridges",
-    ):
-        assert hasattr(pkg, symbol)
-    assert _GATEWAY_HOST not in sys.modules
-
-
-def test_run_integrated_gateway_loads_on_access(reload_integration):
-    """Accessing run_integrated_gateway lazily loads gateway_host on first use."""
-    pkg = reload_integration()
-    assert _GATEWAY_HOST not in sys.modules
-
-    resolved = pkg.run_integrated_gateway
-    assert resolved is not None
-    assert _GATEWAY_HOST in sys.modules
-
-    from praisonai.integration.gateway_host import run_integrated_gateway
-
-    assert resolved is run_integrated_gateway
-
-
-def test_unknown_attribute_raises_attribute_error(reload_integration):
-    """__getattr__ preserves normal AttributeError for unknown names."""
-    pkg = reload_integration()
-    with pytest.raises(AttributeError):
-        pkg.does_not_exist
-
-
-def test_all_exports_unchanged(reload_integration):
-    """__all__ still advertises the full public surface."""
-    pkg = reload_integration()
-    assert set(pkg.__all__) == {
-        "build_host_app",
-        "configure_host",
-        "create_host_app",
-        "is_legacy_host",
-        "setup_bridges",
-        "run_integrated_gateway",
-    }

src/praisonai/tests/unit/test_auto_generator.py

@@ -707,50 +707,6 @@ class TestSingleAgentOption:
             )
             assert generator.single_agent == True
 
-    def test_auto_generator_single_agent_true_warns(self):
-        """AutoGenerator emits a DeprecationWarning when single_agent is truthy."""
-        import warnings
-        with patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'}):
-            with pytest.warns(DeprecationWarning, match="deprecated and inert"):
-                AutoGenerator(
-                    topic="Write a haiku",
-                    framework="praisonai",
-                    single_agent=True,
-                )
-
-    def test_auto_generator_single_agent_false_no_warning(self):
-        """AutoGenerator does not warn when single_agent is left False."""
-        import warnings
-        with patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'}):
-            with warnings.catch_warnings():
-                warnings.simplefilter("error", DeprecationWarning)
-                generator = AutoGenerator(
-                    topic="Write a haiku",
-                    framework="praisonai",
-                )
-                assert generator.single_agent is False
-
-    def test_workflow_generator_single_agent_true_warns(self):
-        """WorkflowAutoGenerator emits a DeprecationWarning when single_agent is truthy."""
-        import warnings
-        with patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'}):
-            with pytest.warns(DeprecationWarning, match="deprecated and inert"):
-                WorkflowAutoGenerator(
-                    topic="Write a haiku",
-                    single_agent=True,
-                )
-
-    def test_workflow_generator_single_agent_false_no_warning(self):
-        """WorkflowAutoGenerator does not warn when single_agent is left False."""
-        import warnings
-        with patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'}):
-            with warnings.catch_warnings():
-                warnings.simplefilter("error", DeprecationWarning)
-                generator = WorkflowAutoGenerator(
-                    topic="Write a haiku",
-                )
-                assert generator.single_agent is False
-
 
 # =============================================================================
 # TODO 3: LLM-Based Pattern Recommendation Tests

src/praisonai/tests/unit/test_autogen_task_callback.py

@@ -1,57 +0,0 @@
-"""
-Focused unit tests for AutoGenAdapter task_callback dispatch.
-
-These exercise ``AutoGenAdapter._fire_task_callback`` directly, so they need no
-live AutoGen chat service or ``autogen`` install. They guard the contract that:
-
-- sync callbacks receive the event payload,
-- async (coroutine) callbacks are actually run (not left un-awaited), and
-- a raising callback is swallowed and logged, never propagated into the chat.
-"""
-
-import unittest
-
-
-class TestAutoGenFireTaskCallback(unittest.TestCase):
-    """Guard sync/async execution and exception isolation of task_callback."""
-
-    def setUp(self):
-        from praisonai.framework_adapters.autogen_adapter import AutoGenAdapter
-        self.adapter = AutoGenAdapter()
-
-    def test_sync_callback_receives_payload(self):
-        seen = []
-        payload = {"event": "task_prepared", "task_spec": "x"}
-
-        self.adapter._fire_task_callback(seen.append, payload)
-
-        self.assertEqual(seen, [payload])
-
-    def test_async_callback_is_executed(self):
-        seen = []
-
-        async def cb(payload):
-            seen.append(payload)
-
-        payload = {"event": "task_completed", "summary": "done"}
-        self.adapter._fire_task_callback(cb, payload)
-
-        # The coroutine must have actually run (not returned un-awaited).
-        self.assertEqual(seen, [payload])
-
-    def test_sync_callback_exception_is_swallowed(self):
-        def boom(_payload):
-            raise ValueError("boom")
-
-        # Must not propagate — a broken observability sink can't abort the chat.
-        self.adapter._fire_task_callback(boom, {"event": "task_prepared"})
-
-    def test_async_callback_exception_is_swallowed(self):
-        async def boom(_payload):
-            raise ValueError("boom")
-
-        self.adapter._fire_task_callback(boom, {"event": "task_completed"})
-
-
-if __name__ == "__main__":
-    unittest.main()

src/praisonai/tests/unit/test_chat_approval_authorization.py

@@ -351,208 +351,3 @@ class TestDiscordAuthorization:
         backend._discord_api = mock_api
         decision = asyncio.run(backend.request_approval(_make_request()))
         assert decision.approved is True
-
-
-# ── HTTP dashboard (out-of-band) ─────────────────────────────────────────────
-
-
-class TestHTTPAuthorization:
-    def _backend(self, **kwargs):
-        from praisonai.bots._http_approval import HTTPApproval
-
-        backend = HTTPApproval(port=0, timeout=1, **kwargs)
-
-        async def mock_server():
-            backend._server_started = True
-
-        backend._ensure_server = mock_server
-        return backend
-
-    def test_no_allowlist_mints_no_token(self):
-        """Legacy: no allowlist → no token on the pending approval."""
-        backend = self._backend()
-
-        async def run():
-            async def simulate():
-                await asyncio.sleep(0.2)
-                for info in backend._pending.values():
-                    assert info.get("token") is None
-                    info["decided"] = True
-                    info["approved"] = True
-                    info["approver"] = "anyone"
-                    break
-
-            task = asyncio.create_task(simulate())
-            decision = await backend.request_approval(_make_request())
-            await task
-            return decision
-
-        decision = asyncio.run(run())
-        assert decision.approved is True
-
-    def test_allowlist_mints_token(self):
-        """With an allowlist a single-use token is bound to the approval."""
-        backend = self._backend(allowed_approvers=["owner"])
-
-        async def run():
-            async def simulate():
-                await asyncio.sleep(0.2)
-                for info in backend._pending.values():
-                    assert info.get("token")
-                    info["decided"] = True
-                    info["approved"] = True
-                    info["approver"] = "owner"
-                    break
-
-            task = asyncio.create_task(simulate())
-            decision = await backend.request_approval(_make_request())
-            await task
-            return decision
-
-        decision = asyncio.run(run())
-        assert decision.approved is True
-
-    def test_decide_without_token_rejected(self):
-        """A POST lacking the per-approval token is refused (403)."""
-        backend = self._backend(allowed_approvers=["owner"])
-        backend._pending["r1"] = {
-            "decided": False, "approved": False, "token": "secret-tok", "info": {},
-        }
-
-        class _Req:
-            match_info = {"request_id": "r1"}
-            query: dict = {}
-            headers: dict = {}
-
-            async def json(self):
-                return {"decision": "approve", "approver": "owner"}
-
-        resp = asyncio.run(backend._handle_decide(_Req()))
-        assert resp.status == 403
-        assert backend._pending["r1"]["decided"] is False
-
-    def test_decide_with_token_but_unauthorized_approver_rejected(self):
-        """Correct token but approver not in allowlist is refused."""
-        backend = self._backend(allowed_approvers=["owner"])
-        backend._pending["r1"] = {
-            "decided": False, "approved": False, "token": "secret-tok", "info": {},
-        }
-
-        class _Req:
-            match_info = {"request_id": "r1"}
-            query = {"token": "secret-tok"}
-            headers: dict = {}
-
-            async def json(self):
-                return {"decision": "approve", "approver": "intruder"}
-
-        resp = asyncio.run(backend._handle_decide(_Req()))
-        assert resp.status == 403
-        assert backend._pending["r1"]["decided"] is False
-
-    def test_decide_with_token_and_authorized_approver_resolves(self):
-        backend = self._backend(allowed_approvers=["owner"])
-        backend._pending["r1"] = {
-            "decided": False, "approved": False, "token": "secret-tok", "info": {},
-        }
-
-        class _Req:
-            match_info = {"request_id": "r1"}
-            query = {"token": "secret-tok"}
-            headers: dict = {}
-
-            async def json(self):
-                return {"decision": "approve", "approver": "owner"}
-
-        resp = asyncio.run(backend._handle_decide(_Req()))
-        assert resp.status == 200
-        assert backend._pending["r1"]["decided"] is True
-        assert backend._pending["r1"]["approved"] is True
-        assert backend._pending["r1"]["approver"] == "owner"
-
-
-# ── Webhook (out-of-band) ────────────────────────────────────────────────────
-
-
-class TestWebhookAuthorization:
-    def _backend(self, **kwargs):
-        from praisonai.bots._webhook_approval import WebhookApproval
-
-        return WebhookApproval(
-            webhook_url="https://example.com", timeout=1, poll_interval=0.05, **kwargs
-        )
-
-    def test_unauthorized_immediate_decision_dropped(self):
-        """Immediate POST approval from a non-allowlisted approver is dropped."""
-        backend = self._backend(allowed_approvers=["owner"])
-
-        async def mock_http(method, url, payload=None, **kwargs):
-            if method == "POST":
-                return {"approved": True, "approver": "intruder"}
-            return {"status": "pending"}
-
-        backend._http_request = mock_http
-        decision = asyncio.run(backend.request_approval(_make_request()))
-        assert decision.approved is False
-        assert "timed out" in decision.reason.lower()
-
-    def test_authorized_immediate_decision_resolves(self):
-        backend = self._backend(allowed_approvers=["owner"])
-
-        async def mock_http(method, url, payload=None, **kwargs):
-            if method == "POST":
-                return {"approved": True, "approver": "owner"}
-            return {"status": "pending"}
-
-        backend._http_request = mock_http
-        decision = asyncio.run(backend.request_approval(_make_request()))
-        assert decision.approved is True
-        assert decision.approver == "owner"
-
-    def test_unauthorized_poll_decision_dropped(self):
-        backend = self._backend(allowed_approvers=["owner"])
-
-        async def mock_http(method, url, payload=None, **kwargs):
-            if method == "POST":
-                return {"status": "pending"}
-            if method == "GET":
-                return {"status": "approved", "approver": "intruder"}
-            return {}
-
-        backend._http_request = mock_http
-        decision = asyncio.run(backend.request_approval(_make_request()))
-        assert decision.approved is False
-        assert "timed out" in decision.reason.lower()
-
-    def test_authorized_poll_decision_resolves(self):
-        backend = self._backend(allowed_approvers=["owner"])
-
-        async def mock_http(method, url, payload=None, **kwargs):
-            if method == "POST":
-                return {"status": "pending"}
-            if method == "GET":
-                return {"approved": True, "approver": "owner"}
-            return {}
-
-        backend._http_request = mock_http
-        decision = asyncio.run(backend.request_approval(_make_request()))
-        assert decision.approved is True
-        assert decision.approver == "owner"
-
-    def test_no_allowlist_permits_any_approver(self):
-        """Backward compatible: without an allowlist, any approver resolves."""
-        backend = self._backend()
-
-        async def mock_http(method, url, payload=None, **kwargs):
-            if method == "POST":
-                return {"approved": True, "approver": "anyone"}
-            return {}
-
-        backend._http_request = mock_http
-        decision = asyncio.run(backend.request_approval(_make_request()))
-        assert decision.approved is True
-
-    def test_env_var_allowlist(self, monkeypatch):
-        monkeypatch.setenv("WEBHOOK_APPROVERS", "owner, boss")
-        backend = self._backend()
-        assert backend._allowed_approvers == {"owner", "boss"}