fix(store): address PR review feedback for message serialization

Improvements based on PR #129 review:

- Move json import to top of file for consistency
- Add exact key check (set equality) to prevent collisions with user data
  that might contain __serde_type__ and __serde_data__ keys
- Add comprehensive error handling for deserialization failures with logging
- Add hex validation for non-json type deserialization
- Fix async store to also apply value serialization in _prepare_batch_PUT_queries_async
- Make type handling explicit in serialization (json vs bytes/msgpack)

New tests added:
- Backward compatibility test for plain JSON values (filters continue to work)
- Serde key collision prevention test
- Bytes value serialization test
- Full async store test suite mirroring sync tests

Author: bsbodden

langgraph/store/redis/aio.py

@@ -578,7 +578,7 @@ async def _aprepare_batch_PUT_queries(
                 doc = RedisDocument(
                     prefix=_namespace_to_text(op.namespace),
                     key=op.key,
-                    value=op.value,
+                    value=self._serialize_value(op.value),
                     created_at=now,
                     updated_at=now,
                     ttl_minutes=ttl_minutes,

langgraph/store/redis/base.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import copy
+import json
 import logging
 import threading
 from collections import defaultdict
@@ -374,16 +375,17 @@ def _serialize_value(self, value: Any) -> Any:
         The method is smart about serialization:
         - If the value is a simple JSON-serializable dict/list, it's stored as-is
         - If the value contains complex objects (HumanMessage, etc.), it uses
-          the serde wrapper format
+          the serde wrapper format with __serde_type__ and __serde_data__ keys
+
+        Note: Values containing LangChain messages will be wrapped in a serde format,
+        which means filters on nested fields won't work for such values.
 
         Args:
             value: The value to serialize (can contain HumanMessage, AIMessage, etc.)
 
         Returns:
             A JSON-serializable representation of the value
         """
-        import json
-
         if value is None:
             return None
 
@@ -393,18 +395,23 @@ def _serialize_value(self, value: Any) -> Any:
             # Value is already JSON-serializable, return as-is for backward
             # compatibility and to preserve filter functionality
             return value
-        except (TypeError, ValueError):
+        except TypeError:
             # Value contains non-JSON-serializable objects, use serde wrapper
             pass
 
         # Use the serializer to handle complex objects
         type_str, data_bytes = self._serde.dumps_typed(value)
         # Store the serialized data with type info for proper deserialization
+        # Handle different type formats explicitly for clarity
+        if type_str == "json":
+            data_encoded = data_bytes.decode("utf-8")
+        else:
+            # bytes, bytearray, msgpack, and other types are hex-encoded
+            data_encoded = data_bytes.hex()
+
         return {
             "__serde_type__": type_str,
-            "__serde_data__": (
-                data_bytes.decode("utf-8") if type_str == "json" else data_bytes.hex()
-            ),
+            "__serde_data__": data_encoded,
         }
 
     def _deserialize_value(self, value: Any) -> Any:
@@ -423,21 +430,40 @@ def _deserialize_value(self, value: Any) -> Any:
             return None
 
         # Check if this is a serialized value (new format)
-        if (
-            isinstance(value, dict)
-            and "__serde_type__" in value
-            and "__serde_data__" in value
-        ):
+        # Use exact key check to prevent collisions with user data
+        if isinstance(value, dict) and set(value.keys()) == {
+            "__serde_type__",
+            "__serde_data__",
+        }:
             type_str = value["__serde_type__"]
             data_str = value["__serde_data__"]
 
-            # Convert back to bytes
-            if type_str == "json":
-                data_bytes = data_str.encode("utf-8")
-            else:
-                data_bytes = bytes.fromhex(data_str)
-
-            return self._serde.loads_typed((type_str, data_bytes))
+            try:
+                # Convert back to bytes based on type
+                if type_str == "json":
+                    data_bytes = data_str.encode("utf-8")
+                else:
+                    # bytes, bytearray, msgpack types are hex-encoded
+                    data_bytes = bytes.fromhex(data_str)
+
+                return self._serde.loads_typed((type_str, data_bytes))
+            except (ValueError, TypeError) as e:
+                # Handle hex decoding errors or deserialization failures
+                logger.error(
+                    "Failed to deserialize value from Redis: type=%r, error=%s",
+                    type_str,
+                    e,
+                )
+                # Return None to indicate deserialization failure
+                return None
+            except Exception as e:
+                # Handle any other unexpected errors during deserialization
+                logger.error(
+                    "Unexpected error deserializing value from Redis: type=%r, error=%s",
+                    type_str,
+                    e,
+                )
+                return None
 
         # Legacy format: value is stored as-is (plain JSON-serializable data)
         # Return as-is for backward compatibility

tests/test_issue_128_async_store_message_serialization.py

@@ -0,0 +1,201 @@
+"""Async tests for issue #128: HumanMessage serialization in AsyncRedisStore.
+
+This test suite mirrors the sync tests to ensure the async store implementation
+properly handles LangChain message serialization.
+"""
+
+from __future__ import annotations
+
+from typing import AsyncIterator
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
+
+from langgraph.store.redis.aio import AsyncRedisStore
+
+
+@pytest.fixture
+async def async_store(redis_url: str) -> AsyncIterator[AsyncRedisStore]:
+    """Fixture to create an async Redis store."""
+    async with AsyncRedisStore.from_conn_string(redis_url) as store:
+        await store.setup()
+        yield store
+
+
+class TestIssue128AsyncMessageSerialization:
+    """Async test suite for issue #128: HumanMessage serialization."""
+
+    @pytest.mark.asyncio
+    async def test_store_human_message_in_value(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test storing a HumanMessage object in async store."""
+        namespace = ("test", "async_messages")
+        key = "message1"
+
+        value = {
+            "messages": [
+                HumanMessage(content="Hello from async!"),
+            ]
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+        assert "messages" in item.value
+        assert len(item.value["messages"]) == 1
+
+        retrieved_message = item.value["messages"][0]
+        assert isinstance(retrieved_message, HumanMessage)
+        assert retrieved_message.content == "Hello from async!"
+
+    @pytest.mark.asyncio
+    async def test_store_multiple_message_types(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test storing multiple message types in async store."""
+        namespace = ("test", "async_conversation")
+        key = "conv1"
+
+        value = {
+            "messages": [
+                SystemMessage(content="You are a helpful assistant."),
+                HumanMessage(content="What is Python?"),
+                AIMessage(content="Python is a programming language."),
+            ]
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+        assert len(item.value["messages"]) == 3
+
+        messages = item.value["messages"]
+        assert isinstance(messages[0], SystemMessage)
+        assert isinstance(messages[1], HumanMessage)
+        assert isinstance(messages[2], AIMessage)
+
+    @pytest.mark.asyncio
+    async def test_store_ai_message_with_tool_calls(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test storing AIMessage with tool calls in async store."""
+        namespace = ("test", "async_tools")
+        key = "tool_call"
+
+        value = {
+            "messages": [
+                AIMessage(
+                    content="",
+                    tool_calls=[
+                        {
+                            "id": "call_123",
+                            "name": "get_weather",
+                            "args": {"location": "NYC"},
+                        }
+                    ],
+                ),
+            ]
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+
+        retrieved_message = item.value["messages"][0]
+        assert isinstance(retrieved_message, AIMessage)
+        assert len(retrieved_message.tool_calls) == 1
+        assert retrieved_message.tool_calls[0]["name"] == "get_weather"
+
+    @pytest.mark.asyncio
+    async def test_search_with_message_values(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test async search for items containing message values."""
+        namespace = ("test", "async_searchable")
+
+        for i in range(3):
+            await async_store.aput(
+                namespace,
+                f"msg{i}",
+                {
+                    "topic": f"topic_{i}",
+                    "messages": [HumanMessage(content=f"Async Message {i}")],
+                },
+            )
+
+        results = await async_store.asearch(namespace)
+        assert len(results) == 3
+
+        for result in results:
+            assert "messages" in result.value
+            assert isinstance(result.value["messages"][0], HumanMessage)
+
+    @pytest.mark.asyncio
+    async def test_backward_compatibility_plain_json(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test that plain JSON values work correctly in async store."""
+        namespace = ("test", "async_plain_json")
+        key = "simple_value"
+
+        value = {
+            "name": "async_test",
+            "count": 42,
+            "tags": ["a", "b", "c"],
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+        assert item.value == value
+
+        # Verify filters work on plain JSON values
+        results = await async_store.asearch(namespace, filter={"name": "async_test"})
+        assert len(results) == 1
+        assert results[0].key == key
+
+    @pytest.mark.asyncio
+    async def test_serde_key_collision_prevention(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test that user data with serde-like keys is handled correctly."""
+        namespace = ("test", "async_collision")
+        key = "user_data_with_serde_keys"
+
+        value = {
+            "__serde_type__": "user_defined_type",
+            "__serde_data__": "user_defined_data",
+            "extra_key": "this makes it not a serde wrapper",
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+        assert item.value == value
+        assert item.value["extra_key"] == "this makes it not a serde wrapper"
+
+    @pytest.mark.asyncio
+    async def test_bytes_value_serialization(
+        self, async_store: AsyncRedisStore
+    ) -> None:
+        """Test that bytes values are properly serialized in async store."""
+        namespace = ("test", "async_bytes")
+        key = "binary_data"
+
+        value = {
+            "data": b"async binary content",
+            "name": "test",
+        }
+
+        await async_store.aput(namespace, key, value)
+
+        item = await async_store.aget(namespace, key)
+        assert item is not None
+        assert item.value["data"] == b"async binary content"
+        assert item.value["name"] == "test"

tests/test_issue_128_store_message_serialization.py

@@ -209,3 +209,74 @@ def test_search_with_message_values(self, store: RedisStore) -> None:
         assert item.value["topic"] == "topic_1"
         assert isinstance(item.value["messages"][0], HumanMessage)
         assert item.value["messages"][0].content == "Message 1"
+
+    def test_backward_compatibility_plain_json(self, store: RedisStore) -> None:
+        """Test that plain JSON-serializable values are stored as-is.
+
+        This ensures backward compatibility: simple dict/list values should
+        be stored without the serde wrapper, preserving filter functionality.
+        """
+        namespace = ("test", "plain_json")
+        key = "simple_value"
+
+        # Plain JSON-serializable value (no LangChain objects)
+        value = {
+            "name": "test",
+            "count": 42,
+            "tags": ["a", "b", "c"],
+            "nested": {"foo": "bar"},
+        }
+
+        store.put(namespace, key, value)
+
+        # Verify we can retrieve it
+        item = store.get(namespace, key)
+        assert item is not None
+        assert item.value == value
+
+        # Verify filters work on plain JSON values
+        results = store.search(namespace, filter={"name": "test"})
+        assert len(results) == 1
+        assert results[0].key == key
+
+    def test_serde_key_collision_prevention(self, store: RedisStore) -> None:
+        """Test that user data with serde-like keys is not incorrectly deserialized.
+
+        If a user stores a dict with __serde_type__ and __serde_data__ keys
+        but also other keys, it should NOT be treated as serialized data.
+        """
+        namespace = ("test", "collision")
+        key = "user_data_with_serde_keys"
+
+        # User data that happens to have serde-like keys plus other keys
+        value = {
+            "__serde_type__": "user_defined_type",
+            "__serde_data__": "user_defined_data",
+            "extra_key": "this makes it not a serde wrapper",
+        }
+
+        store.put(namespace, key, value)
+
+        # Verify the value is retrieved as-is (not deserialized)
+        item = store.get(namespace, key)
+        assert item is not None
+        assert item.value == value
+        assert item.value["extra_key"] == "this makes it not a serde wrapper"
+
+    def test_bytes_value_serialization(self, store: RedisStore) -> None:
+        """Test that bytes values are properly serialized and deserialized."""
+        namespace = ("test", "bytes")
+        key = "binary_data"
+
+        # Value containing bytes (not JSON-serializable)
+        value = {
+            "data": b"binary content here",
+            "name": "test",
+        }
+
+        store.put(namespace, key, value)
+
+        item = store.get(namespace, key)
+        assert item is not None
+        assert item.value["data"] == b"binary content here"
+        assert item.value["name"] == "test"