Merge branch 'main' into fix/JsonPlusRedisSerializer

Author: keenborder786

langgraph/checkpoint/redis/aio.py

@@ -1724,7 +1724,7 @@ async def _abatch_load_pending_sends(
                 return_fields=[
                     "checkpoint_id",
                     "type",
-                    "blob",
+                    "$.blob",
                     "task_path",
                     "task_id",
                     "idx",
@@ -1745,20 +1745,27 @@ async def _abatch_load_pending_sends(
             # Sort and format results for each parent checkpoint
             for parent_checkpoint_id in parent_checkpoint_ids:
                 batch_key = (thread_id, checkpoint_ns, parent_checkpoint_id)
-                docs = writes_by_checkpoint.get(parent_checkpoint_id, [])
-
-                # Sort for deterministic order
-                sorted_docs = sorted(
-                    docs,
-                    key=lambda d: (
-                        getattr(d, "task_path", ""),
-                        getattr(d, "task_id", ""),
-                        getattr(d, "idx", 0),
+                writes = writes_by_checkpoint.get(parent_checkpoint_id, [])
+
+                # Sort results by task_path, task_id, idx
+                sorted_writes = sorted(
+                    writes,
+                    key=lambda x: (
+                        getattr(x, "task_path", ""),
+                        getattr(x, "task_id", ""),
+                        getattr(x, "idx", 0),
                     ),
                 )
 
-                # Convert to expected format
-                results_map[batch_key] = [(d.type, d.blob) for d in sorted_docs]
+                # Extract type and blob pairs
+                # Handle both direct attribute access and JSON path access
+                results_map[batch_key] = [
+                    (
+                        getattr(doc, "type", ""),
+                        getattr(doc, "$.blob", getattr(doc, "blob", b"")),
+                    )
+                    for doc in sorted_writes
+                ]
 
         return results_map
 

langgraph/checkpoint/redis/jsonplus_redis.py

@@ -62,23 +62,22 @@ def loads(self, data: bytes) -> Any:
                 return super().loads_typed(("json", data))
 
     def _revive_if_needed(self, obj: Any) -> Any:
-        """Recursively apply reviver to handle LangChain serialized objects.
+        """Recursively apply reviver to handle LangChain and LangGraph serialized objects.
 
         This method is crucial for preventing MESSAGE_COERCION_FAILURE by ensuring
         that LangChain message objects stored in their serialized format are properly
         reconstructed. Without this, messages would remain as dictionaries with
         'lc', 'type', and 'constructor' fields, causing errors when the application
         expects actual message objects with 'role' and 'content' attributes.
 
-        This also handles Interrupt objects that may be stored as plain dictionaries
-        with 'value' and 'id' keys, reconstructing them as proper Interrupt instances
-        to prevent AttributeError when accessing the 'id' attribute.
+        It also handles LangGraph Interrupt objects which serialize to {"value": ..., "resumable": ..., "ns": ..., "when": ...}
+        and must be reconstructed to prevent AttributeError when accessing Interrupt attributes.
 
         Args:
             obj: The object to potentially revive, which may be a dict, list, or primitive.
 
         Returns:
-            The revived object with LangChain objects properly reconstructed.
+            The revived object with LangChain/LangGraph objects properly reconstructed.
         """
         if isinstance(obj, dict):
             # Check if this is a LangChain serialized object
@@ -88,23 +87,31 @@ def _revive_if_needed(self, obj: Any) -> Any:
                 # the actual LangChain object (e.g., HumanMessage, AIMessage)
                 return self._reviver(obj)
 
-            # Check if this looks like an Interrupt object stored as a plain dict
-            # Interrupt objects have 'value' and 'id' keys, and possibly nothing else
-            # We need to be careful not to accidentally convert other dicts
+            # Check if this is a serialized Interrupt object
+            # Interrupt objects serialize to {"value": ..., "resumable": ..., "ns": ..., "when": ...}
+            # This must be done before recursively processing to avoid losing the structure
             if (
                 "value" in obj
-                and "id" in obj
-                and len(obj) == 2
-                and isinstance(obj.get("id"), str)
+                and "resumable" in obj
+                and "when" in obj
+                and len(obj) == 4
+                and isinstance(obj.get("resumable"), bool)
             ):
                 # Try to reconstruct as an Interrupt object
                 try:
                     from langgraph.types import Interrupt
 
-                    return Interrupt(value=obj["value"], id=obj["id"])  # type: ignore[call-arg]
-                except (ImportError, TypeError, ValueError):
-                    # If we can't import or construct Interrupt, fall through
-                    pass
+                    return Interrupt(
+                        value=self._revive_if_needed(obj["value"]),
+                        resumable=obj["resumable"],
+                        ns=obj["ns"],
+                        when=obj["when"],
+                    )
+                except (ImportError, TypeError, ValueError) as e:
+                    # If we can't import or construct Interrupt, log and fall through
+                    logger.debug(
+                        "Failed to deserialize Interrupt object: %s", e, exc_info=True
+                    )
 
             # Recursively process nested dicts
             return {k: self._revive_if_needed(v) for k, v in obj.items()}

langgraph/store/redis/aio.py

@@ -365,7 +365,7 @@ async def __aexit__(
     ) -> None:
         """Async context manager exit."""
         # Cancel the background task created by AsyncBatchedBaseStore
-        if hasattr(self, "_task") and not self._task.done():
+        if hasattr(self, "_task") and self._task is not None and not self._task.done():
             self._task.cancel()
             try:
                 await self._task