fix: JsonPlusRedisSerializer handling proper Serialization of Interrupt

langgraph/checkpoint/redis/jsonplus_redis.py

@@ -70,6 +70,10 @@ def _revive_if_needed(self, obj: Any) -> Any:
         '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.
+
         Args:
             obj: The object to potentially revive, which may be a dict, list, or primitive.
 
@@ -83,6 +87,25 @@ def _revive_if_needed(self, obj: Any) -> Any:
                 # This converts {'lc': 1, 'type': 'constructor', ...} back to
                 # 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
+            if (
+                "value" in obj
+                and "id" in obj
+                and len(obj) == 2
+                and isinstance(obj.get("id"), str)
+            ):
+                # 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
+
             # Recursively process nested dicts
             return {k: self._revive_if_needed(v) for k, v in obj.items()}
         elif isinstance(obj, list):

tests/test_async_store.py

@@ -6,13 +6,7 @@
 from uuid import uuid4
 
 import pytest
-from langgraph.store.base import (
-    GetOp,
-    Item,
-    ListNamespacesOp,
-    PutOp,
-    SearchOp,
-)
+from langgraph.store.base import GetOp, Item, ListNamespacesOp, PutOp, SearchOp
 
 from langgraph.store.redis import AsyncRedisStore
 from tests.embed_test_utils import CharacterEmbeddings

tests/test_crossslot_integration.py

@@ -1,9 +1,6 @@
 """Integration tests for CrossSlot error fix in checkpoint operations."""
 
-from langgraph.checkpoint.base import (
-    create_checkpoint,
-    empty_checkpoint,
-)
+from langgraph.checkpoint.base import create_checkpoint, empty_checkpoint
 
 from langgraph.checkpoint.redis import RedisSaver