fix(checkpoint): decode base64 blobs when loading pending writes

Add static _decode_blob_static() method to handle base64 decoding
of blob data retrieved from Redis before passing to serializer.
This fixes orjson.JSONDecodeError when loading pending writes.

Author: bsbodden

langgraph/checkpoint/redis/base.py

@@ -732,12 +732,25 @@ def _load_writes(
             (
                 task_id,
                 data["channel"],
-                serde.loads_typed((data["type"], data["blob"])),
+                serde.loads_typed((data["type"], BaseRedisSaver._decode_blob_static(data["blob"]))),
             )
             for (task_id, _), data in task_id_to_data.items()
         ]
         return writes
 
+    @staticmethod
+    def _decode_blob_static(blob: bytes | str) -> bytes:
+        """Decode blob data from Redis storage (static method)."""
+        try:
+            # If it's already bytes, try to decode as base64
+            if isinstance(blob, bytes):
+                return base64.b64decode(blob)
+            # If it's a string, encode to bytes first then decode
+            return base64.b64decode(blob.encode("utf-8"))
+        except (binascii.Error, TypeError, ValueError):
+            # Handle both malformed base64 data and incorrect input types
+            return blob.encode("utf-8") if isinstance(blob, str) else blob
+
     @staticmethod
     def _parse_redis_checkpoint_writes_key(redis_key: str) -> dict:
         # Ensure redis_key is a string