fix: address valkey integration issues from code review

- Fix JSON encoding consistency in ValkeyStateStore to prevent type confusion
- Add TTL guards for ExpirySet/ExpiryType to prevent cryptic RuntimeError
- Add warning for score threshold limitation (cosine distance only)
- Fix metadata persistence in KnowledgeDocument (now properly round-tripped)
- Improve exception handling - only catch specific 'index not found' errors
- Fix URL parameter handling in registry factories (reject with clear error)
- Add input validation for vector dimensions and k parameter

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

.env.example

@@ -37,6 +37,14 @@ WHATSAPP_PHONE_NUMBER_ID=...   # WhatsApp phone number ID
 # === Optional: Database ===
 DATABASE_URL=sqlite:///~/.praison/database.sqlite  # Database connection string
 
+# === Optional: Tool Security ===
+# ALLOWED_TOOLS=search,send_message       # Whitelist specific tools to prevent name collisions
+                                        # Useful in multi-environment setups with overlapping tool names
+                                        # Leave unset to use all tools (with collision warnings)
+                                        # Do NOT set to empty string: that's treated as configuration error
+                                        # In CI, unknown tool names in the whitelist cause strict startup failure
+# HERMES_ONLY_TOOLS=search,send_message  # (Deprecated) Backward compatibility alias for ALLOWED_TOOLS
+
 # === Optional: Development ===
 LOGLEVEL=INFO                  # Logging level (DEBUG, INFO, WARNING, ERROR)
 CHAINLIT_AUTH_SECRET=...       # Chainlit authentication secret

api.md

@@ -183,6 +183,7 @@ Methods:
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_available_tools</a>(context: Optional[Dict[str, Any]] = None) -> List[Union[BaseTool, Callable]]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_base_tools</a>() -> List[BaseTool]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools</a>() -> List[str]</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools_with_allowed_filter</a>(context: Optional[Dict[str, Any]] = None) -> List[str]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, overwrite: bool = False) -> None</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">unregister</a>(name: str) -> bool</code>
 * <code title="class Tools">Tools.<a href="./src/praisonai-agents/praisonaiagents/tools/tools.py">internet_search</a>(*args, **kwargs)</code>

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

@@ -753,11 +753,11 @@ Your Goal: {self.goal}"""
             
             # Create LLMError with contextual metadata
             error = LLMError(
-                str(e), 
-                model_name=model_name, 
-                agent_id=self.name, 
+                str(e),
+                model_name=model_name,
+                agent_id=self.name,
                 is_retryable=is_retryable,
-                session_id=session_id
+                context={"session_id": session_id},
             )
             
             # Call error hook if available for error interception
@@ -902,7 +902,7 @@ Your Goal: {self.goal}"""
                 max_tokens=getattr(self, 'max_tokens', None),
                 stream=stream,
                 response_format=response_format,
-                execute_tool_fn=getattr(self, 'execute_tool', None),
+                execute_tool_fn=getattr(self, 'execute_tool_async', None),
                 console=self.console if (self.verbose or stream) else None,
                 display_fn=self._display_generating if self.verbose else None,
                 stream_callback=stream_callback,
@@ -935,6 +935,151 @@ Your Goal: {self.goal}"""
             logging.error(f"Unified async chat completion failed: {e}")
             raise
 
+    async def _finalize_unified_achat_response(
+        self,
+        response,
+        original_prompt,
+        temperature,
+        tools,
+        output_json,
+        output_pydantic,
+        reasoning_steps,
+        stream,
+        messages,
+        chat_history_length,
+        start_time,
+        task_name=None,
+        task_description=None,
+        task_id=None,
+    ):
+        """Post-process unified achat completion (parity with sync _chat_impl)."""
+        if not response:
+            self._truncate_chat_history(chat_history_length)
+            return None
+
+        if hasattr(response, "choices") and response.choices:
+            msg = response.choices[0].message
+            content = getattr(msg, "content", None)
+            response_text = content.strip() if content else ""
+            if (
+                reasoning_steps
+                and getattr(msg, "reasoning_content", None)
+            ):
+                response_text = msg.reasoning_content.strip()
+        else:
+            extracted = self._extract_llm_response_content(response)
+            response_text = (extracted or "").strip() if isinstance(extracted, str) else str(extracted or "")
+
+        if output_json or output_pydantic:
+            self._append_to_chat_history({"role": "assistant", "content": response_text})
+            self._persist_message("assistant", response_text)
+            try:
+                validated_response = self._apply_guardrail_with_retry(
+                    response_text, original_prompt, temperature, tools,
+                    task_name, task_description, task_id,
+                )
+                self._execute_callback_and_display(
+                    original_prompt, validated_response, time.time() - start_time,
+                    task_name, task_description, task_id,
+                )
+                return await self._atrigger_after_agent_hook(
+                    original_prompt, validated_response, start_time,
+                )
+            except Exception as e:
+                logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
+                self._truncate_chat_history(chat_history_length)
+                return None
+
+        if not self.self_reflect:
+            self._append_to_chat_history({"role": "assistant", "content": response_text})
+            self._persist_message("assistant", response_text)
+            try:
+                validated_response = self._apply_guardrail_with_retry(
+                    response_text, original_prompt, temperature, tools,
+                    task_name, task_description, task_id,
+                )
+                self._execute_callback_and_display(
+                    original_prompt, validated_response, time.time() - start_time,
+                    task_name, task_description, task_id,
+                )
+                return await self._atrigger_after_agent_hook(
+                    original_prompt, validated_response, start_time,
+                )
+            except Exception as e:
+                logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
+                self._truncate_chat_history(chat_history_length)
+                return None
+
+        # Self-reflection (legacy OpenAI path)
+        reflection_count = 0
+        while True:
+            reflection_prompt = f"""
+Reflect on your previous response: '{response_text}'.
+{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
+Provide a "satisfactory" status ('yes' or 'no').
+Output MUST be JSON with 'reflection' and 'satisfactory'.
+            """
+            reflection_messages = messages + [
+                {"role": "assistant", "content": response_text},
+                {"role": "user", "content": reflection_prompt},
+            ]
+            try:
+                if self._openai_client is None:
+                    self._append_to_chat_history({"role": "assistant", "content": response_text})
+                    return await self._atrigger_after_agent_hook(
+                        original_prompt, response_text, start_time,
+                    )
+                reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
+                    model=self.reflect_llm if self.reflect_llm else self.llm,
+                    messages=reflection_messages,
+                    temperature=temperature,
+                    response_format=_get_display_functions()['ReflectionOutput'],
+                )
+                reflection_output = reflection_response.choices[0].message.parsed
+                if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
+                    break
+                if reflection_count >= self.max_reflect - 1:
+                    break
+                regenerate_messages = reflection_messages + [
+                    {
+                        "role": "assistant",
+                        "content": (
+                            f"Self Reflection: {reflection_output.reflection} "
+                            f"Satisfactory?: {reflection_output.satisfactory}"
+                        ),
+                    },
+                    {"role": "user", "content": "Now regenerate your response using the reflection you made"},
+                ]
+                new_response = await self._openai_client.async_client.chat.completions.create(
+                    model=self.llm,
+                    messages=regenerate_messages,
+                    temperature=temperature,
+                )
+                response_text = new_response.choices[0].message.content
+                reflection_count += 1
+            except Exception as e:
+                logging.error("Reflection parsing failed.", exc_info=True)
+                reflection_count += 1
+                if reflection_count >= self.max_reflect:
+                    break
+
+        try:
+            validated_response = self._apply_guardrail_with_retry(
+                response_text, original_prompt, temperature, tools,
+                task_name, task_description, task_id,
+            )
+            self._execute_callback_and_display(
+                original_prompt, validated_response, time.time() - start_time,
+                task_name, task_description, task_id,
+            )
+            return await self._atrigger_after_agent_hook(
+                original_prompt, validated_response, start_time,
+            )
+        except Exception as e:
+            logging.error(f"Agent {self.name}: Guardrail validation failed for OpenAI client: {e}")
+            self._truncate_chat_history(chat_history_length)
+            return None
+
     def _execute_callback_and_display(self, prompt: str, response: str, generation_time: float, task_name=None, task_description=None, task_id=None):
         """Helper method to execute callbacks and display interaction.
         
@@ -2042,8 +2187,22 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                             task_id=task_id,
                             response_format=response_format
                         )
-                        # Continue to handle structured outputs and other processing
-                        # Don't return immediately - fall through to existing logic
+                        return await self._finalize_unified_achat_response(
+                            response=response,
+                            original_prompt=original_prompt,
+                            temperature=temperature,
+                            tools=tools,
+                            output_json=output_json,
+                            output_pydantic=output_pydantic,
+                            reasoning_steps=reasoning_steps,
+                            stream=stream,
+                            messages=messages,
+                            chat_history_length=chat_history_length,
+                            start_time=start_time,
+                            task_name=task_name,
+                            task_description=task_description,
+                            task_id=task_id,
+                        )
                     else:
                         # LEGACY: Check if OpenAI client is available
                         if self._openai_client is None:

src/praisonai-agents/praisonaiagents/allowed_tools_filter.py

@@ -0,0 +1,257 @@
+"""
+Tool whitelist filter module with diagnostics.
+
+This module provides a canonical implementation of the ALLOWED_TOOLS filter
+that prevents tool name collisions in multi-environment agent systems by 
+whitelisting only specified tools.
+
+The filter is designed to solve the "tool shadowing" problem where multiple
+modules register tools with overlapping or ambiguous names, causing agents
+to invoke the wrong implementation.
+
+Environment Variables:
+    ALLOWED_TOOLS: Primary variable (comma-separated tool names)
+    HERMES_ONLY_TOOLS: Backward compatibility alias for ALLOWED_TOOLS
+
+Usage:
+    from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+    
+    # Create filter from environment variable  
+    filter = AllowedToolsFilter()
+    
+    # Filter tools
+    available_tools = {"search", "send_message", "extract_pdf"}
+    filtered_tools = filter.filter_tools(available_tools)
+    
+    # Get diagnostics
+    filter.log_diagnostics()
+"""
+
+import os
+import logging
+from typing import Any, Dict, List, Optional, Set, Union
+
+logger = logging.getLogger(__name__)
+
+
+class AllowedToolsFilter:
+    """
+    Canonical tool whitelist filter with diagnostics and backward compatibility.
+    
+    This filter implements the semantics for tool filtering:
+    - ALLOWED_TOOLS unset (or HERMES_ONLY_TOOLS): all tools visible (with warning about collisions)
+    - ALLOWED_TOOLS empty string: error - must specify tools or unset
+    - ALLOWED_TOOLS with values: only whitelisted tools visible
+    - Unknown tools in list: warn and strip in dev, strict fail in CI
+    
+    Backward Compatibility:
+    - Supports both ALLOWED_TOOLS and HERMES_ONLY_TOOLS
+    - ALLOWED_TOOLS takes precedence if both are set
+    """
+    
+    def __init__(self, env_var_name: str = "ALLOWED_TOOLS"):
+        """
+        Initialize the filter with backward compatibility support.
+        
+        Args:
+            env_var_name: Primary environment variable name (default: "ALLOWED_TOOLS")
+        """
+        # Support both new (ALLOWED_TOOLS) and legacy (HERMES_ONLY_TOOLS) naming
+        self.primary_var = env_var_name
+        self.legacy_var = "HERMES_ONLY_TOOLS"
+        
+        # ALLOWED_TOOLS takes precedence over HERMES_ONLY_TOOLS for backward compatibility
+        self.env_value = os.environ.get(self.primary_var) or os.environ.get(self.legacy_var)
+        self.env_var_name = self.primary_var if os.environ.get(self.primary_var) else self.legacy_var
+        
+        self.is_ci = os.environ.get("CI", "").lower() in ("true", "1", "yes")
+        self._whitelist: Optional[Set[str]] = self._parse_whitelist()
+        self._diagnostics: Dict[str, Any] = {}
+        
+    def _parse_whitelist(self) -> Optional[Set[str]]:
+        """
+        Parse the environment variable value into a whitelist set.
+        
+        Returns:
+            None if unset, Set of tool names if set, raises ValueError for empty string
+        """
+        if self.env_value is None:
+            return None
+        
+        # Empty string is an error - must be explicit
+        if self.env_value.strip() == "":
+            raise ValueError(
+                f"{self.env_var_name} cannot be empty. Either unset it for all tools "
+                "or provide a comma-separated list of tool names."
+            )
+        
+        # Parse comma-separated values, strip whitespace
+        tools = {name.strip() for name in self.env_value.split(",") if name.strip()}
+        
+        if not tools:
+            raise ValueError(
+                f"{self.env_var_name} contains no valid tool names. "
+                "Provide comma-separated tool names or unset the variable."
+            )
+        
+        return tools
+    
+    def filter_tools(self, available_tools: Union[Set[str], List[str], Dict[str, Any]]) -> Set[str]:
+        """
+        Filter tools based on ALLOWED_TOOLS whitelist.
+        
+        Args:
+            available_tools: Available tools as set, list, or dict with tool names as keys
+            
+        Returns:
+            Filtered set of tool names
+            
+        Raises:
+            ValueError: If unknown tools in CI mode or configuration errors
+        """
+        # Convert input to set of tool names
+        if isinstance(available_tools, dict):
+            tool_names = set(available_tools.keys())
+        else:
+            tool_names = set(available_tools)
+        
+        # Store for diagnostics
+        self._diagnostics["registered_before_filter"] = sorted(tool_names)
+        
+        # If no whitelist, return all tools (with warning)
+        if self._whitelist is None:
+            logger.warning(
+                "Tool whitelist is unset. All %d tools are visible. "
+                "Consider using ALLOWED_TOOLS to prevent tool name collisions.",
+                len(tool_names)
+            )
+            self._diagnostics["registered_after_filter"] = sorted(tool_names)
+            self._diagnostics["dropped_tools"] = []
+            self._diagnostics["unknown_tools"] = []
+            return tool_names
+        
+        # Find intersection and unknown tools
+        available_whitelisted = tool_names.intersection(self._whitelist)
+        unknown_tools = self._whitelist.difference(tool_names)
+        dropped_tools = tool_names.difference(self._whitelist)
+        
+        # Store for diagnostics
+        self._diagnostics["registered_after_filter"] = sorted(available_whitelisted)
+        self._diagnostics["dropped_tools"] = sorted(dropped_tools)
+        self._diagnostics["unknown_tools"] = sorted(unknown_tools)
+        
+        # Handle unknown tools based on environment
+        if unknown_tools:
+            unknown_list = sorted(unknown_tools)
+            if self.is_ci:
+                raise ValueError(
+                    f"Unknown tools in {self.env_var_name}: {unknown_list}. "
+                    "All specified tools must be available in CI mode."
+                )
+            else:
+                logger.warning(
+                    "Unknown tools in %s will be ignored: %s",
+                    self.env_var_name,
+                    unknown_list
+                )
+        
+        if dropped_tools:
+            logger.info(
+                "Dropped %d tools not in %s whitelist: %s",
+                len(dropped_tools),
+                self.env_var_name,
+                sorted(dropped_tools)
+            )
+        
+        logger.info(
+            "%s filtered tools: %d available, %d after filter",
+            self.env_var_name,
+            len(tool_names),
+            len(available_whitelisted)
+        )
+        
+        return available_whitelisted
+    
+    def log_diagnostics(self) -> None:
+        """Log startup diagnostics section as specified in the issue."""
+        if not self._diagnostics:
+            logger.warning("No diagnostics available. Call filter_tools() first.")
+            return
+        
+        logger.info("=" * 50)
+        logger.info("ALLOWED_TOOLS FILTER DIAGNOSTICS")
+        logger.info("=" * 50)
+        logger.info("%s=%s", self.env_var_name, self.env_value or "<unset>")
+        logger.info("RegisteredBeforeFilter=%s", self._diagnostics.get("registered_before_filter", []))
+        logger.info("RegisteredAfterFilter=%s", self._diagnostics.get("registered_after_filter", []))
+        logger.info("DroppedTools=%s", self._diagnostics.get("dropped_tools", []))
+        logger.info("UnknownTools=%s", self._diagnostics.get("unknown_tools", []))
+        logger.info("IsCI=%s", self.is_ci)
+        logger.info("=" * 50)
+    
+    def get_diagnostics(self) -> Dict[str, Any]:
+        """
+        Get diagnostics data as a dictionary.
+        
+        Returns:
+            Dictionary with diagnostics information
+        """
+        return {
+            "env_var_name": self.env_var_name,
+            "env_value": self.env_value,
+            "is_ci": self.is_ci,
+            "whitelist": list(self._whitelist) if self._whitelist else None,
+            **self._diagnostics
+        }
+    
+    def is_enabled(self) -> bool:
+        """
+        Check if filtering is enabled.
+        
+        Returns:
+            True if tool whitelist is set and filtering is active
+        """
+        return self._whitelist is not None
+    
+    def get_whitelist(self) -> Optional[Set[str]]:
+        """
+        Get the current whitelist.
+        
+        Returns:
+            Set of whitelisted tool names, or None if not set
+        """
+        return self._whitelist.copy() if self._whitelist else None
+
+
+def filter_tools_with_allowed_tools(
+    available_tools: Union[Set[str], List[str], Dict[str, Any]],
+    env_var_name: str = "ALLOWED_TOOLS",
+    log_diagnostics: bool = True
+) -> Set[str]:
+    """
+    Convenience function to filter tools with ALLOWED_TOOLS.
+    
+    Args:
+        available_tools: Available tools to filter
+        env_var_name: Environment variable name (default: "ALLOWED_TOOLS")
+        log_diagnostics: Whether to log diagnostics (default: True)
+        
+    Returns:
+        Filtered set of tool names
+    """
+    filter_instance = AllowedToolsFilter(env_var_name)
+    filtered = filter_instance.filter_tools(available_tools)
+    
+    if log_diagnostics:
+        filter_instance.log_diagnostics()
+    
+    return filtered
+
+
+# Backward compatibility aliases
+filter_tools_with_hermes = filter_tools_with_allowed_tools
+hermes_filter = filter_tools_with_allowed_tools
+apply_hermes_filter = filter_tools_with_allowed_tools
+
+# Legacy class alias for backward compatibility
+HermesToolFilter = AllowedToolsFilter

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

@@ -42,6 +42,7 @@ class MemoryBackend(str, Enum):
     FILE = "file"
     SQLITE = "sqlite"
     REDIS = "redis"
+    VALKEY = "valkey"
     POSTGRES = "postgres"
     MEM0 = "mem0"
     MONGODB = "mongodb"

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

@@ -346,13 +346,15 @@ class _LazyStorageModule:
     
     _BACKENDS = {
         "RedisBackend": "praisonai.storage",
-        "MongoDBBackend": "praisonai.storage", 
+        "MongoDBBackend": "praisonai.storage",
         "PostgreSQLBackend": "praisonai.storage",
         "DynamoDBBackend": "praisonai.storage",
+        "ValkeyBackend": "praisonai.storage",
+        "ValkeySearchBackend": "praisonai.storage",
         # Legacy names for compatibility
         "RedisStorageAdapter": "praisonai.storage",
         "MongoDBStorageAdapter": "praisonai.storage",
-        "PostgreSQLStorageAdapter": "praisonai.storage", 
+        "PostgreSQLStorageAdapter": "praisonai.storage",
         "DynamoDBStorageAdapter": "praisonai.storage",
     }
     
@@ -390,7 +392,7 @@ def get_backend(
     Factory function to get a storage backend.
     
     Args:
-        backend_type: Type of backend ("file", "sqlite", "redis", "mongodb", "postgresql", "dynamodb")
+        backend_type: Type of backend ("file", "sqlite", "redis", "mongodb", "postgresql", "dynamodb", "valkey")
         **kwargs: Backend-specific arguments
         
     Returns:
@@ -431,9 +433,13 @@ def get_backend(
         # Lazy load from wrapper
         backend_class = getattr(_heavy_backends, "DynamoDBBackend")
         return backend_class(**kwargs)
+    elif backend_type == "valkey":
+        # Lazy load from wrapper
+        backend_class = getattr(_heavy_backends, "ValkeyBackend")
+        return backend_class(**kwargs)
     else:
         raise ValueError(f"Unknown backend type: {backend_type}. "
-                        f"Supported: file, sqlite, redis, mongodb, postgresql, dynamodb")
+                        f"Supported: file, sqlite, redis, mongodb, postgresql, dynamodb, valkey")
 
 def __getattr__(name: str):
     """
@@ -442,7 +448,8 @@ def __getattr__(name: str):
     This allows `from praisonaiagents.storage.backends import RedisBackend` to work
     while maintaining lazy loading.
     """
-    if name in {"RedisBackend", "MongoDBBackend", "PostgreSQLBackend", "DynamoDBBackend"}:
+    if name in {"RedisBackend", "MongoDBBackend", "PostgreSQLBackend", "DynamoDBBackend",
+                "ValkeyBackend", "ValkeySearchBackend"}:
         return getattr(_heavy_backends, name)
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 

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

@@ -10,7 +10,7 @@ _tools_lazy_cache = {}
 # Export core tool items for organized imports (lightweight)
 from .base import BaseTool, ToolResult, ToolValidationError, validate_tool
 from .decorator import tool, FunctionTool
-from .registry import get_registry, register_tool, get_tool, add_tool, has_tool, remove_tool, list_tools, list_available_tools, ToolRegistry
+from .registry import get_registry, register_tool, get_tool, add_tool, has_tool, remove_tool, list_tools, list_available_tools, list_tools_with_allowed_filter, list_tools_with_hermes_filter, ToolRegistry
 from .tools import Tools
 
 # Export Injected type directly for easy access
@@ -363,7 +363,7 @@ __all__ = list(TOOL_MAPPINGS.keys()) + [
     'BaseTool', 'ToolResult', 'ToolValidationError', 'validate_tool',
     'tool', 'FunctionTool',
     'get_registry', 'register_tool', 'get_tool', 'add_tool', 'has_tool', 'remove_tool', 
-    'list_tools', 'list_available_tools', 'ToolRegistry',
+    'list_tools', 'list_available_tools', 'list_tools_with_allowed_filter', 'list_tools_with_hermes_filter', 'ToolRegistry',
     'Tools',
     # Validation and retry protocols
     'ValidationResult', 'ToolValidatorProtocol', 'AsyncToolValidatorProtocol', 'PassthroughValidator',

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

@@ -184,6 +184,41 @@ class ToolRegistry:
             
             return available
     
+    def list_tools_with_allowed_filter(self, context: Optional[Dict[str, Any]] = None) -> List[str]:
+        """List tool names filtered by ALLOWED_TOOLS environment variable.
+        
+        This applies the canonical ALLOWED_TOOLS filter to prevent tool
+        name collisions in multi-environment agent systems.
+        
+        Args:
+            context: Optional context for availability checks
+            
+        Returns:
+            List of filtered tool names
+        """
+        try:
+            # Lazy import to avoid circular dependencies
+            from ..allowed_tools_filter import filter_tools_with_allowed_tools
+            
+            # Get all available tool names
+            all_tools = self.list_tools()
+            
+            # Apply ALLOWED_TOOLS filter with diagnostics
+            filtered_tools = filter_tools_with_allowed_tools(all_tools, log_diagnostics=True)
+            
+            return sorted(filtered_tools)
+            
+        except ImportError:
+            logging.warning("ALLOWED_TOOLS filter not available, returning all tools")
+            return self.list_tools()
+        except ValueError:
+            # Preserve strict ALLOWED_TOOLS semantics (e.g., CI unknown tools, empty value)
+            raise
+        except Exception as e:
+            logging.error("Error applying ALLOWED_TOOLS filter: %s", e)
+            # Fallback to all tools on error
+            return self.list_tools()
+    
     def get_all(self) -> Dict[str, Union[BaseTool, Callable]]:
         """Get all registered tools as a dict. Thread-safe."""
         with self._lock:
@@ -381,3 +416,24 @@ def list_available_tools(context: Optional[Dict[str, Any]] = None) -> List[Union
         List of available tools
     """
     return get_registry().list_available_tools(context)
+
+
+def list_tools_with_allowed_filter(context: Optional[Dict[str, Any]] = None) -> List[str]:
+    """List tool names filtered by ALLOWED_TOOLS from the global registry.
+    
+    This is the canonical entry point for applying ALLOWED_TOOLS filtering
+    across the PraisonAI ecosystem.
+    
+    Args:
+        context: Optional context for availability checks
+        
+    Returns:
+        List of filtered tool names
+    """
+    return get_registry().list_tools_with_allowed_filter(context)
+
+
+# Backward compatibility alias
+def list_tools_with_hermes_filter(context: Optional[Dict[str, Any]] = None) -> List[str]:
+    """Backward compatibility alias for list_tools_with_allowed_filter."""
+    return list_tools_with_allowed_filter(context)

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

@@ -0,0 +1,328 @@
+"""Integration tests for the ALLOWED_TOOLS filter module."""
+import os
+import pytest
+from unittest.mock import patch, MagicMock
+from praisonai.cli.features.agents import MultiAgentHandler
+
+
+class TestAllowedToolsFilterRegistryIntegration:
+    """Integration tests for ALLOWED_TOOLS filter with the tool registry."""
+    
+    def test_registry_allowed_filter_unset(self):
+        """Test registry filter behavior when ALLOWED_TOOLS is unset."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        with patch.dict(os.environ, {}, clear=True):
+            registry = get_registry()
+            
+            # Mock some tools in the registry
+            with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email']):
+                filtered_tools = registry.list_tools_with_allowed_filter()
+                
+                # Should return all tools when unset
+                assert set(filtered_tools) == {'search_web', 'send_email'}
+    
+    def test_registry_allowed_filter_with_whitelist(self):
+        """Test registry filter with ALLOWED_TOOLS whitelist."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        with patch.dict(os.environ, {
+            "ALLOWED_TOOLS": "search_web,send_email,unknown_tool",
+            "CI": "false",
+        }):
+            registry = get_registry()
+            
+            # Mock some tools in the registry
+            with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email', 'extract_pdf']):
+                filtered_tools = registry.list_tools_with_allowed_filter()
+                
+                # Should only include whitelisted tools that are available
+                assert set(filtered_tools) == {'search_web', 'send_email'}
+    
+    def test_registry_allowed_filter_empty_result(self):
+        """Test registry filter when no tools match whitelist."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        with patch.dict(os.environ, {
+            "ALLOWED_TOOLS": "nonexistent_tool1,nonexistent_tool2",
+            "CI": "false",
+        }):
+            registry = get_registry()
+            
+            # Mock some tools in the registry
+            with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email']):
+                filtered_tools = registry.list_tools_with_allowed_filter()
+                
+                # Should return empty list when no matches
+                assert filtered_tools == []
+    
+    def test_registry_allowed_filter_import_error_fallback(self):
+        """Test registry filter fallback when import fails."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        registry = get_registry()
+        
+        with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email']):
+            # Mock import error for the filter
+            with patch('praisonaiagents.tools.registry.filter_tools_with_allowed_tools', side_effect=ImportError("Filter not available")):
+                filtered_tools = registry.list_tools_with_allowed_filter()
+                
+                # Should fallback to all tools
+                assert set(filtered_tools) == {'search_web', 'send_email'}
+    
+    def test_registry_allowed_filter_value_error_propagation(self):
+        """Test that ValueError from filter is properly propagated."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ""}):  # Empty string should cause ValueError
+            registry = get_registry()
+            
+            with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email']):
+                with pytest.raises(ValueError, match="cannot be empty"):
+                    registry.list_tools_with_allowed_filter()
+    
+    def test_registry_allowed_filter_general_error_fallback(self):
+        """Test registry filter fallback on general errors."""
+        from praisonaiagents.tools.registry import get_registry
+        
+        registry = get_registry()
+        
+        with patch.object(registry, 'list_tools', return_value=['search_web', 'send_email']):
+            # Mock a generic error during filtering
+            with patch('praisonaiagents.tools.registry.filter_tools_with_allowed_tools', side_effect=RuntimeError("Some error")):
+                filtered_tools = registry.list_tools_with_allowed_filter()
+                
+                # Should fallback to all tools on general errors
+                assert set(filtered_tools) == {'search_web', 'send_email'}
+
+
+class TestAllowedToolsFilterGlobalRegistryPath:
+    """Test ALLOWED_TOOLS filter through the global registry function."""
+    
+    def test_global_list_tools_with_allowed_filter(self):
+        """Test the global list_tools_with_allowed_filter function."""
+        from praisonaiagents.tools import list_tools_with_allowed_filter
+        from praisonaiagents.tools.registry import get_registry
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search_web,send_email"}):
+            registry = get_registry()
+            
+            with patch.object(registry, 'list_tools_with_allowed_filter', return_value=['search_web', 'send_email']) as mock_filter:
+                result = list_tools_with_allowed_filter()
+                
+                # Should call the registry method
+                mock_filter.assert_called_once_with(None)
+                assert result == ['search_web', 'send_email']
+    
+    def test_global_backward_compatibility_alias(self):
+        """Test the backward compatibility alias for hermes filter."""
+        from praisonaiagents.tools import list_tools_with_hermes_filter, list_tools_with_allowed_filter
+        from praisonaiagents.tools.registry import get_registry
+        
+        registry = get_registry()
+        
+        with patch.object(registry, 'list_tools_with_allowed_filter', return_value=['search_web']) as mock_filter:
+            # Both should call the same underlying function
+            result1 = list_tools_with_allowed_filter()
+            result2 = list_tools_with_hermes_filter()
+            
+            assert result1 == result2 == ['search_web']
+            assert mock_filter.call_count == 2
+
+
+class TestAllowedToolsFilterCLIIntegration:
+    """Integration tests for ALLOWED_TOOLS filter with CLI agent handler."""
+    
+    def test_cli_agents_load_tools_no_filter(self):
+        """Test CLI agent tool loading without ALLOWED_TOOLS filter."""
+        handler = MultiAgentHandler(verbose=False)
+        
+        with patch.dict(os.environ, {}, clear=True):
+            # Mock the tools import to avoid actual import dependencies
+            with patch('praisonai.cli.features.agents.internet_search'), \
+                 patch('praisonai.cli.features.agents.read_file'):
+                
+                tools = handler._load_tools(["internet_search", "read_file"])
+                
+                # Should load all requested tools when no filter
+                assert len(tools) == 2
+    
+    def test_cli_agents_load_tools_with_filter(self):
+        """Test CLI agent tool loading with ALLOWED_TOOLS filter."""
+        handler = MultiAgentHandler(verbose=False)
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "internet_search"}):
+            # Mock the tools and filter
+            mock_internet_search = MagicMock()
+            mock_read_file = MagicMock()
+            
+            with patch('praisonai.cli.features.agents.internet_search', mock_internet_search), \
+                 patch('praisonai.cli.features.agents.read_file', mock_read_file):
+                
+                tools = handler._load_tools(["internet_search", "read_file"])
+                
+                # Should only load tools allowed by filter
+                assert len(tools) == 1
+                assert mock_internet_search in tools
+                assert mock_read_file not in tools
+    
+    def test_cli_agents_load_tools_filter_error_handling(self):
+        """Test CLI tool loading with ALLOWED_TOOLS filter error handling."""
+        handler = MultiAgentHandler(verbose=False)
+        requested = ["internet_search", "read_file"]
+        
+        with patch('os.environ.get', return_value=None):
+            with patch('praisonai.cli.features.agents.internet_search'), \
+                 patch('praisonai.cli.features.agents.read_file'):
+                baseline_tools = handler._load_tools(requested)
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ""}):
+            with pytest.raises(ValueError):
+                handler._load_tools(requested)
+    
+    def test_cli_agents_load_tools_import_error_fallback(self):
+        """Test CLI tool loading fallback when ALLOWED_TOOLS filter import fails."""
+        handler = MultiAgentHandler(verbose=False)
+        
+        # Mock ImportError for the filter module
+        with patch('praisonai.cli.features.agents.filter_tools_with_allowed_tools', side_effect=ImportError("Filter module not available")):
+            with patch('praisonai.cli.features.agents.internet_search'), \
+                 patch('praisonai.cli.features.agents.read_file'):
+                
+                tools = handler._load_tools(["internet_search", "read_file"])
+                
+                # Should load all tools on import error
+                assert len(tools) == 2
+    
+    def test_cli_agents_load_tools_filter_value_error_propagation(self):
+        """Test that ValueError from filter is properly propagated in CLI."""
+        handler = MultiAgentHandler(verbose=False)
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ""}):  # Empty string causes ValueError
+            with pytest.raises(ValueError, match="cannot be empty"):
+                handler._load_tools(["internet_search"])
+
+
+class TestAllowedToolsFilterSpecSemantics:
+    """Test ALLOWED_TOOLS filter specification semantics."""
+    
+    def test_spec_semantics_unset_all_tools_visible(self):
+        """Test spec: unset ALLOWED_TOOLS means all tools visible with warning."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {}, clear=True):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message", "extract_pdf"}
+            
+            import logging
+            with patch.object(logging.getLogger(), 'warning') as mock_warn:
+                filtered = filter_instance.filter_tools(available_tools)
+            
+            # Should return all tools
+            assert filtered == available_tools
+            
+            # Should warn about unset whitelist
+            mock_warn.assert_called_once()
+            assert "Tool whitelist is unset" in mock_warn.call_args[0][0]
+    
+    def test_spec_semantics_empty_string_error(self):
+        """Test spec: empty ALLOWED_TOOLS string is an error."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ""}):
+            with pytest.raises(ValueError, match="ALLOWED_TOOLS cannot be empty"):
+                AllowedToolsFilter()
+    
+    def test_spec_semantics_whitelist_only_specified_visible(self):
+        """Test spec: with ALLOWED_TOOLS values, only whitelisted tools visible."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message", "extract_pdf", "analyze_code"}
+            
+            filtered = filter_instance.filter_tools(available_tools)
+            
+            # Should only return whitelisted tools
+            assert filtered == {"search", "send_message"}
+    
+    def test_spec_semantics_unknown_tools_dev_warn_strip(self):
+        """Test spec: unknown tools in dev mode are warned and stripped."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,unknown_tool", "CI": "false"}):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message"}
+            
+            import logging
+            with patch.object(logging.getLogger(), 'warning') as mock_warn:
+                filtered = filter_instance.filter_tools(available_tools)
+            
+            # Should return only known tools
+            assert filtered == {"search"}
+            
+            # Should warn about unknown tools
+            mock_warn.assert_called()
+            assert "unknown_tool" in str(mock_warn.call_args[0][2])
+    
+    def test_spec_semantics_unknown_tools_ci_strict_fail(self):
+        """Test spec: unknown tools in CI mode cause strict failure."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,unknown_tool", "CI": "true"}):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message"}
+            
+            with pytest.raises(ValueError, match="Unknown tools in ALLOWED_TOOLS"):
+                filter_instance.filter_tools(available_tools)
+    
+    def test_spec_semantics_order_independence(self):
+        """Test spec: whitelist order doesn't matter."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"}):
+            filter1 = AllowedToolsFilter()
+            result1 = filter1.filter_tools(available_tools)
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "send_message,search"}):
+            filter2 = AllowedToolsFilter()
+            result2 = filter2.filter_tools(available_tools)
+        
+        # Results should be the same regardless of order
+        assert result1 == result2 == {"search", "send_message"}
+    
+    def test_spec_semantics_duplicate_isolation(self):
+        """Test spec: duplicate tool names in whitelist are handled properly."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,search,send_message,search"}):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message", "extract_pdf"}
+            
+            filtered = filter_instance.filter_tools(available_tools)
+            
+            # Duplicates should be de-duplicated
+            assert filtered == {"search", "send_message"}
+            
+            # Whitelist should be a set (no duplicates)
+            whitelist = filter_instance.get_whitelist()
+            assert whitelist == {"search", "send_message"}
+    
+    def test_spec_backward_compatibility_hermes_precedence(self):
+        """Test spec: ALLOWED_TOOLS takes precedence over HERMES_ONLY_TOOLS."""
+        from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
+        
+        with patch.dict(os.environ, {
+            "ALLOWED_TOOLS": "search,send_message",
+            "HERMES_ONLY_TOOLS": "extract_pdf"
+        }):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message", "extract_pdf"}
+            
+            filtered = filter_instance.filter_tools(available_tools)
+            
+            # Should use ALLOWED_TOOLS, not HERMES_ONLY_TOOLS
+            assert filtered == {"search", "send_message"}
+            assert filter_instance.env_var_name == "ALLOWED_TOOLS"

src/praisonai-agents/tests/managed/test_managed_persistence_all_dbs.py

@@ -6,6 +6,7 @@ Databases tested (all via local Docker):
   2. PostgreSQL     — ConversationStore + ManagedAgent roundtrip
   3. MySQL          — ConversationStore + ManagedAgent roundtrip
   4. Redis          — StateStore + metadata persistence
+  4a. Valkey        — StateStore + metadata persistence
   5. MongoDB        — StateStore + metadata persistence
   6. ClickHouse     — Raw connectivity + data write/read
   7. JSON file      — DefaultSessionStore + ManagedAgent roundtrip
@@ -27,7 +28,6 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -562,6 +562,206 @@ class TestRedisStateStore:
             r.close()
 
 
+
+# ===========================================================================
+# 4a. Valkey — StateStore operations + metadata persistence
+# ===========================================================================
+class TestValkeyStateStore:
+    """Valkey StateStore: real operations against a live Valkey instance."""
+
+    @pytest.fixture(autouse=True)
+    def check_valkey(self):
+        """Skip if Valkey not available."""
+        try:
+            from praisonai.persistence._valkey_client import create_valkey_client
+            client = create_valkey_client(host="localhost", port=6379)
+            client.ping()
+            client.close()
+        except Exception:
+            pytest.skip("Valkey not available on localhost:6379")
+
+    def test_valkey_state_operations(self):
+        """StateStore get/set/delete/exists/hash with real Valkey."""
+        from praisonai.persistence.state.valkey import ValkeyStateStore
+        prefix = f"test_{uuid.uuid4().hex[:6]}:"
+        store = ValkeyStateStore(host="localhost", port=6379, prefix=prefix)
+
+        try:
+            # Basic set/get
+            store.set("agent_id", "agent_valkey_001")
+            assert store.get("agent_id") == "agent_valkey_001"
+
+            # JSON roundtrip
+            state = {
+                "agent_id": "agent_valkey_001",
+                "agent_version": 3,
+                "total_input_tokens": 500,
+                "total_output_tokens": 200,
+                "compute_instance_id": "docker_valkey_test",
+                "session_history": [{"id": "s1", "status": "idle"}],
+            }
+            store.set_json("managed_state", state)
+            recovered = store.get_json("managed_state")
+            assert recovered["agent_id"] == "agent_valkey_001"
+            assert recovered["total_input_tokens"] == 500
+            assert recovered["compute_instance_id"] == "docker_valkey_test"
+            assert len(recovered["session_history"]) == 1
+
+            # Hash operations
+            store.hset("agent_meta", "version", "5")
+            store.hset("agent_meta", "env_id", "env_abc")
+            assert str(store.hget("agent_meta", "version")) == "5"
+            all_meta = store.hgetall("agent_meta")
+            assert str(all_meta["env_id"]) == "env_abc"
+
+            # Exists and delete
+            assert store.exists("agent_id")
+            store.delete("agent_id")
+            assert not store.exists("agent_id")
+
+        finally:
+            for k in store.keys():
+                store.delete(k)
+            store.close()
+
+    def test_valkey_managed_state_persist_restore(self):
+        """Simulate managed agent state persist/restore across store instances."""
+        from praisonai.persistence.state.valkey import ValkeyStateStore
+        prefix = f"mgd_{uuid.uuid4().hex[:6]}:"
+        store = ValkeyStateStore(host="localhost", port=6379, prefix=prefix)
+
+        try:
+            session_id = f"valkey_session_{uuid.uuid4().hex[:8]}"
+            state = {
+                "agent_id": "agent_valkey_mgd",
+                "agent_version": 2,
+                "environment_id": "env_valkey_001",
+                "total_input_tokens": 1000,
+                "total_output_tokens": 500,
+                "compute_instance_id": "modal_valkey_xyz",
+                "session_history": [
+                    {"id": session_id, "status": "idle", "title": "Valkey test"},
+                ],
+            }
+            store.set_json(f"managed:{session_id}", state)
+
+            # Simulate restart — new store instance, same prefix
+            store2 = ValkeyStateStore(host="localhost", port=6379, prefix=prefix)
+            recovered = store2.get_json(f"managed:{session_id}")
+            assert recovered is not None
+            assert recovered["agent_id"] == "agent_valkey_mgd"
+            assert recovered["total_input_tokens"] == 1000
+            assert recovered["compute_instance_id"] == "modal_valkey_xyz"
+            assert len(recovered["session_history"]) == 1
+
+            store.close()
+            store2.close()
+        finally:
+            store3 = ValkeyStateStore(host="localhost", port=6379, prefix=prefix)
+            for k in store3.keys():
+                store3.delete(k)
+            store3.close()
+
+
+# ===========================================================================
+# 4b. Valkey Vector Search — ValkeyVectorKnowledgeStore against a live instance
+# ===========================================================================
+class TestValkeyVectorSearch:
+    """Valkey vector search: real operations against a live ValkeySearch instance."""
+
+    @pytest.fixture(autouse=True)
+    def check_valkey_search(self):
+        """Skip if Valkey is unreachable or ValkeySearch (FT.*) is not loaded."""
+        try:
+            from glide_sync import ft
+        except ImportError:
+            pytest.skip("valkey-glide-sync not installed")
+
+        try:
+            from praisonai.persistence._valkey_client import create_valkey_client
+        except ImportError as e:
+            pytest.skip(f"praisonai not importable: {e}")
+
+        client = None
+        try:
+            client = create_valkey_client(host="localhost", port=6379)
+            client.ping()
+        except Exception as e:
+            if client is not None:
+                client.close()
+            pytest.skip(f"Valkey not available on localhost:6379: {e}")
+
+        try:
+            # Probe FT.* support; raises if module not loaded
+            ft.list(client)
+        except Exception as e:
+            pytest.skip(f"ValkeySearch (FT.*) module not loaded on localhost:6379: {e}")
+        finally:
+            if client is not None:
+                client.close()
+
+    def test_valkey_vector_search_roundtrip(self):
+        """Full create → insert → search → delete_collection roundtrip."""
+        from praisonai.persistence.knowledge.valkey_vector import ValkeyVectorKnowledgeStore
+        from praisonai.persistence.knowledge.base import KnowledgeDocument
+
+        prefix = f"praisontest_{uuid.uuid4().hex[:6]}_"
+        collection = f"vec_{uuid.uuid4().hex[:6]}"
+        dimension = 4
+        store = ValkeyVectorKnowledgeStore(host="localhost", port=6379, prefix=prefix)
+
+        try:
+            # Create index
+            store.create_collection(collection, dimension=dimension, distance="cosine")
+            assert store.collection_exists(collection)
+
+            # Insert documents with simple, distinguishable embeddings
+            docs = [
+                KnowledgeDocument(
+                    id="doc_a",
+                    content="alpha document",
+                    embedding=[1.0, 0.0, 0.0, 0.0],
+                ),
+                KnowledgeDocument(
+                    id="doc_b",
+                    content="bravo document",
+                    embedding=[0.0, 1.0, 0.0, 0.0],
+                ),
+                KnowledgeDocument(
+                    id="doc_c",
+                    content="charlie document",
+                    embedding=[0.0, 0.0, 1.0, 0.0],
+                ),
+            ]
+            inserted = store.insert(collection, docs)
+            assert set(inserted) == {"doc_a", "doc_b", "doc_c"}
+
+            # Vector search: query closest to doc_a
+            results = store.search(collection, query_embedding=[1.0, 0.0, 0.0, 0.0], limit=2)
+            assert len(results) >= 1, "Expected at least one search result"
+            assert results[0].id == "doc_a", f"Expected doc_a as nearest neighbor, got {results[0].id}"
+            assert results[0].content == "alpha document"
+
+            # Verify the bare ID is returned (no prefix leakage)
+            for r in results:
+                assert ":" not in r.id or r.id in {"doc_a", "doc_b", "doc_c"}
+
+            # get() roundtrip
+            got = store.get(collection, ["doc_b"])
+            assert len(got) == 1
+            assert got[0].id == "doc_b"
+            assert got[0].content == "bravo document"
+
+            # Delete one doc
+            deleted = store.delete(collection, ids=["doc_c"])
+            assert deleted == 1
+        finally:
+            try:
+                store.delete_collection(collection)
+            finally:
+                store.close()
+
+
 # ===========================================================================
 # 5. MongoDB — StateStore operations + metadata persistence
 # ===========================================================================
@@ -673,7 +873,6 @@ class TestClickHouseConnectivity:
     def test_clickhouse_write_read(self):
         """Write and read data from ClickHouse."""
         import clickhouse_connect
-
         table = f"praison_test_{uuid.uuid4().hex[:8]}"
         c = clickhouse_connect.get_client(
             host="localhost", port=8123, username="clickhouse", password="clickhouse",

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

@@ -0,0 +1,84 @@
+"""Unit tests for unified achat dispatch (Issue #1703)."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from praisonaiagents import Agent
+
+
+def _fake_completion(content: str = "Mock async reply"):
+    return SimpleNamespace(
+        choices=[SimpleNamespace(message=SimpleNamespace(content=content, reasoning_content=None))]
+    )
+
+
+@pytest.mark.asyncio
+async def test_unified_achat_returns_after_single_dispatch():
+    """achat must return after one unified dispatch, not loop forever."""
+    agent = Agent(name="test", instructions="You are helpful", llm="gpt-4o-mini")
+    agent._using_custom_llm = False
+    agent.self_reflect = False
+
+    mock_unified = AsyncMock(return_value=_fake_completion())
+
+    with patch.object(agent, "_execute_unified_achat_completion", mock_unified), patch.object(
+        agent, "_apply_guardrail_with_retry", return_value="Mock async reply"
+    ), patch.object(
+        agent, "_atrigger_after_agent_hook", new_callable=AsyncMock, return_value="Mock async reply"
+    ), patch.object(
+        agent, "_execute_callback_and_display"
+    ), patch.object(
+        agent, "_build_messages", return_value=([{"role": "user", "content": "hi"}], "hi")
+    ), patch.object(
+        agent, "_persist_message"
+    ), patch.object(
+        agent, "_format_tools_for_completion", return_value=[]
+    ):
+        result = await asyncio.wait_for(agent.achat("hi", stream=False), timeout=5)
+
+    assert result == "Mock async reply"
+    mock_unified.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_unified_achat_completion_uses_execute_tool_async():
+    """Async unified dispatch must wire execute_tool_async, not sync execute_tool."""
+    agent = Agent(name="test", instructions="You are helpful", llm="gpt-4o-mini")
+    agent._using_custom_llm = False
+
+    mock_dispatcher = MagicMock()
+    mock_dispatcher.achat_completion = AsyncMock(return_value=_fake_completion())
+    agent._unified_dispatcher = mock_dispatcher
+
+    with patch.object(agent, "_apply_guardrail_with_retry", return_value="ok"), patch.object(
+        agent, "_atrigger_after_agent_hook", new_callable=AsyncMock, return_value="ok"
+    ), patch.object(agent, "_execute_callback_and_display"), patch.object(
+        agent, "_build_messages", return_value=([{"role": "user", "content": "hi"}], "hi")
+    ), patch.object(agent, "_persist_message"), patch.object(
+        agent, "_format_tools_for_completion", return_value=[]
+    ):
+        await agent._achat_impl(
+            prompt="hi",
+            temperature=1.0,
+            tools=None,
+            output_json=None,
+            output_pydantic=None,
+            reasoning_steps=False,
+            stream=False,
+            task_name=None,
+            task_description=None,
+            task_id=None,
+            config=None,
+            force_retrieval=False,
+            skip_retrieval=True,
+            attachments=None,
+            _trace_emitter=MagicMock(),
+        )
+
+    call_kwargs = mock_dispatcher.achat_completion.call_args[1]
+    tool_fn = call_kwargs.get("execute_tool_fn")
+    assert tool_fn is not None
+    assert getattr(tool_fn, "__name__", "") == "execute_tool_async"

src/praisonai-agents/tests/unit/config/test_precedence_ladder.py

@@ -48,6 +48,12 @@ class TestResolveMemory:
         result = resolve_memory("redis")
         assert isinstance(result, MemoryConfig)
         assert result.backend == MemoryBackend.REDIS
+
+    def test_string_valkey_backend(self):
+        """'valkey' string returns MemoryConfig with VALKEY backend."""
+        result = resolve_memory("valkey")
+        assert isinstance(result, MemoryConfig)
+        assert result.backend == MemoryBackend.VALKEY
     
     def test_string_custom_backend(self):
         """Custom string backend is preserved."""

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

@@ -0,0 +1,300 @@
+"""Unit tests for the ALLOWED_TOOLS filter module."""
+import os
+import logging
+from unittest.mock import patch, MagicMock
+import pytest
+
+from praisonaiagents.allowed_tools_filter import AllowedToolsFilter, filter_tools_with_allowed_tools
+
+
+class TestAllowedToolsFilter:
+    """Test cases for AllowedToolsFilter class."""
+    
+    def test_filter_initialization_unset(self):
+        """Test filter initialization with no environment variable set."""
+        with patch.dict(os.environ, {}, clear=True):
+            filter_instance = AllowedToolsFilter()
+            assert filter_instance.env_value is None
+            assert not filter_instance.is_enabled()
+            assert filter_instance.get_whitelist() is None
+    
+    def test_filter_initialization_primary_var(self):
+        """Test filter initialization with ALLOWED_TOOLS set."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter()
+            assert filter_instance.env_value == "search,send_message"
+            assert filter_instance.env_var_name == "ALLOWED_TOOLS"
+            assert filter_instance.is_enabled()
+            assert filter_instance.get_whitelist() == {"search", "send_message"}
+    
+    def test_filter_initialization_legacy_var(self):
+        """Test filter initialization with HERMES_ONLY_TOOLS set (backward compatibility)."""
+        with patch.dict(os.environ, {"HERMES_ONLY_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter()
+            assert filter_instance.env_value == "search,send_message"
+            assert filter_instance.env_var_name == "HERMES_ONLY_TOOLS"
+            assert filter_instance.is_enabled()
+            assert filter_instance.get_whitelist() == {"search", "send_message"}
+    
+    def test_filter_initialization_primary_precedence(self):
+        """Test that ALLOWED_TOOLS takes precedence over HERMES_ONLY_TOOLS."""
+        with patch.dict(os.environ, {
+            "ALLOWED_TOOLS": "search,send_message", 
+            "HERMES_ONLY_TOOLS": "old_tool"
+        }):
+            filter_instance = AllowedToolsFilter()
+            assert filter_instance.env_value == "search,send_message"
+            assert filter_instance.env_var_name == "ALLOWED_TOOLS"
+            assert filter_instance.get_whitelist() == {"search", "send_message"}
+    
+    def test_filter_initialization_empty_string_error(self):
+        """Test that empty string raises ValueError."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ""}):
+            with pytest.raises(ValueError, match="ALLOWED_TOOLS cannot be empty"):
+                AllowedToolsFilter()
+    
+    def test_filter_initialization_whitespace_only_error(self):
+        """Test that whitespace-only string raises ValueError."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "   "}):
+            with pytest.raises(ValueError, match="ALLOWED_TOOLS cannot be empty"):
+                AllowedToolsFilter()
+    
+    def test_filter_initialization_empty_after_split_error(self):
+        """Test that comma-only string raises ValueError."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": ",,,"}):
+            with pytest.raises(ValueError, match="contains no valid tool names"):
+                AllowedToolsFilter()
+    
+    def test_parse_whitelist_with_spaces(self):
+        """Test parsing comma-separated values with spaces."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": " search , send_message , extract_pdf "}):
+            filter_instance = AllowedToolsFilter()
+            assert filter_instance.get_whitelist() == {"search", "send_message", "extract_pdf"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,unknown_tool,send_message", "CI": "false"})
+    def test_filter_tools_with_unknown_tools_dev_mode(self):
+        """Test filtering tools with unknown tools in development mode."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        with patch.object(logging.getLogger(), 'warning') as mock_warn:
+            filtered = filter_instance.filter_tools(available_tools)
+            
+        # Should return intersection of available and whitelisted tools
+        assert filtered == {"search", "send_message"}
+        
+        # Should warn about unknown tools
+        mock_warn.assert_called()
+        call_args = mock_warn.call_args[0]
+        assert "Unknown tools" in call_args[0]
+        assert ["unknown_tool"] == call_args[2]
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,unknown_tool,send_message", "CI": "true"})
+    def test_filter_tools_with_unknown_tools_ci_mode(self):
+        """Test filtering tools with unknown tools in CI mode."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        with pytest.raises(ValueError, match="Unknown tools in ALLOWED_TOOLS"):
+            filter_instance.filter_tools(available_tools)
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_filter_tools_all_available(self):
+        """Test filtering when all whitelisted tools are available."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        assert filtered == {"search", "send_message"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_filter_tools_partial_match(self):
+        """Test filtering when only some whitelisted tools are available."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "extract_pdf"}
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        assert filtered == {"search"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_filter_tools_dict_input(self):
+        """Test filtering with dictionary input."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search": lambda: None, "send_message": lambda: None, "extract_pdf": lambda: None}
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        assert filtered == {"search", "send_message"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_filter_tools_list_input(self):
+        """Test filtering with list input."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = ["search", "send_message", "extract_pdf"]
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        assert filtered == {"search", "send_message"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message,unknown_in_whitelist"})
+    def test_diagnostics_data(self):
+        """Test diagnostics data collection."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        diagnostics = filter_instance.get_diagnostics()
+        
+        assert diagnostics["env_var_name"] == "ALLOWED_TOOLS"
+        assert diagnostics["env_value"] == "search,send_message,unknown_in_whitelist"
+        assert diagnostics["whitelist"] == ["search", "send_message", "unknown_in_whitelist"]
+        assert set(diagnostics["registered_before_filter"]) == {"search", "send_message", "extract_pdf"}
+        assert set(diagnostics["registered_after_filter"]) == {"search", "send_message"}
+        assert set(diagnostics["dropped_tools"]) == {"extract_pdf"}
+        assert set(diagnostics["unknown_tools"]) == {"unknown_in_whitelist"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_log_diagnostics(self, caplog):
+        """Test diagnostics logging."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        with caplog.at_level(logging.INFO):
+            filter_instance.filter_tools(available_tools)
+            filter_instance.log_diagnostics()
+        
+        log_text = caplog.text
+        assert "ALLOWED_TOOLS FILTER DIAGNOSTICS" in log_text
+        assert "ALLOWED_TOOLS=search,send_message" in log_text
+        assert "RegisteredBeforeFilter" in log_text
+        assert "RegisteredAfterFilter" in log_text
+        assert "DroppedTools" in log_text
+        assert "UnknownTools" in log_text
+    
+    def test_log_diagnostics_before_filtering(self, caplog):
+        """Test logging diagnostics before calling filter_tools."""
+        filter_instance = AllowedToolsFilter()
+        
+        with caplog.at_level(logging.WARNING):
+            filter_instance.log_diagnostics()
+        
+        assert "No diagnostics available" in caplog.text
+    
+    def test_filter_tools_unset_env_var(self, caplog):
+        """Test filtering with unset environment variable."""
+        with patch.dict(os.environ, {}, clear=True):
+            filter_instance = AllowedToolsFilter()
+            available_tools = {"search", "send_message", "extract_pdf"}
+            
+            with caplog.at_level(logging.WARNING):
+                filtered = filter_instance.filter_tools(available_tools)
+            
+            # Should return all tools
+            assert filtered == available_tools
+            assert "Tool whitelist is unset" in caplog.text
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "tool1", "CI": "false"})
+    def test_no_matching_tools(self):
+        """Test filtering when no whitelisted tools match available tools."""
+        filter_instance = AllowedToolsFilter()
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        filtered = filter_instance.filter_tools(available_tools)
+        assert filtered == set()  # Empty set when no matches
+    
+    def test_custom_env_var_name(self):
+        """Test using custom environment variable name."""
+        with patch.dict(os.environ, {"CUSTOM_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter("CUSTOM_TOOLS")
+            assert filter_instance.env_var_name == "CUSTOM_TOOLS"
+            assert filter_instance.get_whitelist() == {"search", "send_message"}
+
+
+class TestConvenienceFunction:
+    """Test cases for the convenience function."""
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"})
+    def test_filter_tools_with_allowed_tools(self):
+        """Test the convenience function."""
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        filtered = filter_tools_with_allowed_tools(available_tools, log_diagnostics=False)
+        assert filtered == {"search", "send_message"}
+    
+    @patch.dict(os.environ, {"CUSTOM_TOOLS": "search"})
+    def test_filter_tools_with_custom_env_var(self):
+        """Test the convenience function with custom env var."""
+        available_tools = {"search", "send_message"}
+        
+        filtered = filter_tools_with_allowed_tools(available_tools, env_var_name="CUSTOM_TOOLS", log_diagnostics=False)
+        assert filtered == {"search"}
+    
+    @patch.dict(os.environ, {"ALLOWED_TOOLS": "search"})
+    def test_filter_tools_with_diagnostics(self, caplog):
+        """Test the convenience function with diagnostics enabled."""
+        available_tools = {"search", "send_message"}
+        
+        with caplog.at_level(logging.INFO):
+            filtered = filter_tools_with_allowed_tools(available_tools, log_diagnostics=True)
+        
+        assert filtered == {"search"}
+        assert "ALLOWED_TOOLS FILTER DIAGNOSTICS" in caplog.text
+
+
+class TestBackwardCompatibility:
+    """Test backward compatibility aliases."""
+    
+    @patch.dict(os.environ, {"HERMES_ONLY_TOOLS": "search,send_message"})
+    def test_legacy_class_alias(self):
+        """Test HermesToolFilter alias works."""
+        from praisonaiagents.allowed_tools_filter import HermesToolFilter
+        
+        filter_instance = HermesToolFilter()
+        assert isinstance(filter_instance, AllowedToolsFilter)
+        assert filter_instance.env_var_name == "HERMES_ONLY_TOOLS"
+        assert filter_instance.get_whitelist() == {"search", "send_message"}
+    
+    @patch.dict(os.environ, {"HERMES_ONLY_TOOLS": "search,send_message"})
+    def test_legacy_function_aliases(self):
+        """Test legacy function aliases work."""
+        from praisonaiagents.allowed_tools_filter import (
+            filter_tools_with_hermes, hermes_filter, apply_hermes_filter
+        )
+        
+        available_tools = {"search", "send_message", "extract_pdf"}
+        
+        # All aliases should work and return the same result
+        result1 = filter_tools_with_hermes(available_tools, log_diagnostics=False)
+        result2 = hermes_filter(available_tools, log_diagnostics=False)
+        result3 = apply_hermes_filter(available_tools, log_diagnostics=False)
+        
+        expected = {"search", "send_message"}
+        assert result1 == expected
+        assert result2 == expected
+        assert result3 == expected
+    
+    def test_empty_tools_edge_case(self):
+        """Test filtering with empty available tools."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter()
+            
+            filtered = filter_instance.filter_tools(set())
+            assert filtered == set()
+            
+            # Also test with empty list and dict
+            assert filter_instance.filter_tools([]) == set()
+            assert filter_instance.filter_tools({}) == set()
+    
+    def test_whitelist_immutability(self):
+        """Test that get_whitelist returns a copy, not the original."""
+        with patch.dict(os.environ, {"ALLOWED_TOOLS": "search,send_message"}):
+            filter_instance = AllowedToolsFilter()
+            
+            whitelist1 = filter_instance.get_whitelist()
+            whitelist2 = filter_instance.get_whitelist()
+            
+            # Should be equal but not the same object
+            assert whitelist1 == whitelist2
+            assert whitelist1 is not whitelist2
+            
+            # Modifying one shouldn't affect the other
+            whitelist1.add("new_tool")
+            assert whitelist1 != whitelist2

src/praisonai/praisonai/cli/features/agents.py

@@ -123,6 +123,8 @@ class MultiAgentHandler:
     def _load_tools(self, tool_names: List[str]) -> List:
         """Load tool functions by name.
         
+        Applies ALLOWED_TOOLS filter if enabled to prevent tool name collisions.
+        
         Args:
             tool_names: List of tool names to load
             
@@ -148,6 +150,26 @@ class MultiAgentHandler:
                 'analyze_csv': analyze_csv,
             }
             
+            # Apply ALLOWED_TOOLS filter if enabled
+            try:
+                from praisonaiagents.allowed_tools_filter import filter_tools_with_allowed_tools
+                available_tools = set(tool_map.keys())
+                filtered_tools = filter_tools_with_allowed_tools(available_tools, log_diagnostics=False)
+                
+                # Filter tool_map to only include whitelisted tools
+                tool_map = {name: func for name, func in tool_map.items() if name in filtered_tools}
+                
+            except ImportError:
+                # ALLOWED_TOOLS filter not available, proceed normally
+                pass
+            except ValueError:
+                # Preserve strict filter semantics (invalid config / CI unknown tools)
+                raise
+            except Exception as e:
+                if self.verbose:
+                    print(f"⚠ ALLOWED_TOOLS filter error: {e}")
+                raise
+            
             for name in tool_names:
                 if name in tool_map:
                     tools.append(tool_map[name])

src/praisonai/praisonai/persistence/_valkey_client.py

@@ -0,0 +1,44 @@
+"""
+Shared Valkey client factory.
+
+Centralises GlideClientSync creation so that ValkeyStateStore,
+ValkeyVectorKnowledgeStore, and ValkeyStorageAdapter all use the
+same connection logic.
+"""
+
+from typing import Optional
+
+try:
+    from glide_sync import (
+        GlideClient as GlideClientSync,
+        GlideClientConfiguration,
+        NodeAddress,
+        ServerCredentials,
+    )
+except ImportError:
+    GlideClientSync = None
+    GlideClientConfiguration = NodeAddress = ServerCredentials = None
+
+_MISSING_MSG = (
+    "valkey-glide-sync is required for Valkey support. "
+    "Install with: pip install 'praisonai[valkey]'"
+)
+
+
+def create_valkey_client(
+    host: str = "localhost",
+    port: int = 6379,
+    password: Optional[str] = None,
+    db: int = 0,
+):
+    """Create and return a GlideClientSync instance."""
+    if GlideClientSync is None:
+        raise ImportError(_MISSING_MSG)
+    addresses = [NodeAddress(host, port)]
+    creds = ServerCredentials(password=password) if password else None
+    config = GlideClientConfiguration(
+        addresses=addresses,
+        credentials=creds,
+        database_id=db,
+    )
+    return GlideClientSync.create(config)

src/praisonai/praisonai/persistence/knowledge/valkey_vector.py

@@ -0,0 +1,397 @@
+"""
+Valkey Vector implementation of KnowledgeStore.
+
+Requires: valkey-glide-sync
+Install: pip install 'praisonai[valkey]'
+"""
+
+import re
+import struct
+from typing import Any, Dict, List, Optional
+
+from .base import KnowledgeStore, KnowledgeDocument, validate_identifier
+from .._valkey_client import create_valkey_client
+
+try:
+    from glide_sync import ft
+    from glide_sync import (
+        TextField, NumericField, VectorField, VectorFieldAttributesHnsw,
+        VectorAlgorithm, DistanceMetricType, VectorType,
+        FtCreateOptions, DataType, FtSearchOptions, ReturnField,
+    )
+    from glide_sync import Batch
+except ImportError:
+    ft = None
+    Batch = None
+    TextField = NumericField = VectorField = VectorFieldAttributesHnsw = None
+    VectorAlgorithm = DistanceMetricType = VectorType = None
+    FtCreateOptions = DataType = FtSearchOptions = ReturnField = None
+
+
+_SEARCH_SPECIAL_RE = re.compile(r'([,.<>{}\[\]"\'\\:;!@#$%^&*()\-+=~?/| \t])')
+_FIELD_NAME_RE = re.compile(r'^[a-zA-Z0-9_]+$')
+
+
+def _escape_search_value(value: str) -> str:
+    return _SEARCH_SPECIAL_RE.sub(r'\\\1', str(value))
+
+
+def _decode(v) -> str:
+    return v.decode() if isinstance(v, bytes) else v
+
+
+class ValkeyVectorKnowledgeStore(KnowledgeStore):
+    """
+    Valkey-based knowledge store using ValkeySearch FT commands for vector search.
+
+    Requires Valkey with ValkeySearch module.
+
+    Example:
+        store = ValkeyVectorKnowledgeStore(
+            host="localhost",
+            port=6379,
+        )
+    """
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        password: Optional[str] = None,
+        prefix: str = "praison:vec:",
+    ):
+        self.host = host
+        self.port = port
+        self.password = password
+        self.prefix = prefix
+        self._client = None
+
+    def _get_client(self):
+        """Lazy initialize Valkey client."""
+        if self._client is None:
+            self._client = create_valkey_client(
+                host=self.host,
+                port=self.port,
+                password=self.password,
+            )
+        return self._client
+
+    def _index_name(self, collection: str) -> str:
+        return f"{self.prefix}{collection}:idx"
+
+    def _doc_key(self, collection: str, doc_id: str) -> str:
+        return f"{self.prefix}{collection}:{doc_id}"
+
+    def create_collection(
+        self,
+        name: str,
+        dimension: int,
+        distance: str = "cosine",
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Create a new index."""
+        validate_identifier(name, "collection name")
+        client = self._get_client()
+        index_name = self._index_name(name)
+
+        try:
+            dist_enum = {"cosine": DistanceMetricType.COSINE, "euclidean": DistanceMetricType.L2, "dot": DistanceMetricType.IP}.get(distance, DistanceMetricType.COSINE)
+            schema = [
+                TextField("content"),
+                TextField("content_hash"),
+                NumericField("created_at"),
+                VectorField("embedding", VectorAlgorithm.HNSW, VectorFieldAttributesHnsw(dimension, dist_enum, VectorType.FLOAT32)),
+            ]
+            options = FtCreateOptions(DataType.HASH, prefixes=[f"{self.prefix}{name}:"])
+            ft.create(client, index_name, schema, options)
+        except Exception as e:
+            err = str(e)
+            if "already exists" not in err.lower():
+                raise RuntimeError(f"Failed to create collection in Valkey: {e}") from e
+
+    def delete_collection(self, name: str) -> bool:
+        """Delete an index and all documents."""
+        client = self._get_client()
+        index_name = self._index_name(name)
+        try:
+            ft.dropindex(client, index_name)
+        except Exception as e:
+            # Only return False for index not found, re-raise other errors
+            err_msg = str(e).lower()
+            if "index not found" in err_msg or "unknown index" in err_msg:
+                return False
+            raise RuntimeError(f"Failed to delete collection: {e}") from e
+
+        # Manually remove orphaned document hashes left behind (no DD flag in ValkeySearch)
+        pattern = f"{self.prefix}{name}:*"
+        cursor: bytes = b"0"
+        while True:
+            result = client.scan(cursor, match=pattern, count=100)
+            cursor = result[0]
+            keys = result[1] or []
+            if keys:
+                client.delete(keys)
+            if not cursor or cursor in (b"0", "0", 0):
+                break
+
+        return True
+
+    def collection_exists(self, name: str) -> bool:
+        """Check if an index exists."""
+        client = self._get_client()
+        index_name = self._index_name(name)
+        try:
+            ft.info(client, index_name)
+            return True
+        except Exception as e:
+            # Only return False for index not found, re-raise other errors
+            err_msg = str(e).lower()
+            if "index not found" in err_msg or "unknown index" in err_msg:
+                return False
+            raise RuntimeError(f"Failed to check collection existence: {e}") from e
+
+    def list_collections(self) -> List[str]:
+        """List all collections."""
+        client = self._get_client()
+        try:
+            indexes = ft.list(client)
+            if not indexes:
+                return []
+            result = []
+            for idx in indexes:
+                idx_str = _decode(idx)
+                if idx_str.startswith(self.prefix):
+                    name = idx_str[len(self.prefix):]
+                    if name.endswith(":idx"):
+                        name = name[:-4]
+                    result.append(name)
+            return result
+        except Exception as e:
+            # Only return empty list for connection/auth issues in some contexts
+            # but generally we should propagate errors
+            raise RuntimeError(f"Failed to list collections: {e}") from e
+
+    def insert(
+        self,
+        collection: str,
+        documents: List[KnowledgeDocument],
+    ) -> List[str]:
+        """Insert documents using a non-atomic batch (pipeline)."""
+        client = self._get_client()
+        ids = []
+        batch = Batch(is_atomic=False)
+
+        for doc in documents:
+            if doc.embedding is None:
+                raise ValueError(f"Document {doc.id} has no embedding")
+
+            key = self._doc_key(collection, doc.id)
+            embedding_bytes = struct.pack(f"{len(doc.embedding)}f", *doc.embedding)
+
+            batch.hset(key, {
+                "content": doc.content,
+                "content_hash": doc.content_hash or "",
+                "created_at": str(doc.created_at),
+                "metadata": json.dumps(doc.metadata or {}),
+                "embedding": embedding_bytes,
+            })
+            ids.append(doc.id)
+
+        try:
+            client.exec(batch, raise_on_error=True)
+        except Exception as e:
+            raise RuntimeError(f"Failed to insert documents into Valkey: {e}") from e
+
+        return ids
+
+    def upsert(
+        self,
+        collection: str,
+        documents: List[KnowledgeDocument],
+    ) -> List[str]:
+        """Insert or update documents."""
+        return self.insert(collection, documents)
+
+    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.
+
+        Args:
+            collection: Collection name.
+            query_embedding: Query vector.
+            limit: Max results.
+            filters: Optional dict of field-value equality filters applied as
+                a ValkeySearch pre-filter (e.g. {"content_hash": "abc"}).
+            score_threshold: Minimum similarity (1 - distance) to include.
+                NOTE: Currently only accurate for cosine distance. For euclidean/L2 
+                and inner-product distances, the threshold behavior may be incorrect.
+        """
+        client = self._get_client()
+        index_name = self._index_name(collection)
+        packed = struct.pack(f"{len(query_embedding)}f", *query_embedding)
+
+        # Build pre-filter expression
+        if filters:
+            filter_parts = []
+            for field, value in filters.items():
+                if not _FIELD_NAME_RE.match(field):
+                    raise ValueError(f"Invalid filter field name: {field!r}")
+                if not isinstance(value, str):
+                    raise TypeError(
+                        f"Filter value for {field!r} must be a string, got {type(value).__name__}"
+                    )
+                escaped = _escape_search_value(value)
+                filter_parts.append(f"@{field}:({escaped})")
+            pre_filter = " ".join(filter_parts)
+            query = f"({pre_filter})=>[KNN {limit} @embedding $vec AS score]"
+        else:
+            query = f"*=>[KNN {limit} @embedding $vec AS score]"
+
+        try:
+            options = FtSearchOptions(
+                return_fields=[
+                    ReturnField("content"),
+                    ReturnField("content_hash"),
+                    ReturnField("created_at"),
+                    ReturnField("metadata"),
+                    ReturnField("score"),
+                ],
+                params={"vec": packed},
+            )
+            raw = ft.search(client, index_name, query, options)
+        except Exception as e:
+            raise RuntimeError(f"Failed to search Valkey: {e}") from e
+
+        # ft.search returns [count, {doc_id: {field: value, ...}, ...}]
+        if not raw or len(raw) < 2:
+            return []
+
+        documents = []
+        key_prefix = f"{self.prefix}{collection}:"
+        for doc_id_raw, fields in raw[1].items():
+            doc_id_str = _decode(doc_id_raw)
+            bare_id = doc_id_str.removeprefix(key_prefix)
+
+            field_map: Dict[str, str] = {}
+            for fname, fval in fields.items():
+                fname = _decode(fname)
+                fval = _decode(fval)
+                field_map[fname] = fval
+
+            score = float(field_map.get("score", 0))
+            similarity = 1 - score
+
+            if score_threshold is not None and similarity < score_threshold:
+                continue
+
+            documents.append(KnowledgeDocument(
+                id=bare_id,
+                content=field_map.get("content", ""),
+                embedding=None,
+                metadata=json.loads(field_map.get("metadata", "{}") or "{}"),
+                content_hash=field_map.get("content_hash", ""),
+                created_at=float(field_map.get("created_at", 0)),
+            ))
+
+        return documents
+
+    def get(
+        self,
+        collection: str,
+        ids: List[str],
+    ) -> List[KnowledgeDocument]:
+        """Get documents by IDs using a non-atomic batch (pipeline)."""
+        client = self._get_client()
+        keys = [self._doc_key(collection, doc_id) for doc_id in ids]
+
+        batch = Batch(is_atomic=False)
+        for key in keys:
+            batch.hgetall(key)
+
+        try:
+            results = client.exec(batch, raise_on_error=True)
+        except Exception as e:
+            raise RuntimeError(f"Failed to get documents from Valkey: {e}") from e
+
+        documents = []
+        for doc_id, data in zip(ids, results):
+            if data:
+                embedding = None
+                emb_key = b"embedding" if b"embedding" in data else "embedding"
+                if emb_key in data:
+                    raw_bytes = data[emb_key]
+                    count = len(raw_bytes) // 4
+                    embedding = list(struct.unpack(f"{count}f", raw_bytes))
+
+                content = _decode(data.get(b"content", data.get("content", b"")))
+                content_hash = _decode(data.get(b"content_hash", data.get("content_hash", b"")))
+                created_at_raw = data.get(b"created_at", data.get("created_at", b"0"))
+                created_at = float(_decode(created_at_raw))
+                metadata_raw = _decode(data.get(b"metadata", data.get("metadata", b"{}")))
+                metadata = json.loads(metadata_raw or "{}")
+
+                doc = KnowledgeDocument(
+                    id=doc_id,
+                    content=content,
+                    embedding=embedding,
+                    metadata=metadata,
+                    content_hash=content_hash,
+                    created_at=created_at,
+                )
+                documents.append(doc)
+
+        return documents
+
+    def delete(
+        self,
+        collection: str,
+        ids: Optional[List[str]] = None,
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> int:
+        """Delete documents by IDs. Filter-based deletion is not supported."""
+        if filters and not ids:
+            raise NotImplementedError("Filter-based deletion is not supported by the Valkey backend")
+        if not ids:
+            return 0
+
+        client = self._get_client()
+        keys = [self._doc_key(collection, doc_id) for doc_id in ids]
+
+        try:
+            deleted = client.delete(keys)
+        except Exception as e:
+            raise RuntimeError(f"Failed to delete documents from Valkey: {e}") from e
+
+        return deleted
+
+    def count(self, collection: str) -> int:
+        """Count documents in a collection."""
+        client = self._get_client()
+        index_name = self._index_name(collection)
+        try:
+            info = ft.info(client, index_name)
+            # ft.info returns a Mapping; num_docs may be bytes key
+            for k in (b"num_docs", "num_docs"):
+                if k in info:
+                    return int(info[k])
+            return 0
+        except Exception as e:
+            # Only return 0 for index not found, re-raise other errors
+            err_msg = str(e).lower()
+            if "index not found" in err_msg or "unknown index" in err_msg:
+                return 0
+            raise RuntimeError(f"Failed to count documents: {e}") from e
+
+    def close(self) -> None:
+        """Close the store."""
+        if self._client:
+            try:
+                self._client.close()
+            finally:
+                self._client = None

src/praisonai/praisonai/persistence/registry.py

@@ -198,7 +198,16 @@ def _register_builtin_knowledge_stores(registry: StoreRegistry):
     def _redis_vector(url=None, **kwargs):
         from .knowledge.redis_vector import RedisVectorKnowledgeStore
         return RedisVectorKnowledgeStore(url=url, **kwargs)
-    
+
+    def _valkey_vector(url=None, **kwargs):
+        from .knowledge.valkey_vector import ValkeyVectorKnowledgeStore
+        if url is not None:
+            raise ValueError(
+                "Valkey vector store does not accept 'url' parameter. "
+                "Use host/port/password kwargs instead."
+            )
+        return ValkeyVectorKnowledgeStore(**kwargs)
+
     def _cassandra(url=None, **kwargs):
         kwargs.pop("url", None)  # Cassandra doesn't use url parameter
         from .knowledge.cassandra import CassandraKnowledgeStore
@@ -258,9 +267,10 @@ def _register_builtin_knowledge_stores(registry: StoreRegistry):
     registry.register("milvus", _milvus)
     registry.register("pgvector", _pgvector)
     registry.register("redis", _redis_vector)
+    registry.register("valkey", _valkey_vector)
     registry.register("cassandra", _cassandra)
     registry.register("clickhouse", _clickhouse)
-    registry.register("mongodb_vector", _mongodb_vector, 
+    registry.register("mongodb_vector", _mongodb_vector,
         aliases=("mongodb_atlas", "mongo_vector"))
     
     # Register missing backends that were supported in old factory
@@ -286,7 +296,16 @@ def _register_builtin_state_stores(registry: StoreRegistry):
     def _redis(url=None, **kwargs):
         from .state.redis import RedisStateStore
         return RedisStateStore(url=url, **kwargs)
-    
+
+    def _valkey(url=None, **kwargs):
+        from .state.valkey import ValkeyStateStore
+        if url is not None:
+            raise ValueError(
+                "Valkey state store does not accept 'url' parameter. "
+                "Use host/port/password kwargs instead."
+            )
+        return ValkeyStateStore(**kwargs)
+
     def _dynamodb(url=None, **kwargs):
         kwargs.pop("url", None)  # DynamoDB doesn't use url parameter
         from .state.dynamodb import DynamoDBStateStore
@@ -322,10 +341,11 @@ def _register_builtin_state_stores(registry: StoreRegistry):
         return GCSStateStore(bucket_name=bucket, **kwargs)
     
     registry.register("redis", _redis)
+    registry.register("valkey", _valkey)
     registry.register("dynamodb", _dynamodb)
     registry.register("firestore", _firestore)
     registry.register("mongodb", _mongodb)
-    registry.register("async_mongodb", _async_mongodb, 
+    registry.register("async_mongodb", _async_mongodb,
         aliases=("motor", "mongodb_async"))
     registry.register("upstash", _upstash)
     registry.register("memory", _memory)

src/praisonai/praisonai/persistence/state/valkey.py

@@ -0,0 +1,233 @@
+"""
+Valkey implementation of StateStore.
+
+Requires: valkey-glide-sync
+Install: pip install 'praisonai[valkey]'
+"""
+
+import json
+from typing import Any, Dict, List, Optional
+
+from .base import StateStore
+from .._valkey_client import create_valkey_client
+
+try:
+    from glide_sync import ExpirySet, ExpiryType
+except ImportError:
+    ExpirySet = None
+    ExpiryType = None
+
+
+class ValkeyStateStore(StateStore):
+    """
+    Valkey-based state store for fast key-value operations.
+
+    Example:
+        store = ValkeyStateStore(
+            host="localhost",
+            port=6379,
+        )
+    """
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        password: Optional[str] = None,
+        prefix: str = "praison:",
+        db: int = 0,
+    ):
+        self.host = host
+        self.port = port
+        self.password = password
+        self.prefix = prefix
+        self.db = db
+        self._client = None
+
+    def _get_client(self):
+        """Lazy initialize Valkey client."""
+        if self._client is None:
+            self._client = create_valkey_client(
+                host=self.host,
+                port=self.port,
+                password=self.password,
+                db=self.db,
+            )
+        return self._client
+
+    def _key(self, key: str) -> str:
+        """Add prefix to key."""
+        return f"{self.prefix}{key}"
+
+    def get(self, key: str) -> Optional[Any]:
+        """Get a value by key."""
+        client = self._get_client()
+        try:
+            value = client.get(self._key(key))
+        except Exception as e:
+            raise RuntimeError(f"Failed to get key from Valkey: {e}") from e
+
+        if value is None:
+            return None
+        if isinstance(value, bytes):
+            value = value.decode()
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            return value
+
+    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
+        """Set a value with optional TTL."""
+        client = self._get_client()
+        value = json.dumps(value)
+
+        try:
+            if ttl is not None:
+                if ExpirySet is None or ExpiryType is None:
+                    from .._valkey_client import _MISSING_MSG
+                    raise ImportError(_MISSING_MSG)
+                client.set(self._key(key), value, expiry=ExpirySet(ExpiryType.SEC, ttl))
+            else:
+                client.set(self._key(key), value)
+        except ImportError:
+            raise
+        except Exception as e:
+            raise RuntimeError(f"Failed to set key in Valkey: {e}") from e
+
+    def delete(self, key: str) -> bool:
+        """Delete a key."""
+        client = self._get_client()
+        try:
+            return client.delete([self._key(key)]) > 0
+        except Exception as e:
+            raise RuntimeError(f"Failed to delete key from Valkey: {e}") from e
+
+    def exists(self, key: str) -> bool:
+        """Check if a key exists."""
+        client = self._get_client()
+        try:
+            return client.exists([self._key(key)]) > 0
+        except Exception as e:
+            raise RuntimeError(f"Failed to check key existence in Valkey: {e}") from e
+
+    def keys(self, pattern: str = "*") -> List[str]:
+        """List keys matching pattern."""
+        client = self._get_client()
+        full_pattern = self._key(pattern)
+        try:
+            cursor: bytes = b"0"
+            all_keys = []
+            while True:
+                result = client.scan(cursor, match=full_pattern, count=100)
+                cursor = result[0]
+                all_keys.extend(result[1] or [])
+                if not cursor or cursor in (b"0", "0"):
+                    break
+            prefix_len = len(self.prefix)
+            result_keys = []
+            for k in all_keys:
+                k_str = k.decode() if isinstance(k, bytes) else k
+                result_keys.append(k_str[prefix_len:] if k_str.startswith(self.prefix) else k_str)
+            return result_keys
+        except Exception as e:
+            raise RuntimeError(f"Failed to list keys from Valkey: {e}") from e
+
+    def ttl(self, key: str) -> Optional[int]:
+        """Get remaining TTL in seconds."""
+        client = self._get_client()
+        try:
+            result = client.ttl(self._key(key))
+        except Exception as e:
+            raise RuntimeError(f"Failed to get TTL from Valkey: {e}") from e
+
+        if result is None or result < 0:
+            return None
+        return int(result)
+
+    def expire(self, key: str, ttl: int) -> bool:
+        """Set TTL on existing key."""
+        client = self._get_client()
+        try:
+            return bool(client.expire(self._key(key), ttl))
+        except Exception as e:
+            raise RuntimeError(f"Failed to set expire in Valkey: {e}") from e
+
+    def hget(self, key: str, field: str) -> Optional[Any]:
+        """Get a field from a hash."""
+        client = self._get_client()
+        try:
+            value = client.hget(self._key(key), field)
+        except Exception as e:
+            raise RuntimeError(f"Failed to hget from Valkey: {e}") from e
+
+        if value is None:
+            return None
+        if isinstance(value, bytes):
+            value = value.decode()
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            return value
+
+    def hset(self, key: str, field: str, value: Any) -> None:
+        """Set a field in a hash."""
+        client = self._get_client()
+        value = json.dumps(value)
+        try:
+            client.hset(self._key(key), {field: value})
+        except Exception as e:
+            raise RuntimeError(f"Failed to hset in Valkey: {e}") from e
+
+    def hgetall(self, key: str) -> Dict[str, Any]:
+        """Get all fields from a hash."""
+        client = self._get_client()
+        try:
+            data = client.hgetall(self._key(key))
+        except Exception as e:
+            raise RuntimeError(f"Failed to hgetall from Valkey: {e}") from e
+
+        result = {}
+        if not data:
+            return result
+        for k, v in data.items():
+            k_str = k.decode() if isinstance(k, bytes) else k
+            v_str = v.decode() if isinstance(v, bytes) else v
+            try:
+                result[k_str] = json.loads(v_str)
+            except (json.JSONDecodeError, TypeError):
+                result[k_str] = v_str
+        return result
+
+    def hdel(self, key: str, *fields: str) -> int:
+        """Delete fields from a hash."""
+        if not fields:
+            return 0
+        client = self._get_client()
+        try:
+            return client.hdel(self._key(key), list(fields))
+        except Exception as e:
+            raise RuntimeError(f"Failed to hdel from Valkey: {e}") from e
+
+    def incr(self, key: str, amount: int = 1) -> int:
+        """Increment a counter."""
+        client = self._get_client()
+        try:
+            return client.incrby(self._key(key), amount)
+        except Exception as e:
+            raise RuntimeError(f"Failed to increment key in Valkey: {e}") from e
+
+    def decr(self, key: str, amount: int = 1) -> int:
+        """Decrement a counter."""
+        client = self._get_client()
+        try:
+            return client.decrby(self._key(key), amount)
+        except Exception as e:
+            raise RuntimeError(f"Failed to decrement key in Valkey: {e}") from e
+
+    def close(self) -> None:
+        """Close the store."""
+        if self._client:
+            try:
+                self._client.close()
+            finally:
+                self._client = None

src/praisonai/praisonai/persistence/tests/test_valkey_backends.py

@@ -0,0 +1,492 @@
+"""
+Tests for ValkeyVectorKnowledgeStore, ValkeyStateStore, and registry integration.
+
+Tests cover:
+- ValkeyVectorKnowledgeStore collection and document operations (mocked glide client)
+- ValkeyStateStore key-value operations (mocked glide client)
+- KNOWLEDGE_STORES and STATE_STORES registry entries for "valkey"
+"""
+
+import json
+import struct
+from unittest.mock import MagicMock, patch
+
+from praisonai.persistence.knowledge.base import KnowledgeDocument
+from praisonai.persistence.registry import KNOWLEDGE_STORES, STATE_STORES
+
+
+def _make_knowledge_store():
+    """Return a ValkeyVectorKnowledgeStore with a pre-injected MagicMock client."""
+    from praisonai.persistence.knowledge.valkey_vector import ValkeyVectorKnowledgeStore
+
+    store = ValkeyVectorKnowledgeStore.__new__(ValkeyVectorKnowledgeStore)
+    store.host = "localhost"
+    store.port = 6379
+    store.password = None
+    store.prefix = "praison:vec:"
+    mock_client = MagicMock()
+    store._client = mock_client
+    store._get_client = lambda: mock_client
+
+    return store, mock_client
+
+
+def _make_state_store():
+    """Return a ValkeyStateStore with a pre-injected MagicMock client."""
+    from praisonai.persistence.state.valkey import ValkeyStateStore
+
+    store = ValkeyStateStore.__new__(ValkeyStateStore)
+    store.host = "localhost"
+    store.port = 6379
+    store.password = None
+    store.prefix = "praison:"
+    store.db = 0
+    mock_client = MagicMock()
+    store._client = mock_client
+    store._get_client = lambda: mock_client
+
+    return store, mock_client
+
+
+class TestValkeyVectorKnowledgeStore:
+    """Tests for ValkeyVectorKnowledgeStore."""
+
+    _FT_MODULE = "praisonai.persistence.knowledge.valkey_vector.ft"
+
+    def _mock_ft(self, **method_returns):
+        """Return a mock ft module with configured method returns."""
+        mock_ft = MagicMock()
+        for name, val in method_returns.items():
+            if isinstance(val, Exception):
+                getattr(mock_ft, name).side_effect = val
+            else:
+                getattr(mock_ft, name).return_value = val
+        return mock_ft
+
+    def test_create_collection(self):
+        """Test create_collection calls ft.create with HNSW schema."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(create=b"OK")
+
+        with patch(self._FT_MODULE, mock_ft):
+            store.create_collection("docs", dimension=4)
+
+        mock_ft.create.assert_called_once()
+        assert mock_ft.create.call_args[0][1] == "praison:vec:docs:idx"
+
+    def test_create_collection_already_exists(self):
+        """Test create_collection silently ignores 'already exists' error."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(create=Exception("already exists"))
+
+        with patch(self._FT_MODULE, mock_ft):
+            store.create_collection("docs", dimension=4)  # should not raise
+
+    def test_delete_collection(self):
+        """Test delete_collection calls ft.dropindex and returns True on success."""
+        store, mock_client = _make_knowledge_store()
+        mock_ft = self._mock_ft(dropindex=b"OK")
+        mock_client.scan.return_value = [b"0", []]
+
+        with patch(self._FT_MODULE, mock_ft):
+            result = store.delete_collection("docs")
+
+        assert result is True
+        mock_ft.dropindex.assert_called_once()
+
+    def test_collection_exists_true(self):
+        """Test collection_exists returns True when ft.info succeeds."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(info={b"index_name": b"praison:vec:docs:idx"})
+
+        with patch(self._FT_MODULE, mock_ft):
+            result = store.collection_exists("docs")
+
+        assert result is True
+
+    def test_collection_exists_false(self):
+        """Test collection_exists returns False when ft.info raises an exception."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(info=Exception("Unknown index name"))
+
+        with patch(self._FT_MODULE, mock_ft):
+            result = store.collection_exists("docs")
+
+        assert result is False
+
+    def test_list_collections(self):
+        """Test list_collections returns only names matching the prefix, stripped."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(list=[b"praison:vec:docs:idx", b"other:idx"])
+
+        with patch(self._FT_MODULE, mock_ft):
+            result = store.list_collections()
+
+        assert "docs" in result
+        assert all("other" not in name for name in result)
+
+    def test_insert(self):
+        """Test insert uses Batch pipeline and calls client.exec."""
+        store, mock_client = _make_knowledge_store()
+        mock_client.exec.return_value = [4]
+
+        embedding = [0.1, 0.2, 0.3, 0.4]
+        doc = KnowledgeDocument(id="doc1", content="hello", embedding=embedding)
+        ids = store.insert("docs", [doc])
+
+        mock_client.exec.assert_called_once()
+        assert ids == ["doc1"]
+
+    def test_upsert_delegates_to_insert(self):
+        """Test upsert calls insert (same path)."""
+        store, mock_client = _make_knowledge_store()
+        mock_client.exec.return_value = [4]
+
+        doc = KnowledgeDocument(id="doc1", content="hello", embedding=[0.1, 0.2, 0.3, 0.4])
+        store.upsert("docs", [doc])
+
+        mock_client.exec.assert_called_once()
+
+    def test_search(self):
+        """Test search calls ft.search and returns a list of KnowledgeDocument."""
+        store, _ = _make_knowledge_store()
+        ft_result = [1, {
+            b"praison:vec:docs:doc1": {
+                b"content": b"hello",
+                b"content_hash": b"abc",
+                b"created_at": b"1234.0",
+                b"score": b"0.1",
+            }
+        }]
+        mock_ft = self._mock_ft(search=ft_result)
+
+        with patch(self._FT_MODULE, mock_ft):
+            results = store.search("docs", query_embedding=[0.1, 0.2, 0.3, 0.4], limit=5)
+
+        assert isinstance(results, list)
+        assert len(results) == 1
+        assert isinstance(results[0], KnowledgeDocument)
+        assert results[0].id == "doc1"
+        mock_ft.search.assert_called_once()
+
+    def test_search_with_filters(self):
+        """Test search with filters builds a pre-filter KNN query."""
+        store, _ = _make_knowledge_store()
+        ft_result = [1, {
+            b"praison:vec:docs:doc1": {
+                b"content": b"hello",
+                b"content_hash": b"abc",
+                b"created_at": b"1234.0",
+                b"score": b"0.1",
+            }
+        }]
+        mock_ft = self._mock_ft(search=ft_result)
+
+        with patch(self._FT_MODULE, mock_ft):
+            results = store.search(
+                "docs",
+                query_embedding=[0.1, 0.2, 0.3, 0.4],
+                limit=5,
+                filters={"content_hash": "abc"},
+            )
+
+        assert len(results) == 1
+        query_arg = mock_ft.search.call_args[0][2]
+        assert "KNN" in query_arg
+        assert query_arg.startswith("(")
+
+    def test_search_score_threshold_excludes_low_similarity(self):
+        """Test search excludes documents whose similarity is below score_threshold."""
+        store, _ = _make_knowledge_store()
+        ft_result = [2, {
+            b"praison:vec:docs:doc1": {
+                b"content": b"close",
+                b"content_hash": b"",
+                b"created_at": b"1.0",
+                b"score": b"0.05",  # similarity = 0.95 — above threshold
+            },
+            b"praison:vec:docs:doc2": {
+                b"content": b"far",
+                b"content_hash": b"",
+                b"created_at": b"2.0",
+                b"score": b"0.8",   # similarity = 0.2 — below threshold
+            },
+        }]
+        mock_ft = self._mock_ft(search=ft_result)
+
+        with patch(self._FT_MODULE, mock_ft):
+            results = store.search(
+                "docs",
+                query_embedding=[0.1, 0.2, 0.3, 0.4],
+                limit=5,
+                score_threshold=0.5,
+            )
+
+        assert len(results) == 1
+        assert results[0].id == "doc1"
+
+    def test_search_empty_result(self):
+        """Test search returns an empty list when ft.search returns no documents."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(search=[])
+
+        with patch(self._FT_MODULE, mock_ft):
+            results = store.search("docs", query_embedding=[0.1, 0.2, 0.3, 0.4])
+
+        assert results == []
+
+    def test_get(self):
+        """Test get uses Batch pipeline and returns a KnowledgeDocument."""
+        store, mock_client = _make_knowledge_store()
+        embedding = [0.1, 0.2, 0.3, 0.4]
+        embedding_bytes = struct.pack(f"{len(embedding)}f", *embedding)
+        mock_client.exec.return_value = [{
+            b"content": b"hello",
+            b"content_hash": b"abc",
+            b"created_at": b"1234.0",
+            b"embedding": embedding_bytes,
+        }]
+
+        results = store.get("docs", ["doc1"])
+
+        assert len(results) == 1
+        assert isinstance(results[0], KnowledgeDocument)
+        assert results[0].id == "doc1"
+        mock_client.exec.assert_called_once()
+
+    def test_delete(self):
+        """Test delete calls client.delete and returns count deleted."""
+        store, mock_client = _make_knowledge_store()
+        mock_client.delete.return_value = 2
+
+        result = store.delete("docs", ids=["doc1", "doc2"])
+
+        assert result == 2
+
+    def test_count(self):
+        """Test count calls ft.info and returns integer document count."""
+        store, _ = _make_knowledge_store()
+        mock_ft = self._mock_ft(info={b"num_docs": 7, b"index_name": b"praison:vec:docs:idx"})
+
+        with patch(self._FT_MODULE, mock_ft):
+            result = store.count("docs")
+
+        assert isinstance(result, int)
+        assert result == 7
+
+    def test_close_closes_client(self):
+        """Test close() calls client.close() and clears the reference."""
+        store, mock_client = _make_knowledge_store()
+
+        store.close()
+
+        mock_client.close.assert_called_once()
+        assert store._client is None
+
+
+class TestValkeyStateStore:
+    """Tests for ValkeyStateStore."""
+
+    def test_get_existing(self):
+        """Test get returns a dict when client.get returns JSON bytes."""
+        store, mock_client = _make_state_store()
+        data = {"result": "ok"}
+        mock_client.get.return_value = json.dumps(data).encode("utf-8")
+
+        result = store.get("mykey")
+
+        mock_client.get.assert_called_once_with("praison:mykey")
+        assert result == data
+
+    def test_get_missing(self):
+        """Test get returns None when client.get returns None."""
+        store, mock_client = _make_state_store()
+        mock_client.get.return_value = None
+
+        result = store.get("missing")
+
+        assert result is None
+
+    def test_set_no_ttl(self):
+        """Test set calls client.set with prefixed key when no TTL is given."""
+        store, mock_client = _make_state_store()
+        mock_client.set.return_value = b"OK"
+
+        store.set("mykey", {"x": 1})
+
+        mock_client.set.assert_called_once()
+        call_args = mock_client.set.call_args[0]
+        assert call_args[0] == "praison:mykey"
+
+    def test_set_with_ttl(self):
+        """Test set with TTL passes SetOptions to client.set (ImportError/RuntimeError tolerated)."""
+        store, mock_client = _make_state_store()
+        mock_client.set.return_value = b"OK"
+
+        # glide's ExpirySet/ExpiryType are lazy-imported inside set(); without a live
+        # glide install this raises ImportError — both paths are valid in CI.
+        try:
+            store.set("ttlkey", {"x": 1}, ttl=300)
+        except (ImportError, RuntimeError):
+            pass
+
+    def test_delete_existing(self):
+        """Test delete returns True when client.delete returns 1."""
+        store, mock_client = _make_state_store()
+        mock_client.delete.return_value = 1
+
+        result = store.delete("mykey")
+
+        mock_client.delete.assert_called_once_with(["praison:mykey"])
+        assert result is True
+
+    def test_delete_missing(self):
+        """Test delete returns False when client.delete returns 0."""
+        store, mock_client = _make_state_store()
+        mock_client.delete.return_value = 0
+
+        result = store.delete("absent")
+
+        assert result is False
+
+    def test_exists_true(self):
+        """Test exists returns True when client.exists returns 1."""
+        store, mock_client = _make_state_store()
+        mock_client.exists.return_value = 1
+
+        assert store.exists("mykey") is True
+        mock_client.exists.assert_called_once_with(["praison:mykey"])
+
+    def test_exists_false(self):
+        """Test exists returns False when client.exists returns 0."""
+        store, mock_client = _make_state_store()
+        mock_client.exists.return_value = 0
+
+        assert store.exists("absent") is False
+
+    def test_keys(self):
+        """Test keys returns stripped key names via client.scan."""
+        store, mock_client = _make_state_store()
+        # scan() returns [cursor, [keys]]; cursor b"0" signals end of iteration
+        mock_client.scan.return_value = [b"0", [b"praison:key1", b"praison:key2"]]
+
+        result = store.keys()
+
+        mock_client.scan.assert_called_once()
+        assert result == ["key1", "key2"]
+
+    def test_ttl_with_value(self):
+        """Test ttl returns seconds remaining when client.ttl returns a positive value."""
+        store, mock_client = _make_state_store()
+        mock_client.ttl.return_value = 30
+
+        result = store.ttl("mykey")
+
+        assert result == 30
+
+    def test_ttl_no_ttl(self):
+        """Test ttl returns None when client.ttl returns -1 (no expiry)."""
+        store, mock_client = _make_state_store()
+        mock_client.ttl.return_value = -1
+
+        result = store.ttl("mykey")
+
+        assert result is None
+
+    def test_expire(self):
+        """Test expire calls client.expire and returns its result."""
+        store, mock_client = _make_state_store()
+        mock_client.expire.return_value = True
+
+        result = store.expire("mykey", 600)
+
+        mock_client.expire.assert_called_once_with("praison:mykey", 600)
+        assert result is True
+
+    def test_hget(self):
+        """Test hget returns a parsed dict when client.hget returns JSON bytes."""
+        store, mock_client = _make_state_store()
+        data = {"field_val": 42}
+        mock_client.hget.return_value = json.dumps(data).encode("utf-8")
+
+        result = store.hget("myhash", "field1")
+
+        mock_client.hget.assert_called_once_with("praison:myhash", "field1")
+        assert result == data
+
+    def test_hset(self):
+        """Test hset calls client.hset with the prefixed key and a mapping dict."""
+        store, mock_client = _make_state_store()
+        mock_client.hset.return_value = 1
+
+        store.hset("myhash", "field1", "value1")
+
+        mock_client.hset.assert_called_once()
+        call_args = mock_client.hset.call_args[0]
+        assert call_args[0] == "praison:myhash"
+        assert isinstance(call_args[1], dict)
+        assert "field1" in call_args[1]
+
+    def test_hgetall(self):
+        """Test hgetall returns a decoded dict when client.hgetall returns bytes."""
+        store, mock_client = _make_state_store()
+        mock_client.hgetall.return_value = {
+            b"field1": b'"value1"',
+            b"field2": b"42",
+        }
+
+        result = store.hgetall("myhash")
+
+        mock_client.hgetall.assert_called_once_with("praison:myhash")
+        assert "field1" in result
+        assert "field2" in result
+
+    def test_hdel(self):
+        """Test hdel calls client.hdel and returns the count of deleted fields."""
+        store, mock_client = _make_state_store()
+        mock_client.hdel.return_value = 2
+
+        result = store.hdel("myhash", "field1", "field2")
+
+        assert result == 2
+
+    def test_incr(self):
+        """Test incr calls client.incrby and returns the new value."""
+        store, mock_client = _make_state_store()
+        mock_client.incrby.return_value = 5
+
+        result = store.incr("counter")
+
+        assert result == 5
+        mock_client.incrby.assert_called_once_with("praison:counter", 1)
+
+    def test_decr(self):
+        """Test decr calls client.decrby and returns the new value."""
+        store, mock_client = _make_state_store()
+        mock_client.decrby.return_value = 3
+
+        result = store.decr("counter")
+
+        assert result == 3
+        mock_client.decrby.assert_called_once_with("praison:counter", 1)
+
+    def test_close_closes_client(self):
+        """Test close() calls client.close() and clears the reference."""
+        store, mock_client = _make_state_store()
+
+        store.close()
+
+        mock_client.close.assert_called_once()
+        assert store._client is None
+
+
+class TestRegistryValkey:
+    """Tests for Valkey backend registration in store registries."""
+
+    def test_knowledge_store_valkey_registered(self):
+        """Test KNOWLEDGE_STORES registry contains 'valkey'."""
+        assert "valkey" in KNOWLEDGE_STORES.list_registered()
+
+    def test_state_store_valkey_registered(self):
+        """Test STATE_STORES registry contains 'valkey'."""
+        assert "valkey" in STATE_STORES.list_registered()

src/praisonai/praisonai/storage/__init__.py

@@ -6,6 +6,7 @@ Heavy implementations that follow StorageBackendProtocol for:
 - MongoDB (praisonai[mongodb])
 - PostgreSQL (praisonai[postgresql])
 - DynamoDB (praisonai[dynamodb])
+- Valkey (praisonai[valkey])
 
 These implementations are kept in the wrapper to avoid bloating the core SDK.
 """
@@ -13,9 +14,12 @@ These implementations are kept in the wrapper to avoid bloating the core SDK.
 # Lazy imports - only import when needed
 __all__ = [
     "RedisStorageAdapter",
-    "MongoDBStorageAdapter", 
+    "MongoDBStorageAdapter",
     "PostgreSQLStorageAdapter",
     "DynamoDBStorageAdapter",
+    "ValkeyStorageAdapter",
+    "ValkeySearchBackend",
+    "ValkeyBackend",
 ]
 
 def __getattr__(name: str):
@@ -32,5 +36,40 @@ def __getattr__(name: str):
     elif name == "DynamoDBStorageAdapter":
         from .dynamodb_adapter import DynamoDBStorageAdapter
         return DynamoDBStorageAdapter
+    elif name == "ValkeyStorageAdapter":
+        from .valkey_adapter import ValkeyStorageAdapter
+        return ValkeyStorageAdapter
+    elif name == "ValkeySearchBackend":
+        from .valkey_adapter import ValkeySearchBackend
+        return ValkeySearchBackend
+    elif name == "ValkeyBackend":
+        cls = _make_valkey_backend_class()
+        globals()["ValkeyBackend"] = cls
+        return cls
     else:
-        raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+        raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
+
+def _make_valkey_backend_class():
+    """Return ValkeyBackend — ValkeyStorageAdapter pre-configured from env vars."""
+    import os
+    from .valkey_adapter import ValkeyStorageAdapter
+
+    class ValkeyBackend(ValkeyStorageAdapter):
+        def __init__(
+            self,
+            host: str = None,
+            port: int = None,
+            prefix: str = None,
+            ttl=None,
+            password: str = None,
+        ):
+            super().__init__(
+                host=host if host is not None else os.environ.get("VALKEY_HOST", "localhost"),
+                port=int(port if port is not None else os.environ.get("VALKEY_PORT", 6379)),
+                prefix=prefix if prefix is not None else os.environ.get("VALKEY_PREFIX", "praisonai:"),
+                ttl=ttl,
+                password=password if password is not None else os.environ.get("VALKEY_PASSWORD"),
+            )
+
+    return ValkeyBackend

src/praisonai/praisonai/storage/valkey_adapter.py

@@ -0,0 +1,348 @@
+"""
+Valkey Storage Adapter for PraisonAI.
+
+Implements StorageBackendProtocol using Valkey for high-speed storage.
+This is the wrapper implementation that contains the heavy valkey-glide-sync dependency.
+"""
+
+import json
+import struct
+from typing import Dict, Any, List, Optional
+
+from praisonai.persistence._valkey_client import create_valkey_client
+
+try:
+    from glide_sync import ExpirySet, ExpiryType
+except ImportError:
+    ExpirySet = None
+    ExpiryType = None
+
+try:
+    from glide_sync import ft
+    from glide_sync import (
+        VectorField, VectorFieldAttributesHnsw, VectorAlgorithm,
+        DistanceMetricType, VectorType, FtCreateOptions, DataType,
+        FtSearchOptions, ReturnField,
+    )
+except ImportError:
+    ft = None
+    VectorField = VectorFieldAttributesHnsw = VectorAlgorithm = None
+    DistanceMetricType = VectorType = FtCreateOptions = DataType = None
+    FtSearchOptions = ReturnField = None
+
+
+class ValkeyStorageAdapter:
+    """
+    Valkey-based storage backend adapter.
+
+    Uses Valkey for high-speed caching and ephemeral data storage.
+    Implements StorageBackendProtocol from praisonaiagents.storage.protocols.
+
+    Example:
+        ```python
+        from praisonai.storage import ValkeyStorageAdapter
+
+        adapter = ValkeyStorageAdapter(host="localhost", port=6379)
+        adapter.save("session_123", {"messages": []})
+        data = adapter.load("session_123")
+        ```
+    """
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        prefix: str = "praisonai:",
+        ttl: Optional[int] = None,
+        password: Optional[str] = None,
+    ):
+        """
+        Initialize the Valkey storage adapter.
+
+        Args:
+            host: Valkey server host
+            port: Valkey server port
+            prefix: Key prefix for all stored data
+            ttl: Optional TTL in seconds for all keys
+            password: Optional Valkey password
+        """
+        self.host = host
+        self.port = port
+        self.prefix = prefix
+        self.ttl = ttl
+        self.password = password
+        self._client = None
+
+    def _get_client(self):
+        """Lazy initialize Valkey client."""
+        if self._client is None:
+            self._client = create_valkey_client(
+                host=self.host,
+                port=self.port,
+                password=self.password,
+            )
+        return self._client
+
+    def _make_key(self, key: str) -> str:
+        """Create prefixed key."""
+        return f"{self.prefix}{key}"
+
+    def _scan_keys(self, pattern: str) -> List[bytes]:
+        """Collect all keys matching pattern via SCAN."""
+        client = self._get_client()
+        cursor: bytes = b"0"
+        all_keys: List[bytes] = []
+        while True:
+            result = client.scan(cursor, match=pattern, count=100)
+            cursor = result[0]
+            all_keys.extend(result[1] or [])
+            if not cursor or cursor in (b"0", "0", 0):
+                break
+        return all_keys
+
+    def save(self, key: str, data: Dict[str, Any]) -> None:
+        """Save data with the given key."""
+        client = self._get_client()
+        full_key = self._make_key(key)
+        json_data = json.dumps(data, default=str, ensure_ascii=False).encode('utf-8')
+
+        try:
+            if self.ttl is not None:
+                if ExpirySet is None or ExpiryType is None:
+                    from praisonai.persistence._valkey_client import _MISSING_MSG
+                    raise ImportError(_MISSING_MSG)
+                client.set(full_key, json_data, expiry=ExpirySet(ExpiryType.SEC, self.ttl))
+            else:
+                client.set(full_key, json_data)
+        except ImportError:
+            raise
+        except Exception as e:
+            raise RuntimeError(f"Failed to save data to Valkey: {e}") from e
+
+    def load(self, key: str) -> Optional[Dict[str, Any]]:
+        """Load data by key."""
+        client = self._get_client()
+        full_key = self._make_key(key)
+
+        try:
+            value = client.get(full_key)
+            if value is not None:
+                try:
+                    if isinstance(value, bytes):
+                        value = value.decode('utf-8')
+                    return json.loads(value)
+                except json.JSONDecodeError as e:
+                    raise ValueError(f"Invalid JSON data for key '{key}': {e}") from e
+            return None
+        except ValueError:
+            raise
+        except Exception as e:
+            raise RuntimeError(f"Failed to load data from Valkey: {e}") from e
+
+    def delete(self, key: str) -> bool:
+        """Delete data by key."""
+        client = self._get_client()
+        full_key = self._make_key(key)
+
+        try:
+            return client.delete([full_key]) > 0
+        except Exception as e:
+            raise RuntimeError(f"Failed to delete data from Valkey: {e}") from e
+
+    def list_keys(self, prefix: str = "") -> List[str]:
+        """List all keys, optionally filtered by prefix."""
+        pattern = self._make_key(f"{prefix}*")
+
+        try:
+            raw_keys = self._scan_keys(pattern)
+            prefix_len = len(self.prefix)
+            return sorted(
+                (k.decode('utf-8') if isinstance(k, bytes) else k)[prefix_len:]
+                for k in raw_keys
+            )
+        except Exception as e:
+            raise RuntimeError(f"Failed to list keys from Valkey: {e}") from e
+
+    def exists(self, key: str) -> bool:
+        """Check if a key exists."""
+        client = self._get_client()
+        full_key = self._make_key(key)
+
+        try:
+            return client.exists([full_key]) > 0
+        except Exception as e:
+            raise RuntimeError(f"Failed to check key existence in Valkey: {e}") from e
+
+    def clear(self) -> int:
+        """Clear all data with our prefix. Returns number of items deleted."""
+        client = self._get_client()
+        pattern = self._make_key("*")
+        batch_size = 500
+
+        try:
+            all_keys = self._scan_keys(pattern)
+            if not all_keys:
+                return 0
+            total = 0
+            for i in range(0, len(all_keys), batch_size):
+                total += client.delete(all_keys[i:i + batch_size])
+            return total
+        except Exception as e:
+            raise RuntimeError(f"Failed to clear data from Valkey: {e}") from e
+
+    def set_ttl(self, key: str, ttl: int) -> bool:
+        """Set TTL on a specific key."""
+        client = self._get_client()
+        full_key = self._make_key(key)
+
+        try:
+            return client.expire(full_key, ttl)
+        except Exception as e:
+            raise RuntimeError(f"Failed to set TTL in Valkey: {e}") from e
+
+    def ping(self) -> bool:
+        """Test connection to Valkey."""
+        try:
+            client = self._get_client()
+            result = client.ping()
+            return result is not None
+        except Exception:
+            return False
+
+    def close(self) -> None:
+        """Close the Valkey connection."""
+        if self._client:
+            try:
+                self._client.close()
+            finally:
+                self._client = None
+
+
+class ValkeySearchBackend:
+    """
+    Valkey vector search backend using RediSearch / ValkeySearch FT commands.
+
+    Example:
+        ```python
+        from praisonai.storage import ValkeySearchBackend
+
+        backend = ValkeySearchBackend(host="localhost", port=6379, vector_dim=1536)
+        backend.create_index()
+        backend.add_document("doc1", "hello world", embedding)
+        results = backend.search(query_embedding, k=5)
+        ```
+    """
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        index_name: str = "praisonai_vectors",
+        password: Optional[str] = None,
+        vector_dim: int = 1536,
+    ):
+        """
+        Initialize the Valkey search backend.
+
+        Args:
+            host: Valkey server host
+            port: Valkey server port
+            index_name: Name of the FT index
+            password: Optional Valkey password
+            vector_dim: Dimension of embedding vectors
+        """
+        self.host = host
+        self.port = port
+        self.index_name = index_name
+        self.password = password
+        self.vector_dim = vector_dim
+        self._client = None
+
+    def _get_client(self):
+        """Lazy initialize Valkey client."""
+        if self._client is None:
+            self._client = create_valkey_client(
+                host=self.host,
+                port=self.port,
+                password=self.password,
+            )
+        return self._client
+
+    def create_index(self, vector_dim: int = None) -> None:
+        """Create the vector search index (no-op if already exists)."""
+        client = self._get_client()
+        dim = vector_dim or self.vector_dim
+
+        try:
+            schema = [VectorField(
+                "embedding",
+                VectorAlgorithm.HNSW,
+                VectorFieldAttributesHnsw(dim, DistanceMetricType.COSINE, VectorType.FLOAT32),
+            )]
+            options = FtCreateOptions(DataType.HASH, prefixes=[f"{self.index_name}:"])
+            ft.create(client, self.index_name, schema, options)
+        except Exception as e:
+            err = str(e)
+            if "already exists" not in err.lower() and "index already exists" not in err.lower():
+                raise RuntimeError(f"Failed to create Valkey search index: {e}") from e
+
+    def add_document(self, doc_id: str, text: str, embedding: List[float]) -> None:
+        """Add a document with its embedding to the index."""
+        if len(embedding) != self.vector_dim:
+            raise ValueError(
+                f"Embedding dimension mismatch: expected {self.vector_dim}, got {len(embedding)}"
+            )
+        client = self._get_client()
+        full_id = f"{self.index_name}:{doc_id}"
+        packed = struct.pack(f"{len(embedding)}f", *embedding)
+
+        try:
+            client.hset(full_id, {"text": text, "embedding": packed})
+        except Exception as e:
+            raise RuntimeError(f"Failed to add document to Valkey: {e}") from e
+
+    def search(self, query_embedding: List[float], k: int = 5) -> List[Dict[str, Any]]:
+        """Search for nearest neighbors by embedding."""
+        if k <= 0:
+            raise ValueError("k must be > 0")
+        if len(query_embedding) != self.vector_dim:
+            raise ValueError(
+                f"Query embedding dimension mismatch: expected {self.vector_dim}, got {len(query_embedding)}"
+            )
+        client = self._get_client()
+        packed = struct.pack(f"{len(query_embedding)}f", *query_embedding)
+
+        try:
+            options = FtSearchOptions(
+                return_fields=[ReturnField("text"), ReturnField("score")],
+                params={"vec": packed},
+            )
+            raw = ft.search(client, self.index_name, f"*=>[KNN {k} @embedding $vec AS score]", options)
+        except Exception as e:
+            raise RuntimeError(f"Failed to search Valkey: {e}") from e
+
+        # ft.search returns [count, {doc_id: {field: value, ...}, ...}]
+        if not raw or len(raw) < 2:
+            return []
+
+        results = []
+        key_prefix = f"{self.index_name}:"
+        for doc_id, fields in raw[1].items():
+            doc: Dict[str, Any] = {}
+            raw_id = doc_id.decode('utf-8') if isinstance(doc_id, bytes) else doc_id
+            doc["id"] = raw_id.removeprefix(key_prefix)
+            for fname, fval in fields.items():
+                fname = fname.decode('utf-8') if isinstance(fname, bytes) else fname
+                fval = fval.decode('utf-8') if isinstance(fval, bytes) else fval
+                doc[fname] = fval
+            results.append(doc)
+
+        return results
+
+    def close(self) -> None:
+        """Close the Valkey connection."""
+        if self._client:
+            try:
+                self._client.close()
+            finally:
+                self._client = None

src/praisonai/pyproject.toml

@@ -180,6 +180,9 @@ postgresql = [
 dynamodb = [
     "boto3>=1.35.0",
 ]
+valkey = [
+    "valkey-glide-sync>=2.3.1",
+]
 # Cloud-native serverless databases
 neon = [
     "psycopg2-binary>=2.9.0",

src/praisonai/tests/unit/storage/test_valkey_backend.py

@@ -0,0 +1,249 @@
+"""
+Tests for ValkeyStorageAdapter and ValkeySearchBackend.
+
+Tests cover:
+- ValkeyStorageAdapter CRUD operations (mocked glide client)
+- ValkeySearchBackend index and search operations (mocked)
+"""
+
+import json
+from unittest.mock import MagicMock, patch
+
+
+def _make_adapter(prefix="praisonai:", ttl=None):
+    """Return a ValkeyStorageAdapter with a pre-injected MagicMock client."""
+    from praisonai.storage.valkey_adapter import ValkeyStorageAdapter
+
+    adapter = ValkeyStorageAdapter.__new__(ValkeyStorageAdapter)
+    adapter.host = "localhost"
+    adapter.port = 6379
+    adapter.prefix = prefix
+    adapter.ttl = ttl
+    adapter.password = None
+
+    mock_client = MagicMock()
+    adapter._client = mock_client
+    adapter._get_client = lambda: mock_client
+
+    return adapter, mock_client
+
+
+def _make_search_backend(index_name="praisonai_vectors", vector_dim=4):
+    """Return a ValkeySearchBackend with a pre-injected MagicMock client."""
+    from praisonai.storage.valkey_adapter import ValkeySearchBackend
+
+    backend = ValkeySearchBackend.__new__(ValkeySearchBackend)
+    backend.host = "localhost"
+    backend.port = 6379
+    backend.index_name = index_name
+    backend.vector_dim = vector_dim
+    backend.password = None
+
+    mock_client = MagicMock()
+    backend._client = mock_client
+    backend._get_client = lambda: mock_client
+
+    return backend, mock_client
+
+
+class TestValkeyStorageAdapter:
+    """Tests for ValkeyStorageAdapter."""
+
+    def test_valkey_adapter_save(self):
+        """Test save calls client.set with prefixed key and JSON bytes."""
+        adapter, mock_client = _make_adapter()
+
+        data = {"name": "test", "value": 42}
+        adapter.save("mykey", data)
+
+        mock_client.set.assert_called_once()
+        call_args = mock_client.set.call_args[0]
+        assert call_args[0] == "praisonai:mykey"
+        assert json.loads(call_args[1]) == data
+
+    def test_valkey_adapter_save_with_ttl(self):
+        """Test save with TTL calls client.set with expiry kwarg (ImportError tolerated without glide)."""
+        adapter, mock_client = _make_adapter(ttl=300)
+
+        # ExpirySet/ExpiryType are lazy-imported inside save(); without a live
+        # glide install this raises RuntimeError — both paths are valid in CI.
+        try:
+            adapter.save("ttlkey", {"key": "val"})
+            mock_client.set.assert_called_once()
+            _, kwargs = mock_client.set.call_args
+            assert "expiry" in kwargs
+        except (ImportError, RuntimeError):
+            pass
+
+    def test_valkey_adapter_load_existing(self):
+        """Test load returns a dict when client.get returns JSON bytes."""
+        adapter, mock_client = _make_adapter()
+
+        data = {"result": "ok"}
+        mock_client.get.return_value = json.dumps(data).encode("utf-8")
+
+        loaded = adapter.load("mykey")
+
+        mock_client.get.assert_called_once_with("praisonai:mykey")
+        assert loaded == data
+
+    def test_valkey_adapter_load_missing(self):
+        """Test load returns None when client.get returns None."""
+        adapter, mock_client = _make_adapter()
+        mock_client.get.return_value = None
+
+        loaded = adapter.load("missing")
+
+        assert loaded is None
+
+    def test_valkey_adapter_delete_existing(self):
+        """Test delete returns True when client.delete returns 1."""
+        adapter, mock_client = _make_adapter()
+        mock_client.delete.return_value = 1
+
+        result = adapter.delete("mykey")
+
+        mock_client.delete.assert_called_once_with(["praisonai:mykey"])
+        assert result is True
+
+    def test_valkey_adapter_delete_missing(self):
+        """Test delete returns False when client.delete returns 0."""
+        adapter, mock_client = _make_adapter()
+        mock_client.delete.return_value = 0
+
+        result = adapter.delete("absent")
+
+        assert result is False
+
+    def test_valkey_adapter_list_keys(self):
+        """Test list_keys returns stripped, sorted key names via client.scan."""
+        adapter, mock_client = _make_adapter()
+        raw_keys = [b"praisonai:beta", b"praisonai:alpha", b"praisonai:gamma"]
+        # scan() returns [cursor, [keys]]; cursor b"0" signals end of iteration
+        mock_client.scan.return_value = [b"0", raw_keys]
+
+        keys = adapter.list_keys()
+
+        mock_client.scan.assert_called_once()
+        assert keys == ["alpha", "beta", "gamma"]
+
+    def test_valkey_adapter_exists_true(self):
+        """Test exists returns True when client.exists returns 1."""
+        adapter, mock_client = _make_adapter()
+        mock_client.exists.return_value = 1
+
+        assert adapter.exists("mykey") is True
+        mock_client.exists.assert_called_once_with(["praisonai:mykey"])
+
+    def test_valkey_adapter_exists_false(self):
+        """Test exists returns False when client.exists returns 0."""
+        adapter, mock_client = _make_adapter()
+        mock_client.exists.return_value = 0
+
+        assert adapter.exists("absent") is False
+
+    def test_valkey_adapter_clear(self):
+        """Test clear deletes all prefixed keys and returns count."""
+        adapter, mock_client = _make_adapter()
+        raw_keys = [b"praisonai:a", b"praisonai:b", b"praisonai:c"]
+        # scan() returns [cursor, [keys]]; cursor b"0" signals end of iteration
+        mock_client.scan.return_value = [b"0", raw_keys]
+        mock_client.delete.return_value = 3
+
+        count = adapter.clear()
+
+        assert count == 3
+        mock_client.delete.assert_called_once()
+
+    def test_valkey_adapter_set_ttl(self):
+        """Test set_ttl calls client.expire with prefixed key and returns result."""
+        adapter, mock_client = _make_adapter()
+        mock_client.expire.return_value = True
+
+        result = adapter.set_ttl("mykey", 600)
+
+        mock_client.expire.assert_called_once_with("praisonai:mykey", 600)
+        assert result is True
+
+    def test_valkey_adapter_import_error(self):
+        """Test ImportError with 'valkey' hint when glide package is not available."""
+        import praisonai.persistence._valkey_client as client_mod
+        from praisonai.storage.valkey_adapter import ValkeyStorageAdapter
+
+        adapter = ValkeyStorageAdapter.__new__(ValkeyStorageAdapter)
+        adapter.host = "localhost"
+        adapter.port = 6379
+        adapter.prefix = "praisonai:"
+        adapter.ttl = None
+        adapter.password = None
+        adapter._client = None
+
+        original = client_mod.GlideClientSync
+        try:
+            client_mod.GlideClientSync = None
+            try:
+                adapter._get_client()
+                assert False, "Should have raised ImportError"
+            except ImportError as e:
+                assert "valkey" in str(e).lower()
+        finally:
+            client_mod.GlideClientSync = original
+
+    def test_valkey_adapter_operation_error(self):
+        """Test RuntimeError raised when a client operation fails."""
+        adapter, mock_client = _make_adapter()
+        mock_client.get.side_effect = Exception("connection refused")
+
+        try:
+            adapter.load("anykey")
+            assert False, "Should have raised RuntimeError"
+        except RuntimeError as e:
+            assert "Valkey" in str(e) or "valkey" in str(e).lower()
+
+
+class TestValkeySearchBackend:
+    """Tests for ValkeySearchBackend."""
+
+    def test_create_index(self):
+        """Test create_index uses ft.create (no-op if already exists)."""
+        backend, _ = _make_search_backend()
+        mock_ft = MagicMock()
+        mock_ft.create.return_value = b"OK"
+
+        with patch("praisonai.storage.valkey_adapter.ft", mock_ft):
+            backend.create_index()
+
+        mock_ft.create.assert_called_once()
+        assert backend.index_name == mock_ft.create.call_args[0][1]
+
+    def test_add_document(self):
+        """Test add_document calls hset with correct key and fields."""
+        backend, mock_client = _make_search_backend()
+        mock_client.hset.return_value = 2
+
+        backend.add_document("doc1", "hello world", [0.1, 0.2, 0.3, 0.4])
+
+        mock_client.hset.assert_called_once()
+        call_args = mock_client.hset.call_args[0]
+        assert call_args[0] == "praisonai_vectors:doc1"
+        mapping = call_args[1]
+        assert "text" in mapping
+        assert mapping["text"] == "hello world"
+        assert "embedding" in mapping
+
+    def test_search(self):
+        """Test search uses ft.search and returns a list of dicts."""
+        backend, _ = _make_search_backend()
+        ft_result = [1, {b"praisonai_vectors:doc1": {b"text": b"hello world", b"score": b"0.05"}}]
+        mock_ft = MagicMock()
+        mock_ft.search.return_value = ft_result
+
+        with patch("praisonai.storage.valkey_adapter.ft", mock_ft):
+            results = backend.search([0.1, 0.2, 0.3, 0.4], k=5)
+
+        assert isinstance(results, list)
+        assert len(results) == 1
+        assert results[0]["id"] == "doc1"
+        assert results[0]["text"] == "hello world"
+        mock_ft.search.assert_called_once()
+