Python: fix(redis): escape RediSearch query syntax in memory search text

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/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