Python: fix(redis): let an injected redis_client take precedence in RedisHistoryProvider

python/packages/redis/agent_framework_redis/_context_provider.py

@@ -10,6 +10,7 @@
 
 import hashlib
 import json
+import re
 import sys
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self
 
@@ -24,6 +25,7 @@
 from redisvl.query import AggregateHybridQuery, FilterQuery, TextQuery
 from redisvl.query.filter import FilterExpression
 from redisvl.redis.utils import array_to_buffer
+from redisvl.utils.token_escaper import TokenEscaper
 from redisvl.utils.vectorize import BaseVectorizer
 
 from ._feature_usage import FeatureIndex
@@ -43,6 +45,14 @@
 _TAG_DEFAULTS = {"separator": ",", "case_sensitive": False, "withsuffixtrie": False}
 _TEXT_DEFAULTS = {"weight": 1.0, "no_stem": False}
 
+# RediSearch reads several punctuation marks as query syntax — `|` unions terms, `?` is a fuzzy
+# marker, backticks quote exact phrases — so a user message like "ps aux | grep python" or
+# "Why ?" reaches the engine as a malformed query and raises a syntax error that fails the whole
+# agent run. The search text is plain prose, not a query: escape the syntax characters before it
+# reaches TextQuery / AggregateHybridQuery. Unlike the default TokenEscaper set (built for tag
+# filter values), spaces stay unescaped so the terms keep tokenizing normally.
+_SEARCH_TEXT_ESCAPER = TokenEscaper(re.compile(r"[,.<>{}\[\]\\\"':;!@#$%^&*()\-+=~/|?`]"))
+
 
 def _scope_token_field(name: str) -> str:
     if name not in _SCOPE_FIELDS:
@@ -405,7 +415,7 @@ async def _redis_search(
         await self._ensure_index()
         self._validate_filters()
 
-        q = (text or "").strip()
+        q = _SEARCH_TEXT_ESCAPER.escape((text or "").strip())
         if not q:
             raise IntegrationInvalidRequestException("text_search() requires non-empty text")
         num_results = max(int(num_results or 10), 1)

python/packages/redis/agent_framework_redis/_history_provider.py

@@ -95,16 +95,17 @@ def __init__(
         Args:
             source_id: Unique identifier for this provider instance.
             redis_url: Redis connection URL (e.g., "redis://localhost:6379").
-                Mutually exclusive with credential_provider and redis_client.
+                Ignored when redis_client is supplied.
             credential_provider: Redis credential provider for Azure AD authentication.
-                Requires host parameter. Mutually exclusive with redis_url and redis_client.
+                Requires host parameter. Ignored when redis_client is supplied.
             host: Redis host name. Required when using credential_provider.
             port: Redis port number. Defaults to 6380 (Azure Redis SSL port).
             ssl: Enable SSL/TLS connection. Defaults to True.
             username: Redis username.
             redis_client: Borrowed standalone async Redis client configured with
-                ``decode_responses=True``. The caller owns its lifetime. Mutually exclusive
-                with redis_url and credential_provider.
+                ``decode_responses=True``. The caller owns its lifetime and takes
+                precedence over redis_url, credential_provider, and connection
+                defaults, mirroring the Redis vector-store constructors.
             key_prefix: Base prefix for Redis keys. Scoped mode appends independently encoded
                 tenant, application, agent, provider source, and session segments.
                 Defaults to 'chat_messages'.
@@ -130,8 +131,8 @@ def __init__(
         Raises:
             TypeError: If redis_client is not a standalone async Redis client.
             ValueError: If no Redis connection source is provided.
-            ValueError: If more than one Redis connection source is provided.
-            ValueError: If credential_provider is used without host parameter.
+            ValueError: If more than one non-client Redis connection source is provided.
+            ValueError: If credential_provider is used without host parameter and no client is provided.
             ValueError: If redis_client does not use decoded string responses.
             ValueError: If max_messages is negative.
             ValueError: If key_format or its scoped identifiers are invalid.
@@ -148,9 +149,9 @@ def __init__(
         connection_sources = sum(source is not None for source in (redis_client, redis_url, credential_provider))
         if connection_sources == 0:
             raise ValueError("Either redis_client, redis_url, or credential_provider must be provided")
-        if connection_sources > 1:
+        if connection_sources > 1 and redis_client is None:
             raise ValueError("redis_client, redis_url, and credential_provider are mutually exclusive")
-        if credential_provider is not None and host is None:
+        if credential_provider is not None and host is None and redis_client is None:
             raise ValueError("host is required when using credential_provider")
         if redis_client is not None:
             if not isinstance(redis_client, redis.Redis):

python/packages/redis/tests/test_context_provider_edges.py

@@ -353,6 +353,68 @@ async def test_redis_search_combines_explicit_filter_expression(
     mock_index.query.assert_awaited_once()
 
 
+@pytest.mark.parametrize(
+    ("message", "broken_shape", "escaped"),
+    [
+        ("Why ?", "| ?)", r"\?"),
+        ("ps aux | grep python", "| |)", r"\|"),
+        ("Is it A | B?", "| |)", r"\|"),
+        ("Run ` ls ` please", "| ` |", r"\`"),
+        ("| name | age |", "| |)", r"\|"),
+    ],
+)
+async def test_redis_search_escapes_query_syntax_in_user_text(
+    message: str,
+    broken_shape: str,
+    escaped: str,
+    mock_index: AsyncMock,
+    patch_index: MagicMock,  # noqa: ARG001
+) -> None:
+    """User prose reaches RediSearch as escaped terms, not as query syntax (#10002).
+
+    A bare `|`, `?` or backtick in a message previously turned the generated query
+    string into malformed grammar (`@content:(ps | aux | | | grep | python)`), and
+    the syntax error failed the whole agent run.
+    """
+    provider = RedisContextProvider(source_id="ctx", user_id="user-1")
+
+    with patch.object(provider, "_ensure_index", AsyncMock()):
+        await provider._redis_search(text=message)
+
+    query = mock_index.query.await_args.args[0]
+    query_string = query.query_string()
+    assert broken_shape not in query_string
+    assert escaped in query_string
+
+
+async def test_redis_search_escapes_text_for_hybrid_query(
+    mock_index: AsyncMock,
+    patch_index: MagicMock,  # noqa: ARG001
+) -> None:
+    """The hybrid vector path gets the same escaped text."""
+    mock_vectorizer = MagicMock(spec=BaseVectorizer)
+    mock_vectorizer.dims = 128
+    mock_vectorizer.dtype = "float32"
+    mock_vectorizer.aembed = AsyncMock(return_value=[0.1] * 128)
+
+    provider = RedisContextProvider(
+        source_id="ctx",
+        user_id="u1",
+        redis_vectorizer=mock_vectorizer,
+        vector_field_name="embedding",
+    )
+
+    with (
+        patch.object(provider, "_ensure_index", AsyncMock()),
+        patch("agent_framework_redis._context_provider.AggregateHybridQuery") as hybrid_query,
+    ):
+        hybrid_query.return_value = MagicMock()
+        await provider._redis_search(text="ps aux | grep python")
+
+    assert hybrid_query.call_args.kwargs["text"] == r"ps aux \| grep python"
+    mock_index.query.assert_awaited_once()
+
+
 async def test_search_all_collects_paginated_batches(
     mock_index: AsyncMock,
     patch_index: MagicMock,  # noqa: ARG001

python/packages/redis/tests/test_providers.py

@@ -417,19 +417,31 @@ def test_borrowed_client_is_used_directly(self):
         "connection_kwargs",
         [
             {"redis_url": "redis://localhost:6379"},
-            {"credential_provider": MagicMock(), "host": "myhost"},
+            {"credential_provider": None, "host": "myhost"},
         ],
     )
-    def test_borrowed_client_rejects_other_connection_sources(self, connection_kwargs: dict[str, Any]):
+    def test_borrowed_client_takes_precedence_over_other_connection_sources(self, connection_kwargs: dict[str, Any]):
+        """An injected client wins; URL / credential settings are never validated (#9966).
+
+        The vector-store constructors already behave this way: a working
+        caller-owned client must not be rejected because a stray URL or
+        credential provider came along with it.
+        """
+        kwargs = dict(connection_kwargs)
+        if kwargs.get("credential_provider") is None:
+            kwargs["credential_provider"] = MagicMock()
+
         borrowed = Redis(decode_responses=True)
 
-        with pytest.raises(ValueError, match="mutually exclusive"):
-            RedisHistoryProvider(
-                "mem",
-                redis_client=borrowed,
-                application_id="test-app",
-                **connection_kwargs,
-            )
+        provider = RedisHistoryProvider(
+            "mem",
+            redis_client=borrowed,
+            application_id="test-app",
+            **kwargs,
+        )
+
+        assert provider._redis_client is borrowed
+        assert provider._owns_client is False
 
     def test_borrowed_client_must_be_async_redis(self):
         with pytest.raises(TypeError, match=r"redis\.asyncio\.Redis"):