fix(checkpoint): handle bytes in nested structures with msgpack fallback

- Add bytes detection in _default_handler to trigger msgpack serialization
- Update dumps_typed to fallback to parent's msgpack for bytes in dicts
- Update _dump_checkpoint to handle both JSON and msgpack types
- Add __bytes__ marker for bytes in channel_values stored via msgpack
- Update _recursive_deserialize to decode __bytes__ markers
- Fix test assertions to expect bytes instead of str from dumps_typed

Author: bsbodden

langgraph/checkpoint/redis/base.py

@@ -310,8 +310,21 @@ def _dump_checkpoint(self, checkpoint: Checkpoint) -> dict[str, Any]:
         """Convert checkpoint to Redis format."""
         type_, data = self.serde.dumps_typed(checkpoint)
 
-        # Since we're keeping JSON format, decode string data
-        checkpoint_data = cast(dict, orjson.loads(data))
+        # Decode the serialized data - handle both JSON and msgpack
+        if type_ == "json":
+            checkpoint_data = cast(dict, orjson.loads(data))
+        else:
+            # For msgpack or other types, deserialize with loads_typed
+            checkpoint_data = cast(dict, self.serde.loads_typed((type_, data)))
+
+            # When using msgpack, bytes are preserved - but Redis JSON.SET can't handle them
+            # Encode bytes in channel_values with type marker for JSON storage
+            if "channel_values" in checkpoint_data:
+                for key, value in checkpoint_data["channel_values"].items():
+                    if isinstance(value, bytes):
+                        checkpoint_data["channel_values"][key] = {
+                            "__bytes__": self._encode_blob(value)
+                        }
 
         # Ensure channel_versions are always strings to fix issue #40
         if "channel_versions" in checkpoint_data:
@@ -379,6 +392,11 @@ def _recursive_deserialize(self, obj: Any) -> Any:
             The deserialized object, with LangChain objects properly reconstructed.
         """
         if isinstance(obj, dict):
+            # Check if this is a bytes marker from msgpack storage
+            if "__bytes__" in obj and len(obj) == 1:
+                # Decode base64-encoded bytes
+                return self._decode_blob(obj["__bytes__"])
+
             # Check if this is a LangChain serialized object
             if obj.get("lc") in (1, 2) and obj.get("type") == "constructor":
                 try:

langgraph/checkpoint/redis/jsonplus_redis.py

@@ -33,6 +33,10 @@ def _default_handler(self, obj: Any) -> Any:
         This handles LangChain objects by delegating to the parent's
         _encode_constructor_args method which creates the LC format.
         """
+        # Bytes/bytearray in nested structures require msgpack - signal to fallback
+        if isinstance(obj, (bytes, bytearray)):
+            raise TypeError("bytes/bytearray in nested structure - use msgpack")
+
         # Try to encode using parent's constructor args encoder
         # This creates the {"lc": 2, "type": "constructor", ...} format
         try:
@@ -57,6 +61,8 @@ def _default_handler(self, obj: Any) -> Any:
     def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
         """Serialize using orjson for JSON.
 
+        Falls back to msgpack for structures containing bytes/bytearray.
+
         Returns:
             tuple[str, bytes]: Type identifier and serialized bytes
         """
@@ -67,9 +73,13 @@ def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
         elif obj is None:
             return "null", b""
         else:
-            # Use orjson for JSON serialization with custom default handler
-            json_bytes = orjson.dumps(obj, default=self._default_handler)
-            return "json", json_bytes
+            try:
+                # Try orjson first with custom default handler
+                json_bytes = orjson.dumps(obj, default=self._default_handler)
+                return "json", json_bytes
+            except (TypeError, orjson.JSONEncodeError):
+                # Fall back to parent's msgpack serialization for bytes in nested structures
+                return super().dumps_typed(obj)
 
     def loads_typed(self, data: tuple[str, bytes]) -> Any:
         """Deserialize with custom revival for LangChain/LangGraph objects.

tests/test_checkpoint_serialization.py

@@ -166,14 +166,13 @@ def test_issue_83_pending_sends_type_compatibility(redis_url: str) -> None:
         # Serialize
         type_str, blob = saver.serde.dumps_typed(test_data)
         assert isinstance(type_str, str)
-        assert isinstance(blob, str)  # JsonPlusRedisSerializer returns strings
+        # Checkpoint 3.0: dumps_typed now returns bytes
+        assert isinstance(blob, bytes)
 
-        # Deserialize - should work with both string and bytes
+        # Deserialize with bytes (checkpoint 3.0 format)
         result1 = saver.serde.loads_typed((type_str, blob))
-        result2 = saver.serde.loads_typed((type_str, blob.encode()))  # bytes version
 
         assert result1 == test_data
-        assert result2 == test_data
 
 
 def test_load_blobs_method(redis_url: str) -> None:
@@ -508,7 +507,8 @@ def test_langchain_message_serialization(redis_url: str) -> None:
     # Serialize
     type_, data = serializer.dumps_typed(human_msg)
     assert type_ == "json"
-    assert isinstance(data, str)
+    # Checkpoint 3.0: dumps_typed now returns bytes
+    assert isinstance(data, bytes)
 
     # Deserialize
     deserialized = serializer.loads_typed((type_, data))