Update pyproject.toml

- Previously the global resume value was passed to subgraphs without
being consumed
- This would result in two parallel subgraph calls being able to use the
same resume value
- Note this behavior can't be implemented over the wire, that will be
fixed in future PR

Closes #3398

.github/workflows/size.yml

@@ -1,29 +0,0 @@
-name: Check File Size
-
-on:
-  push:
-    branches:
-      - main
-  pull_request:
-    branches:
-      - main
-  workflow_dispatch:
-
-jobs:
-  file-size-check:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - name: Get changed files
-        id: changed-files
-        uses: tj-actions/changed-files@v44
-      - name: Filter by size
-        # TODO: roll back the web voyager hack
-        run: |
-          large_added_files=$(find ${{ steps.changed-files.outputs.added_files }} -maxdepth 0 -size +1M | grep -v "web_voyager" || true)
-          if [ -n "$large_added_files" ]; then
-            echo "Large files added: $large_added_files"
-            echo "# Large files added:" >> $GITHUB_STEP_SUMMARY
-            echo "$large_added_files" >> $GITHUB_STEP_SUMMARY
-            exit 1
-          fi

docs/docs/concepts/persistence.md

@@ -232,7 +232,7 @@ from langgraph.store.memory import InMemoryStore
 in_memory_store = InMemoryStore()
 ```
 
-Memories are namespaced by a `tuple`, which in this specific example will be `(<user_id>, "memories")`. The namespace can be any length and represent anything, does not have be user specific.
+Memories are namespaced by a `tuple`, which in this specific example will be `(<user_id>, "memories")`. The namespace can be any length and represent anything, does not have to be user specific.
 
 ```python 
 user_id = "1"
@@ -387,6 +387,9 @@ We can access the memories and use them in our model call.
 def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
     # Get the user id from the config
     user_id = config["configurable"]["user_id"]
+
+    # Namespace the memory
+    namespace = (user_id, "memories")
     
     # Search based on the most recent message
     memories = store.search(

libs/checkpoint-postgres/langgraph/store/postgres/base.py

@@ -79,12 +79,12 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS store_prefix_idx ON store USING btree (p
     """
 -- Add expires_at column to store table
 ALTER TABLE store
-ADD COLUMN expires_at TIMESTAMP WITH TIME ZONE,
-ADD COLUMN ttl_minutes INT;
+ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP WITH TIME ZONE,
+ADD COLUMN IF NOT EXISTS ttl_minutes INT;
 """,
     """
 -- Add indexes for efficient TTL sweeping
-CREATE INDEX idx_store_expires_at ON store (expires_at)
+CREATE INDEX IF NOT EXISTS idx_store_expires_at ON store (expires_at)
 WHERE expires_at IS NOT NULL;
 """,
 ]

libs/checkpoint-postgres/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-checkpoint-postgres"
-version = "2.0.17"
+version = "2.0.18"
 description = "Library with a Postgres implementation of LangGraph checkpoint saver."
 authors = []
 license = "MIT"

libs/checkpoint-postgres/tests/test_async_store.py

@@ -60,15 +60,17 @@ async def store(request) -> AsyncIterator[AsyncPostgresStore]:
         ) as store:
             store.MIGRATIONS = [
                 (
-                    mig.replace(
-                        "ADD COLUMN ttl_minutes INT;", "ADD COLUMN ttl_minutes FLOAT;"
-                    )
+                    mig.replace("ttl_minutes INT;", "ttl_minutes FLOAT;")
                     if isinstance(mig, str)
                     else mig
                 )
                 for mig in store.MIGRATIONS
             ]
             await store.setup()
+            async with store._cursor() as cur:
+                # drop the migration index
+                await cur.execute("DROP TABLE IF EXISTS store_migrations")
+            await store.setup()  # Will fail if migrations aren't idempotent
 
         if request.param == "pipe":
             async with AsyncPostgresStore.from_conn_string(

libs/checkpoint-postgres/tests/test_store.py

@@ -52,9 +52,7 @@ def store(request) -> PostgresStore:
         with PostgresStore.from_conn_string(conn_string, ttl=ttl_config) as store:
             store.MIGRATIONS = [
                 (
-                    mig.replace(
-                        "ADD COLUMN ttl_minutes INT;", "ADD COLUMN ttl_minutes FLOAT;"
-                    )
+                    mig.replace("ttl_minutes INT;", "ttl_minutes FLOAT;")
                     if isinstance(mig, str)
                     else mig
                 )
@@ -415,6 +413,10 @@ def _create_vector_store(
             ttl={"default_ttl": 2, "refresh_on_read": True} if enable_ttl else None,
         ) as store:
             store.setup()
+            with store._cursor() as cur:
+                # drop the migration index
+                cur.execute("DROP TABLE IF EXISTS store_migrations")
+            store.setup()  # Will fail if migrations aren't idempotent
             yield store
     finally:
         with Connection.connect(admin_conn_string, autocommit=True) as conn:

libs/checkpoint/langgraph/checkpoint/serde/base.py

@@ -45,3 +45,18 @@ def maybe_add_typed_methods(serde: SerializerProtocol) -> SerializerProtocol:
         return SerializerCompat(serde)
 
     return serde
+
+
+class CipherProtocol(Protocol):
+    """Protocol for encryption and decryption of data.
+    - `encrypt`: Encrypt plaintext.
+    - `decrypt`: Decrypt ciphertext.
+    """
+
+    def encrypt(self, plaintext: bytes) -> tuple[str, bytes]:
+        """Encrypt plaintext. Returns a tuple (cipher name, ciphertext)."""
+        ...
+
+    def decrypt(self, ciphername: str, ciphertext: bytes) -> bytes:
+        """Decrypt ciphertext. Returns the plaintext."""
+        ...

libs/checkpoint/langgraph/checkpoint/serde/encrypted.py

@@ -0,0 +1,86 @@
+import os
+from typing import Any
+
+from langgraph.checkpoint.serde.base import CipherProtocol, SerializerProtocol
+from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
+
+
+class EncryptedSerializer(SerializerProtocol):
+    """Serializer that encrypts and decrypts data using an encryption protocol."""
+
+    def __init__(
+        self, cipher: CipherProtocol, serde: SerializerProtocol = JsonPlusSerializer()
+    ) -> None:
+        self.cipher = cipher
+        self.serde = serde
+
+    def dumps(self, obj: Any) -> bytes:
+        return self.serde.dumps(obj)
+
+    def loads(self, data: bytes) -> Any:
+        return self.serde.loads(data)
+
+    def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
+        """Serialize an object to a tuple (type, bytes) and encrypt the bytes."""
+        # serialize data
+        typ, data = self.serde.dumps_typed(obj)
+        # encrypt data
+        ciphername, ciphertext = self.cipher.encrypt(data)
+        # add cipher name to type
+        return f"{typ}+{ciphername}", ciphertext
+
+    def loads_typed(self, data: tuple[str, bytes]) -> Any:
+        enc_cipher, ciphertext = data
+        # unencrypted data
+        if "+" not in enc_cipher:
+            return self.serde.loads_typed(data)
+        # extract cipher name
+        typ, ciphername = enc_cipher.split("+", 1)
+        # decrypt data
+        decrypted_data = self.cipher.decrypt(ciphername, ciphertext)
+        # deserialize data
+        return self.serde.loads_typed((typ, decrypted_data))
+
+    @classmethod
+    def from_pycryptodome_aes(
+        cls, serde: SerializerProtocol = JsonPlusSerializer(), **kwargs: Any
+    ) -> "EncryptedSerializer":
+        """Create an EncryptedSerializer using AES encryption."""
+        try:
+            from Crypto.Cipher import AES  # type: ignore
+        except ImportError:
+            raise ImportError(
+                "Pycryptodome is not installed. Please install it with `pip install pycryptodome`."
+            ) from None
+
+        # check if AES key is provided
+        if "key" in kwargs:
+            key: bytes = kwargs.pop("key")
+        else:
+            key_str = os.getenv("LANGGRAPH_AES_KEY")
+            if key_str is None:
+                raise ValueError("LANGGRAPH_AES_KEY environment variable is not set.")
+            key = key_str.encode()
+            if len(key) not in (16, 24, 32):
+                raise ValueError("LANGGRAPH_AES_KEY must be 16, 24, or 32 bytes long.")
+
+        # set default mode to EAX if not provided
+        if kwargs.get("mode") is None:
+            kwargs["mode"] = AES.MODE_EAX
+
+        class PycryptodomeAesCipher(CipherProtocol):
+            def encrypt(self, plaintext: bytes) -> tuple[str, bytes]:
+                cipher = AES.new(key, **kwargs)
+                ciphertext, tag = cipher.encrypt_and_digest(plaintext)
+                return "aes", cipher.nonce + tag + ciphertext
+
+            def decrypt(self, ciphername: str, ciphertext: bytes) -> bytes:
+                assert ciphername == "aes", f"Unsupported cipher: {ciphername}"
+                nonce = ciphertext[:16]
+                tag = ciphertext[16:32]
+                actual_ciphertext = ciphertext[32:]
+
+                cipher = AES.new(key, **kwargs, nonce=nonce)
+                return cipher.decrypt_and_verify(actual_ciphertext, tag)
+
+        return cls(PycryptodomeAesCipher(), serde)

libs/checkpoint/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-checkpoint"
-version = "2.0.20"
+version = "2.0.21"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 license = "MIT"

libs/langgraph/langgraph/graph/graph.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from collections import defaultdict
 from typing import (
@@ -31,6 +32,7 @@ from langgraph.constants import (
 )
 from langgraph.graph.branch import Branch
 from langgraph.pregel import Channel, Pregel
+from langgraph.pregel.protocol import PregelProtocol
 from langgraph.pregel.read import PregelNode
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
 from langgraph.types import All, Checkpointer
@@ -418,7 +420,38 @@ class CompiledGraph(Pregel):
         *,
         xray: Union[int, bool] = False,
     ) -> DrawableGraph:
-        return self.get_graph(config, xray=xray)
+        """Returns a drawable representation of the computation graph."""
+        from langgraph.pregel.remote import RemoteGraph
+
+        # gather subgraphs
+        if xray:
+            subpregels: dict[str, PregelProtocol] = {
+                k: v
+                async for k, v in self.aget_subgraphs()
+                if isinstance(v, (CompiledGraph, RemoteGraph))
+            }
+            subgraphs = {
+                k: v
+                for k, v in zip(
+                    subpregels,
+                    await asyncio.gather(
+                        *(
+                            p.aget_graph(
+                                config,
+                                xray=xray
+                                if isinstance(xray, bool) or xray <= 0
+                                else xray - 1,
+                            )
+                            for p in subpregels.values()
+                        )
+                    ),
+                )
+            }
+        else:
+            subgraphs = {}
+
+        # draw the graph
+        return self._draw_graph(config, subgraphs=subgraphs)
 
     def get_graph(
         self,
@@ -427,17 +460,36 @@ class CompiledGraph(Pregel):
         xray: Union[int, bool] = False,
     ) -> DrawableGraph:
         """Returns a drawable representation of the computation graph."""
+        from langgraph.pregel.remote import RemoteGraph
+
+        # gather subgraphs
+        if xray:
+            subgraphs = {
+                k: v.get_graph(
+                    config,
+                    xray=xray if isinstance(xray, bool) or xray <= 0 else xray - 1,
+                )
+                for k, v in self.get_subgraphs()
+                if isinstance(v, (CompiledGraph, RemoteGraph))
+            }
+        else:
+            subgraphs = {}
+
+        # draw the graph
+        return self._draw_graph(config, subgraphs=subgraphs)
+
+    def _draw_graph(
+        self,
+        config: Optional[RunnableConfig] = None,
+        *,
+        subgraphs: dict[str, DrawableGraph] = {},
+    ) -> DrawableGraph:
+        # create the graph
         graph = DrawableGraph()
         start_nodes: dict[str, DrawableNode] = {
             START: graph.add_node(self.get_input_schema(config), START)
         }
         end_nodes: dict[str, DrawableNode] = {}
-        if xray:
-            subgraphs = {
-                k: v for k, v in self.get_subgraphs() if isinstance(v, CompiledGraph)
-            }
-        else:
-            subgraphs = {}
 
         def add_edge(
             start: str,
@@ -463,13 +515,8 @@ class CompiledGraph(Pregel):
                 metadata["__interrupt"] = "before"
             elif key in self.interrupt_after_nodes:
                 metadata["__interrupt"] = "after"
-            if xray and key in subgraphs:
-                subgraph = subgraphs[key].get_graph(
-                    config=config,
-                    xray=xray - 1
-                    if isinstance(xray, int) and not isinstance(xray, bool) and xray > 0
-                    else xray,
-                )
+            if key in subgraphs:
+                subgraph = subgraphs[key]
                 subgraph.trim_first_node()
                 subgraph.trim_last_node()
                 if len(subgraph.nodes) >= 1:

libs/langgraph/langgraph/graph/state.py

@@ -416,7 +416,7 @@ class StateGraph(Graph):
                         and (vals := get_args(rargs[0]))
                     ):
                         ends = vals
-        except (TypeError, StopIteration):
+        except (NameError, TypeError, StopIteration):
             pass
 
         if destinations is not None:

libs/langgraph/langgraph/pregel/algo.py

@@ -1,4 +1,3 @@
-import functools
 import itertools
 import sys
 from collections import defaultdict, deque
@@ -507,6 +506,7 @@ def prepare_single_task(
                         CONFIG_KEY_CHECKPOINT_ID: None,
                         CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,
                         CONFIG_KEY_SCRATCHPAD: _scratchpad(
+                            config,
                             pending_writes,
                             task_id,
                         ),
@@ -616,6 +616,7 @@ def prepare_single_task(
                             CONFIG_KEY_CHECKPOINT_ID: None,
                             CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,
                             CONFIG_KEY_SCRATCHPAD: _scratchpad(
+                                config,
                                 pending_writes,
                                 task_id,
                             ),
@@ -741,6 +742,7 @@ def prepare_single_task(
                                 CONFIG_KEY_CHECKPOINT_ID: None,
                                 CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,
                                 CONFIG_KEY_SCRATCHPAD: _scratchpad(
+                                    config,
                                     pending_writes,
                                     task_id,
                                 ),
@@ -762,12 +764,32 @@ def prepare_single_task(
 
 
 def _scratchpad(
+    config: RunnableConfig,
     pending_writes: list[PendingWrite],
     task_id: str,
 ) -> PregelScratchpad:
+    # None cannot be used as a resume value, because it would be difficult to
+    # distinguish from missing when used over http
     null_resume_write = next(
         (w for w in pending_writes if w[0] == NULL_TASK_ID and w[1] == RESUME), None
     )
+    parent_scratchpad: Optional[PregelScratchpad] = config[CONF].get(
+        CONFIG_KEY_SCRATCHPAD
+    )
+
+    def get_null_resume(consume: bool = False) -> Any:
+        if null_resume_write is None:
+            if parent_scratchpad is not None:
+                return parent_scratchpad.get_null_resume(consume)
+            return None
+        if consume:
+            try:
+                pending_writes.remove(null_resume_write)
+                return null_resume_write[2]
+            except ValueError:
+                return None
+        return null_resume_write[2]
+
     # using itertools.count as an atomic counter (+= 1 is not thread-safe)
     return PregelScratchpad(
         # call
@@ -777,10 +799,7 @@ def _scratchpad(
         resume=next(
             (w[2] for w in pending_writes if w[0] == task_id and w[1] == RESUME), []
         ),
-        null_resume=null_resume_write[2] if null_resume_write is not None else None,
-        _consume_null_resume=functools.partial(pending_writes.remove, null_resume_write)
-        if null_resume_write is not None
-        else lambda: None,
+        get_null_resume=get_null_resume,
         # subgraph
         subgraph_counter=itertools.count(0).__next__,
     )

libs/langgraph/langgraph/pregel/debug.py

@@ -137,7 +137,12 @@ def map_debug_task_results(
             "result": [
                 w for w in writes if w[0] in stream_channels_list or w[0] == RETURN
             ],
-            "interrupts": [asdict(w[1]) for w in writes if w[0] == INTERRUPT],
+            "interrupts": [
+                asdict(v)
+                for w in writes
+                if w[0] == INTERRUPT
+                for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])
+            ],
         },
     }
 
@@ -293,8 +298,9 @@ def tasks_w_writes(
                 ),
                 tuple(
                     v
-                    for tid, n, v in pending_writes
+                    for tid, n, vv in pending_writes
                     if tid == task.id and n == INTERRUPT
+                    for v in (vv if isinstance(vv, Sequence) else [vv])
                 ),
                 states.get(task.id) if states else None,
                 (

libs/langgraph/langgraph/pregel/io.py

@@ -28,7 +28,7 @@ def is_task_id(task_id: str) -> bool:
     """Check if a string is a valid task id."""
     try:
         UUID(task_id)
-    except ValueError:
+    except Exception:
         return False
     return True
 

libs/langgraph/langgraph/pregel/loop.py

@@ -587,15 +587,6 @@ class PregelLoop(LoopProtocol):
             )
         )
 
-        # take resume value from parent
-        if scratchpad := cast(
-            Optional[PregelScratchpad], configurable.get(CONFIG_KEY_SCRATCHPAD)
-        ):
-            if (
-                isinstance(scratchpad, PregelScratchpad)
-                and scratchpad.null_resume is not None
-            ):
-                self.put_writes(NULL_TASK_ID, [(RESUME, scratchpad.null_resume)])
         # map command to writes
         if isinstance(self.input, Command):
             if self.input.resume is not None and not self.checkpointer:
@@ -794,11 +785,14 @@ class PregelLoop(LoopProtocol):
                     [w for t in self.tasks.values() for w in t.writes],
                     self.channels,
                 )
-            # emit INTERRUPT event
-            self._emit(
-                "updates",
-                lambda: iter([{INTERRUPT: cast(GraphInterrupt, exc_value).args[0]}]),
-            )
+            # emit INTERRUPT if exception is empty (otherwise emitted by put_writes)
+            if exc_value is not None and (not exc_value.args or not exc_value.args[0]):
+                self._emit(
+                    "updates",
+                    lambda: iter(
+                        [{INTERRUPT: cast(GraphInterrupt, exc_value).args[0]}]
+                    ),
+                )
             # save final output
             self.output = read_channels(self.channels, self.output_keys)
             # suppress interrupt
@@ -829,7 +823,25 @@ class PregelLoop(LoopProtocol):
                 "tags", EMPTY_SEQ
             ):
                 return
-            if writes[0][0] != ERROR and writes[0][0] != INTERRUPT:
+            if writes[0][0] == INTERRUPT:
+                self._emit(
+                    "updates",
+                    lambda: iter(
+                        [
+                            {
+                                INTERRUPT: tuple(
+                                    v
+                                    for w in writes
+                                    if w[0] == INTERRUPT
+                                    for v in (
+                                        w[1] if isinstance(w[1], Sequence) else (w[1],)
+                                    )
+                                )
+                            }
+                        ]
+                    ),
+                )
+            elif writes[0][0] != ERROR:
                 self._emit(
                     "updates",
                     map_output_updates,

libs/langgraph/langgraph/pregel/runner.py

@@ -543,10 +543,11 @@ class PregelRunner:
         elif exception:
             if isinstance(exception, GraphInterrupt):
                 # save interrupt to checkpointer
-                if interrupts := [(INTERRUPT, i) for i in exception.args[0]]:
+                if exception.args[0]:
+                    writes = [(INTERRUPT, exception.args[0])]
                     if resumes := [w for w in task.writes if w[0] == RESUME]:
-                        interrupts.extend(resumes)
-                    self.put_writes(task.id, interrupts)
+                        writes.extend(resumes)
+                    self.put_writes(task.id, writes)
             elif isinstance(exception, GraphBubbleUp):
                 raise exception
             else:
@@ -608,6 +609,7 @@ def _panic_or_proceed(
             done.add(fut)
         else:
             inflight.add(fut)
+    interrupts: list[GraphInterrupt] = []
     while done:
         # if any task failed
         if exc := _exception(done.pop()):
@@ -616,7 +618,14 @@ def _panic_or_proceed(
                 inflight.pop().cancel()
             # raise the exception
             if panic:
-                raise exc
+                if isinstance(exc, GraphInterrupt):
+                    # collect interrupts
+                    interrupts.append(exc)
+                else:
+                    raise exc
+    # raise combined interrupts
+    if interrupts:
+        raise GraphInterrupt(tuple(i for exc in interrupts for i in exc.args[0]))
     if inflight:
         # if we got here means we timed out
         while inflight:

libs/langgraph/langgraph/types.py

@@ -130,7 +130,7 @@ class Interrupt:
     value: Any
     resumable: bool = False
     ns: Optional[Sequence[str]] = None
-    when: Literal["during"] = "during"
+    when: Literal["during"] = dataclasses.field(default="during", repr=False)
 
 
 class PregelTask(NamedTuple):
@@ -351,20 +351,11 @@ class PregelScratchpad:
     call_counter: Callable[[], int]
     # interrupt
     interrupt_counter: Callable[[], int]
+    get_null_resume: Callable[[bool], Any]
     resume: list[Any]
-    null_resume: Optional[Any]
-    _consume_null_resume: Callable[[], None]
     # subgraph
     subgraph_counter: Callable[[], int]
 
-    def consume_null_resume(self) -> Any:
-        if self.null_resume is not None:
-            value = self.null_resume
-            self._consume_null_resume()
-            self.null_resume = None
-            return value
-        raise ValueError("No null resume to consume")
-
 
 def interrupt(value: Any) -> Any:
     """Interrupt the graph with a resumable exception from within a node.
@@ -480,9 +471,9 @@ def interrupt(value: Any) -> Any:
         if idx < len(scratchpad.resume):
             return scratchpad.resume[idx]
     # find current resume value
-    if scratchpad.null_resume is not None:
+    v = scratchpad.get_null_resume(True)
+    if v is not None:
         assert len(scratchpad.resume) == idx, (scratchpad.resume, idx)
-        v = scratchpad.consume_null_resume()
         scratchpad.resume.append(v)
         conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
         return v

libs/langgraph/poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.0.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.0.0 and should not be changed by hand.
 
 [[package]]
 name = "aiosqlite"
@@ -2238,6 +2238,48 @@ files = [
     {file = "pycparser-2.22.tar.gz", hash = "sha256:491c8be9c040f5390f5bf44a5b07752bd07f56edf992381b05c701439eec10f6"},
 ]
 
+[[package]]
+name = "pycryptodome"
+version = "3.21.0"
+description = "Cryptographic library for Python"
+optional = false
+python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,>=2.7"
+groups = ["dev"]
+files = [
+    {file = "pycryptodome-3.21.0-cp27-cp27m-macosx_10_9_x86_64.whl", hash = "sha256:dad9bf36eda068e89059d1f07408e397856be9511d7113ea4b586642a429a4fd"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-manylinux2010_i686.whl", hash = "sha256:a1752eca64c60852f38bb29e2c86fca30d7672c024128ef5d70cc15868fa10f4"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-manylinux2010_x86_64.whl", hash = "sha256:3ba4cc304eac4d4d458f508d4955a88ba25026890e8abff9b60404f76a62c55e"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7cb087b8612c8a1a14cf37dd754685be9a8d9869bed2ffaaceb04850a8aeef7e"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-musllinux_1_1_aarch64.whl", hash = "sha256:26412b21df30b2861424a6c6d5b1d8ca8107612a4cfa4d0183e71c5d200fb34a"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-win32.whl", hash = "sha256:cc2269ab4bce40b027b49663d61d816903a4bd90ad88cb99ed561aadb3888dd3"},
+    {file = "pycryptodome-3.21.0-cp27-cp27m-win_amd64.whl", hash = "sha256:0fa0a05a6a697ccbf2a12cec3d6d2650b50881899b845fac6e87416f8cb7e87d"},
+    {file = "pycryptodome-3.21.0-cp27-cp27mu-manylinux2010_i686.whl", hash = "sha256:6cce52e196a5f1d6797ff7946cdff2038d3b5f0aba4a43cb6bf46b575fd1b5bb"},
+    {file = "pycryptodome-3.21.0-cp27-cp27mu-manylinux2010_x86_64.whl", hash = "sha256:a915597ffccabe902e7090e199a7bf7a381c5506a747d5e9d27ba55197a2c568"},
+    {file = "pycryptodome-3.21.0-cp27-cp27mu-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a4e74c522d630766b03a836c15bff77cb657c5fdf098abf8b1ada2aebc7d0819"},
+    {file = "pycryptodome-3.21.0-cp27-cp27mu-musllinux_1_1_aarch64.whl", hash = "sha256:a3804675283f4764a02db05f5191eb8fec2bb6ca34d466167fc78a5f05bbe6b3"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-macosx_10_9_universal2.whl", hash = "sha256:2480ec2c72438430da9f601ebc12c518c093c13111a5c1644c82cdfc2e50b1e4"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:de18954104667f565e2fbb4783b56667f30fb49c4d79b346f52a29cb198d5b6b"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:2de4b7263a33947ff440412339cb72b28a5a4c769b5c1ca19e33dd6cd1dcec6e"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0714206d467fc911042d01ea3a1847c847bc10884cf674c82e12915cfe1649f8"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:7d85c1b613121ed3dbaa5a97369b3b757909531a959d229406a75b912dd51dd1"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-musllinux_1_1_aarch64.whl", hash = "sha256:8898a66425a57bcf15e25fc19c12490b87bd939800f39a03ea2de2aea5e3611a"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-musllinux_1_2_i686.whl", hash = "sha256:932c905b71a56474bff8a9c014030bc3c882cee696b448af920399f730a650c2"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:18caa8cfbc676eaaf28613637a89980ad2fd96e00c564135bf90bc3f0b34dd93"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-win32.whl", hash = "sha256:280b67d20e33bb63171d55b1067f61fbd932e0b1ad976b3a184303a3dad22764"},
+    {file = "pycryptodome-3.21.0-cp36-abi3-win_amd64.whl", hash = "sha256:b7aa25fc0baa5b1d95b7633af4f5f1838467f1815442b22487426f94e0d66c53"},
+    {file = "pycryptodome-3.21.0-pp27-pypy_73-manylinux2010_x86_64.whl", hash = "sha256:2cb635b67011bc147c257e61ce864879ffe6d03342dc74b6045059dfbdedafca"},
+    {file = "pycryptodome-3.21.0-pp27-pypy_73-win32.whl", hash = "sha256:4c26a2f0dc15f81ea3afa3b0c87b87e501f235d332b7f27e2225ecb80c0b1cdd"},
+    {file = "pycryptodome-3.21.0-pp310-pypy310_pp73-macosx_10_15_x86_64.whl", hash = "sha256:d5ebe0763c982f069d3877832254f64974139f4f9655058452603ff559c482e8"},
+    {file = "pycryptodome-3.21.0-pp310-pypy310_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7ee86cbde706be13f2dec5a42b52b1c1d1cbb90c8e405c68d0755134735c8dc6"},
+    {file = "pycryptodome-3.21.0-pp310-pypy310_pp73-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0fd54003ec3ce4e0f16c484a10bc5d8b9bd77fa662a12b85779a2d2d85d67ee0"},
+    {file = "pycryptodome-3.21.0-pp310-pypy310_pp73-win_amd64.whl", hash = "sha256:5dfafca172933506773482b0e18f0cd766fd3920bd03ec85a283df90d8a17bc6"},
+    {file = "pycryptodome-3.21.0-pp39-pypy39_pp73-macosx_10_15_x86_64.whl", hash = "sha256:590ef0898a4b0a15485b05210b4a1c9de8806d3ad3d47f74ab1dc07c67a6827f"},
+    {file = "pycryptodome-3.21.0-pp39-pypy39_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f35e442630bc4bc2e1878482d6f59ea22e280d7121d7adeaedba58c23ab6386b"},
+    {file = "pycryptodome-3.21.0-pp39-pypy39_pp73-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:ff99f952db3db2fbe98a0b355175f93ec334ba3d01bbde25ad3a5a33abc02b58"},
+    {file = "pycryptodome-3.21.0-pp39-pypy39_pp73-win_amd64.whl", hash = "sha256:8acd7d34af70ee63f9a849f957558e49a98f8f1634f86a59d2be62bb8e93f71c"},
+    {file = "pycryptodome-3.21.0.tar.gz", hash = "sha256:f7787e0d469bdae763b876174cf2e6c0f7be79808af26b1da96f1a64bcf47297"},
+]
+
 [[package]]
 name = "pydantic"
 version = "2.9.2"
@@ -3509,4 +3551,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.9.0,<4.0"
-content-hash = "eb85f0bcc0e8a715ef38afb58cf888f7c2ee8579ea6ed94900244365f24cddd9"
+content-hash = "b8641a0b2d92bee0363602e69f99b23366b2035b7e17ff017708194e6fbd0ac5"

libs/langgraph/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.3.10"
+version = "0.3.14"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"
@@ -37,6 +37,7 @@ uvloop = "0.21.0beta1"
 pyperf = "^2.7.0"
 py-spy = "^0.3.14"
 types-requests = "^2.32.0.20240914"
+pycryptodome = "^3.21.0"
 
 [tool.ruff]
 lint.select = [ "E", "F", "I", "TID251" ]

libs/langgraph/tests/__snapshots__/test_pregel.ambr

@@ -377,6 +377,19 @@
   
   '''
 # ---
+# name: test_in_one_fan_out_state_graph_waiting_edge[sqlite_aes]
+  '''
+  graph TD;
+  	__start__ --> rewrite_query;
+  	analyzer_one --> retriever_one;
+  	qa --> __end__;
+  	retriever_one --> qa;
+  	retriever_two --> qa;
+  	rewrite_query --> analyzer_one;
+  	rewrite_query --> retriever_two;
+  
+  '''
+# ---
 # name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1[memory]
   '''
   graph TD;
@@ -797,6 +810,76 @@
     'type': 'object',
   })
 # ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1[sqlite_aes]
+  '''
+  graph TD;
+  	__start__ --> rewrite_query;
+  	analyzer_one --> retriever_one;
+  	qa --> __end__;
+  	retriever_one --> qa;
+  	retriever_two --> qa;
+  	rewrite_query --> analyzer_one;
+  	rewrite_query -.-> retriever_two;
+  
+  '''
+# ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1[sqlite_aes].1
+  dict({
+    'definitions': dict({
+      'InnerObject': dict({
+        'properties': dict({
+          'yo': dict({
+            'title': 'Yo',
+            'type': 'integer',
+          }),
+        }),
+        'required': list([
+          'yo',
+        ]),
+        'title': 'InnerObject',
+        'type': 'object',
+      }),
+    }),
+    'properties': dict({
+      'inner': dict({
+        '$ref': '#/definitions/InnerObject',
+      }),
+      'query': dict({
+        'title': 'Query',
+        'type': 'string',
+      }),
+    }),
+    'required': list([
+      'query',
+      'inner',
+    ]),
+    'title': 'Input',
+    'type': 'object',
+  })
+# ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1[sqlite_aes].2
+  dict({
+    'properties': dict({
+      'answer': dict({
+        'title': 'Answer',
+        'type': 'string',
+      }),
+      'docs': dict({
+        'items': dict({
+          'type': 'string',
+        }),
+        'title': 'Docs',
+        'type': 'array',
+      }),
+    }),
+    'required': list([
+      'answer',
+      'docs',
+    ]),
+    'title': 'Output',
+    'type': 'object',
+  })
+# ---
 # name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic2[memory]
   '''
   graph TD;
@@ -1217,6 +1300,76 @@
     'type': 'object',
   })
 # ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic2[sqlite_aes]
+  '''
+  graph TD;
+  	__start__ --> rewrite_query;
+  	analyzer_one --> retriever_one;
+  	qa --> __end__;
+  	retriever_one --> qa;
+  	retriever_two --> qa;
+  	rewrite_query --> analyzer_one;
+  	rewrite_query -.-> retriever_two;
+  
+  '''
+# ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic2[sqlite_aes].1
+  dict({
+    '$defs': dict({
+      'InnerObject': dict({
+        'properties': dict({
+          'yo': dict({
+            'title': 'Yo',
+            'type': 'integer',
+          }),
+        }),
+        'required': list([
+          'yo',
+        ]),
+        'title': 'InnerObject',
+        'type': 'object',
+      }),
+    }),
+    'properties': dict({
+      'inner': dict({
+        '$ref': '#/$defs/InnerObject',
+      }),
+      'query': dict({
+        'title': 'Query',
+        'type': 'string',
+      }),
+    }),
+    'required': list([
+      'query',
+      'inner',
+    ]),
+    'title': 'Input',
+    'type': 'object',
+  })
+# ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic2[sqlite_aes].2
+  dict({
+    'properties': dict({
+      'answer': dict({
+        'title': 'Answer',
+        'type': 'string',
+      }),
+      'docs': dict({
+        'items': dict({
+          'type': 'string',
+        }),
+        'title': 'Docs',
+        'type': 'array',
+      }),
+    }),
+    'required': list([
+      'answer',
+      'docs',
+    ]),
+    'title': 'Output',
+    'type': 'object',
+  })
+# ---
 # name: test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic_input[memory]
   '''
   graph TD;
@@ -1715,6 +1868,19 @@
   
   '''
 # ---
+# name: test_in_one_fan_out_state_graph_waiting_edge_via_branch[sqlite_aes]
+  '''
+  graph TD;
+  	__start__ --> rewrite_query;
+  	analyzer_one --> retriever_one;
+  	qa --> __end__;
+  	retriever_one --> qa;
+  	retriever_two --> qa;
+  	rewrite_query --> analyzer_one;
+  	rewrite_query -.-> retriever_two;
+  
+  '''
+# ---
 # name: test_multiple_sinks_subgraphs
   '''
   %%{init: {'flowchart': {'curve': 'linear'}}}%%

libs/langgraph/tests/conftest.py

@@ -16,6 +16,7 @@ from langgraph.checkpoint.postgres.aio import (
     AsyncPostgresSaver,
     AsyncShallowPostgresSaver,
 )
+from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
 from langgraph.checkpoint.sqlite import SqliteSaver
 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 from langgraph.store.base import BaseStore
@@ -61,6 +62,15 @@ def checkpointer_sqlite():
         yield checkpointer
 
 
+@pytest.fixture(scope="function")
+def checkpointer_sqlite_aes():
+    with SqliteSaver.from_conn_string(":memory:") as checkpointer:
+        checkpointer.serde = EncryptedSerializer.from_pycryptodome_aes(
+            key=b"1234567890123456"
+        )
+        yield checkpointer
+
+
 @asynccontextmanager
 async def _checkpointer_sqlite_aio():
     async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
@@ -437,6 +447,7 @@ REGULAR_CHECKPOINTERS_SYNC = [
     "postgres",
     "postgres_pipe",
     "postgres_pool",
+    "sqlite_aes",
 ]
 ALL_CHECKPOINTERS_SYNC = [
     *REGULAR_CHECKPOINTERS_SYNC,

libs/langgraph/tests/test_pregel.py

@@ -7317,3 +7317,297 @@ def test_empty_invoke() -> None:
         "111": 111,
         "222": 222,
     }
+
+
+@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
+def test_parallel_interrupts(
+    request: pytest.FixtureRequest, checkpointer_name: str
+) -> None:
+    from pydantic import BaseModel, Field
+
+    checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
+
+    # --- CHILD GRAPH ---
+
+    class ChildState(BaseModel):
+        prompt: str = Field(..., description="What is going to be asked to the user?")
+        human_input: Optional[str] = Field(None, description="What the human said")
+        human_inputs: Annotated[List[str], operator.add] = Field(
+            default_factory=list, description="All of my messages"
+        )
+
+    def get_human_input(state: ChildState):
+        human_input = interrupt(state.prompt)
+
+        return dict(
+            human_input=human_input,  # update child state
+            human_inputs=[human_input],  # update parent state
+        )
+
+    child_graph_builder = StateGraph(ChildState)
+    child_graph_builder.add_node("get_human_input", get_human_input)
+    child_graph_builder.add_edge(START, "get_human_input")
+    child_graph_builder.add_edge("get_human_input", END)
+    child_graph = child_graph_builder.compile()
+
+    # --- PARENT GRAPH ---
+
+    class ParentState(BaseModel):
+        prompts: List[str] = Field(
+            ..., description="What is going to be asked to the user?"
+        )
+        human_inputs: Annotated[List[str], operator.add] = Field(
+            default_factory=list, description="All of my messages"
+        )
+
+    def assign_workers(state: ParentState):
+        return [
+            Send(
+                "child_graph",
+                dict(
+                    prompt=prompt,
+                ),
+            )
+            for prompt in state.prompts
+        ]
+
+    def cleanup(state: ParentState):
+        assert len(state.human_inputs) == len(state.prompts)
+
+    parent_graph_builder = StateGraph(ParentState)
+    parent_graph_builder.add_node("child_graph", child_graph)
+    parent_graph_builder.add_node("cleanup", cleanup)
+
+    parent_graph_builder.add_conditional_edges(START, assign_workers, ["child_graph"])
+    parent_graph_builder.add_edge("child_graph", "cleanup")
+    parent_graph_builder.add_edge("cleanup", END)
+
+    parent_graph = parent_graph_builder.compile(checkpointer=checkpointer)
+
+    # --- CLIENT INVOCATION ---
+
+    thread_config = dict(
+        configurable=dict(
+            thread_id=str(uuid.uuid4()),
+        )
+    )
+    current_input = dict(
+        prompts=["a", "b"],
+    )
+
+    invokes = 0
+    events: dict[int, list[dict]] = {}
+    while invokes < 10:
+        # reset interrupt
+        invokes += 1
+        events[invokes] = []
+        current_interrupts: list[Interrupt] = []
+
+        # start / resume the graph
+        for event in parent_graph.stream(
+            input=current_input,
+            config=thread_config,
+            stream_mode="updates",
+        ):
+            events[invokes].append(event)
+            # handle the interrupt
+            if "__interrupt__" in event:
+                current_interrupts.extend(event["__interrupt__"])
+                # assume that it breaks here, because it is an interrupt
+
+        # get human input and resume
+        if any(i.resumable for i in current_interrupts):
+            current_input = Command(resume=f"Resume #{invokes}")
+
+        # not more human input required, must be completed
+        else:
+            break
+    else:
+        assert False, "Detected infinite loop"
+
+    assert invokes == 3
+    assert len(events) == 3
+
+    assert events[1] == UnsortedSequence(
+        {
+            "__interrupt__": (
+                Interrupt(
+                    value="a",
+                    resumable=True,
+                    ns=[
+                        AnyStr("child_graph:"),
+                        AnyStr("get_human_input:"),
+                    ],
+                ),
+            )
+        },
+        {
+            "__interrupt__": (
+                Interrupt(
+                    value="b",
+                    resumable=True,
+                    ns=[
+                        AnyStr("child_graph:"),
+                        AnyStr("get_human_input:"),
+                    ],
+                ),
+            )
+        },
+    )
+    assert events[2] in (
+        UnsortedSequence(
+            {
+                "__interrupt__": (
+                    Interrupt(
+                        value="a",
+                        resumable=True,
+                        ns=[
+                            AnyStr("child_graph:"),
+                            AnyStr("get_human_input:"),
+                        ],
+                    ),
+                )
+            },
+            {"child_graph": {"human_inputs": ["Resume #1"]}},
+        ),
+        UnsortedSequence(
+            {
+                "__interrupt__": (
+                    Interrupt(
+                        value="b",
+                        resumable=True,
+                        ns=[
+                            AnyStr("child_graph:"),
+                            AnyStr("get_human_input:"),
+                        ],
+                    ),
+                )
+            },
+            {"child_graph": {"human_inputs": ["Resume #1"]}},
+        ),
+    )
+    assert events[3] == UnsortedSequence(
+        {
+            "child_graph": {"human_inputs": ["Resume #1"]},
+            "__metadata__": {"cached": True},
+        },
+        {"child_graph": {"human_inputs": ["Resume #2"]}},
+        {"cleanup": None},
+    )
+
+
+@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
+def test_parallel_interrupts_double(
+    request: pytest.FixtureRequest, checkpointer_name: str
+) -> None:
+    from pydantic import BaseModel, Field
+
+    checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
+
+    # --- CHILD GRAPH ---
+
+    class ChildState(BaseModel):
+        prompt: str = Field(..., description="What is going to be asked to the user?")
+        human_input: Optional[str] = Field(None, description="What the human said")
+        human_inputs: Annotated[List[str], operator.add] = Field(
+            default_factory=list, description="All of my messages"
+        )
+
+    def get_human_input(state: ChildState):
+        human_input = interrupt(state.prompt)
+
+        return dict(
+            human_inputs=[human_input],  # update parent state
+        )
+
+    def get_dolphin_input(state: ChildState):
+        human_input = interrupt(state.prompt)
+
+        return dict(
+            human_inputs=[human_input],  # update parent state
+        )
+
+    child_graph_builder = StateGraph(ChildState)
+    child_graph_builder.add_node("get_human_input", get_human_input)
+    child_graph_builder.add_node("get_dolphin_input", get_dolphin_input)
+    child_graph_builder.add_edge(START, "get_human_input")
+    child_graph_builder.add_edge(START, "get_dolphin_input")
+    child_graph = child_graph_builder.compile()
+
+    # --- PARENT GRAPH ---
+
+    class ParentState(BaseModel):
+        prompts: List[str] = Field(
+            ..., description="What is going to be asked to the user?"
+        )
+        human_inputs: Annotated[List[str], operator.add] = Field(
+            default_factory=list, description="All of my messages"
+        )
+
+    def assign_workers(state: ParentState):
+        return [
+            Send(
+                "child_graph",
+                dict(
+                    prompt=prompt,
+                ),
+            )
+            for prompt in state.prompts
+        ]
+
+    def cleanup(state: ParentState):
+        assert len(state.human_inputs) == len(state.prompts) * 2
+
+    parent_graph_builder = StateGraph(ParentState)
+    parent_graph_builder.add_node("child_graph", child_graph)
+    parent_graph_builder.add_node("cleanup", cleanup)
+
+    parent_graph_builder.add_conditional_edges(START, assign_workers, ["child_graph"])
+    parent_graph_builder.add_edge("child_graph", "cleanup")
+    parent_graph_builder.add_edge("cleanup", END)
+
+    parent_graph = parent_graph_builder.compile(checkpointer=checkpointer)
+
+    # --- CLIENT INVOCATION ---
+
+    thread_config = dict(
+        configurable=dict(
+            thread_id=str(uuid.uuid4()),
+        )
+    )
+    current_input = dict(
+        prompts=["a", "b"],
+    )
+
+    invokes = 0
+    events: dict[int, list[dict]] = {}
+    while invokes < 10:
+        # reset interrupt
+        invokes += 1
+        events[invokes] = []
+        current_interrupts: list[Interrupt] = []
+
+        # start / resume the graph
+        for event in parent_graph.stream(
+            input=current_input,
+            config=thread_config,
+            stream_mode="updates",
+        ):
+            events[invokes].append(event)
+            # handle the interrupt
+            if "__interrupt__" in event:
+                current_interrupts.extend(event["__interrupt__"])
+                # assume that it breaks here, because it is an interrupt
+
+        # get human input and resume
+        if any(i.resumable for i in current_interrupts):
+            current_input = Command(resume=f"Resume #{invokes}")
+
+        # not more human input required, must be completed
+        else:
+            break
+    else:
+        assert False, "Detected infinite loop"
+
+    assert invokes == 5
+    assert len(events) == 5

libs/langgraph/tests/test_pregel_async.py

@@ -938,10 +938,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
             async for c in tool_two.astream(
                 {"my_key": "value ⛰️", "market": "DE"}, thread2
             )
-        ] == [
-            {
-                "tool_one": {"my_key": " one"},
-            },
+        ] == UnsortedSequence(
             {
                 "__interrupt__": (
                     Interrupt(
@@ -951,7 +948,10 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
                     ),
                 )
             },
-        ]
+            {
+                "tool_one": {"my_key": " one"},
+            },
+        )
         # resume with answer
         assert [
             c async for c in tool_two.astream(Command(resume=" my answer"), thread2)

libs/scheduler-kafka/tests/test_subgraph.py

@@ -202,7 +202,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": None,
@@ -275,7 +275,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -378,7 +378,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -491,7 +491,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": None,
@@ -559,7 +559,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -683,7 +683,7 @@ async def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],

libs/scheduler-kafka/tests/test_subgraph_sync.py

@@ -201,7 +201,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": None,
@@ -274,7 +274,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -377,7 +377,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -489,7 +489,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": None,
@@ -557,7 +557,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],
@@ -681,7 +681,7 @@ def test_subgraph_w_interrupt(
                                 "subgraph_counter": None,
                                 "call_counter": None,
                                 "interrupt_counter": None,
-                                "null_resume": None,
+                                "get_null_resume": None,
                                 "resume": [],
                             },
                             "checkpoint_id": c.config["configurable"]["checkpoint_id"],