checkpoint-postgres: release 2.0.3 (#2455)

checkpoint-postgres: handle cases when conn.pipeline is not supported

docs/docs/cloud/how-tos/interrupt_concurrent.md

@@ -94,6 +94,7 @@ Now we can start our two runs and join the second on euntil it has completed:
         assistant_id,
         input={"messages": [{"role": "user", "content": "what's the weather in sf?"}]},
     )
+    # sleep a bit to get partial outputs from the first run
     await asyncio.sleep(2)
     run = await client.runs.create(
         thread["thread_id"],
@@ -114,6 +115,7 @@ Now we can start our two runs and join the second on euntil it has completed:
       assistantId,
       { input: { messages: [{ role: "human", content: "what's the weather in sf?" }] } }
     );
+    // sleep a bit to get partial outputs from the first run
     await new Promise(resolve => setTimeout(resolve, 2000)); 
 
     let run = await client.runs.create(

docs/docs/cloud/how-tos/rollback_concurrent.md

@@ -95,7 +95,6 @@ Now let's run a thread with the multitask parameter set to "rollback":
         assistant_id,
         input={"messages": [{"role": "user", "content": "what's the weather in sf?"}]},
     )
-    await asyncio.sleep(2)
     run = await client.runs.create(
         thread["thread_id"],
         assistant_id,
@@ -115,7 +114,6 @@ Now let's run a thread with the multitask parameter set to "rollback":
       assistantId,
       { input: { messages: [{ role: "human", content: "what's the weather in sf?" }] } }
     );
-    await new Promise(resolve => setTimeout(resolve, 2000)); 
 
     let run = await client.runs.create(
       thread["thread_id"],
@@ -139,7 +137,7 @@ Now let's run a thread with the multitask parameter set to "rollback":
     --data "{
       \"assistant_id\": \"agent\",
       \"input\": {\"messages\": [{\"role\": \"human\", \"content\": \"what\'s the weather in sf?\"}]},
-    }" && sleep 2 && curl --request POST \
+    }" && curl --request POST \
     --url <DEPLOY<ENT_URL>>/threads/<THREAD_ID>/runs \
     --header 'Content-Type: application/json' \
     --data "{

docs/docs/cloud/reference/api/openapi.json

@@ -2868,9 +2868,18 @@
             "description": "The cron schedule to execute this job on."
           },
           "assistant_id": {
-            "type": "string",
-            "format": "uuid",
-            "title": "Assistant Id"
+            "anyOf": [
+              {
+                "type": "string",
+                "format": "uuid",
+                "title": "Assistant Id"
+              },
+              {
+                "type": "string",
+                "title": "Graph Id"
+              }
+            ],
+            "description": "The assistant ID or graph name to run. If using graph name, will default to the assistant automatically created from that graph by the server."
           },
           "input": {
             "anyOf": [
@@ -3171,6 +3180,66 @@
         ],
         "title": "Run"
       },
+      "Send": {
+        "type": "object",
+        "title": "Send",
+        "description": "A message to send to a node.",
+        "properties": {
+          "node": {
+            "type": "string",
+            "title": "Node",
+            "description": "The node to send the message to."
+          },
+          "input": {
+            "type": "object",
+            "title": "Message",
+            "description": "The message to send."
+          }
+        },
+        "required": [
+          "node",
+          "input"
+        ]
+      },
+      "Command": {
+        "type": "object",
+        "title": "Command",
+        "description": "The command to run.",
+        "properties": {
+          "update": {
+            "type": "object",
+            "title": "Update",
+            "description": "An update to the state."
+          },
+          "resume": {
+            "type": [
+              "object",
+              "array",
+              "number",
+              "string",
+              "null"
+            ],
+            "title": "Resume",
+            "description": "A value to pass to an interrupted node."
+          },
+          "send": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/Send"
+              },
+              {
+                "type": "array",
+                "items": {
+                  "$ref": "#/components/schemas/Send"
+                }
+              },
+              {
+                "type": "null"
+              }
+            ]
+          }
+        }
+      },
       "RunCreateStateful": {
         "properties": {
           "assistant_id": {
@@ -3196,13 +3265,19 @@
           "input": {
             "anyOf": [
               {
-                "items": {
-                  "type": "object"
-                },
-                "type": "array"
+                "type": "object"
               },
               {
-                "type": "object"
+                "type": "null"
+              }
+            ],
+            "title": "Input",
+            "description": "The input to the graph."
+          },
+          "command": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/Command"
               },
               {
                 "type": "null"
@@ -3405,13 +3480,19 @@
           "input": {
             "anyOf": [
               {
-                "items": {
-                  "type": "object"
-                },
-                "type": "array"
+                "type": "object"
               },
               {
-                "type": "object"
+                "type": "null"
+              }
+            ],
+            "title": "Input",
+            "description": "The input to the graph."
+          },
+          "command": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/Command"
               },
               {
                 "type": "null"

docs/docs/concepts/low_level.md

@@ -391,7 +391,7 @@ Read [this how-to](https://langchain-ai.github.io/langgraph/how-tos/recursion-li
 
 It can often be useful to set breakpoints before or after certain nodes execute. This can be used to wait for human approval before continuing. These can be set when you ["compile" a graph](#compiling-your-graph). You can set breakpoints either _before_ a node executes (using `interrupt_before`) or after a node executes (using `interrupt_after`.)
 
-You **MUST** use a [checkpoiner](./persistence.md) when using breakpoints. This is because your graph needs to be able to resume execution.
+You **MUST** use a [checkpointer](./persistence.md) when using breakpoints. This is because your graph needs to be able to resume execution.
 
 In order to resume execution, you can just invoke your graph with `None` as the input.
 

docs/docs/tutorials/code_assistant/langgraph_code_assistant.ipynb

@@ -102,7 +102,7 @@
     "from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader\n",
     "\n",
     "# LCEL docs\n",
-    "url = \"https://python.langchain.com/docs/concepts/#langchain-expression-language-lcel\"\n",
+    "url = \"https://python.langchain.com/docs/concepts/lcel/\"\n",
     "loader = RecursiveUrlLoader(\n",
     "    url=url, max_depth=20, extractor=lambda x: Soup(x, \"html.parser\").text\n",
     ")\n",

libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@@ -3,7 +3,7 @@ from contextlib import contextmanager
 from typing import Any, Iterator, Optional, Sequence, Union
 
 from langchain_core.runnables import RunnableConfig
-from psycopg import Connection, Cursor, Pipeline
+from psycopg import Capabilities, Connection, Cursor, Pipeline
 from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -52,6 +52,7 @@ class PostgresSaver(BasePostgresSaver):
         self.conn = conn
         self.pipe = pipe
         self.lock = threading.Lock()
+        self.supports_pipeline = Capabilities().has_pipeline()
 
     @classmethod
     @contextmanager
@@ -365,6 +366,13 @@ class PostgresSaver(BasePostgresSaver):
 
     @contextmanager
     def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
+        """Create a database cursor as a context manager.
+
+        Args:
+            pipeline (bool): whether to use pipeline for the DB operations inside the context manager.
+                Will be applied regardless of whether the PostgresSaver instance was initialized with a pipeline.
+                If pipeline mode is not supported, will fall back to using transaction context manager.
+        """
         with _get_connection(self.conn) as conn:
             if self.pipe:
                 # a connection in pipeline mode can be used concurrently
@@ -379,10 +387,17 @@ class PostgresSaver(BasePostgresSaver):
             elif pipeline:
                 # a connection not in pipeline mode can only be used by one
                 # thread/coroutine at a time, so we acquire a lock
-                with self.lock, conn.pipeline(), conn.cursor(
-                    binary=True, row_factory=dict_row
-                ) as cur:
-                    yield cur
+                if self.supports_pipeline:
+                    with self.lock, conn.pipeline(), conn.cursor(
+                        binary=True, row_factory=dict_row
+                    ) as cur:
+                        yield cur
+                else:
+                    # Use connection's transaction context manager when pipeline mode not supported
+                    with self.lock, conn.transaction(), conn.cursor(
+                        binary=True, row_factory=dict_row
+                    ) as cur:
+                        yield cur
             else:
                 with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur

libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@@ -3,7 +3,7 @@ from contextlib import asynccontextmanager
 from typing import Any, AsyncIterator, Iterator, Optional, Sequence, Union
 
 from langchain_core.runnables import RunnableConfig
-from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline
+from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
 from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -55,6 +55,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
         self.pipe = pipe
         self.lock = asyncio.Lock()
         self.loop = asyncio.get_running_loop()
+        self.supports_pipeline = Capabilities().has_pipeline()
 
     @classmethod
     @asynccontextmanager
@@ -323,6 +324,13 @@ class AsyncPostgresSaver(BasePostgresSaver):
     async def _cursor(
         self, *, pipeline: bool = False
     ) -> AsyncIterator[AsyncCursor[DictRow]]:
+        """Create a database cursor as a context manager.
+
+        Args:
+            pipeline (bool): whether to use pipeline for the DB operations inside the context manager.
+                Will be applied regardless of whether the AsyncPostgresSaver instance was initialized with a pipeline.
+                If pipeline mode is not supported, will fall back to using transaction context manager.
+        """
         async with _get_connection(self.conn) as conn:
             if self.pipe:
                 # a connection in pipeline mode can be used concurrently
@@ -337,10 +345,17 @@ class AsyncPostgresSaver(BasePostgresSaver):
             elif pipeline:
                 # a connection not in pipeline mode can only be used by one
                 # thread/coroutine at a time, so we acquire a lock
-                async with self.lock, conn.pipeline(), conn.cursor(
-                    binary=True, row_factory=dict_row
-                ) as cur:
-                    yield cur
+                if self.supports_pipeline:
+                    async with self.lock, conn.pipeline(), conn.cursor(
+                        binary=True, row_factory=dict_row
+                    ) as cur:
+                        yield cur
+                else:
+                    # Use connection's transaction context manager when pipeline mode not supported
+                    async with self.lock, conn.transaction(), conn.cursor(
+                        binary=True, row_factory=dict_row
+                    ) as cur:
+                        yield cur
             else:
                 async with self.lock, conn.cursor(
                     binary=True, row_factory=dict_row

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

@@ -133,6 +133,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
     INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
 
     jsonplus_serde = JsonPlusSerializer()
+    supports_pipeline: bool
 
     def _load_checkpoint(
         self,

libs/checkpoint-postgres/pyproject.toml

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

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -39,12 +39,13 @@ PendingWrite = Tuple[str, str, Any]
 class CheckpointMetadata(TypedDict, total=False):
     """Metadata associated with a checkpoint."""
 
-    source: Literal["input", "loop", "update"]
+    source: Literal["input", "loop", "update", "fork"]
     """The source of the checkpoint.
 
     - "input": The checkpoint was created from an input to invoke/stream/batch.
     - "loop": The checkpoint was created from inside the pregel loop.
     - "update": The checkpoint was created from a manual state update.
+    - "fork": The checkpoint was created as a copy of another checkpoint.
     """
     step: int
     """The step number of the checkpoint.

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -1,10 +1,14 @@
 import asyncio
+import logging
+import os
+import pickle
 import random
+import shutil
 from collections import defaultdict
-from contextlib import AbstractAsyncContextManager, AbstractContextManager
+from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from functools import partial
 from types import TracebackType
-from typing import Any, AsyncIterator, Dict, Iterator, Optional, Sequence, Tuple
+from typing import Any, AsyncIterator, Dict, Iterator, Optional, Sequence, Tuple, Type
 
 from langchain_core.runnables import RunnableConfig
 
@@ -20,6 +24,8 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol
 
+logger = logging.getLogger(__name__)
+
 
 class MemorySaver(
     BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
@@ -68,13 +74,18 @@ class MemorySaver(
         self,
         *,
         serde: Optional[SerializerProtocol] = None,
+        factory: Type[defaultdict] = defaultdict,
     ) -> None:
         super().__init__(serde=serde)
-        self.storage = defaultdict(lambda: defaultdict(dict))
-        self.writes = defaultdict(dict)
+        self.storage = factory(lambda: defaultdict(dict))
+        self.writes = factory(dict)
+        self.stack = ExitStack()
+        if factory is not defaultdict:
+            self.stack.enter_context(self.storage)  # type: ignore[arg-type]
+            self.stack.enter_context(self.writes)  # type: ignore[arg-type]
 
     def __enter__(self) -> "MemorySaver":
-        return self
+        return self.stack.__enter__()
 
     def __exit__(
         self,
@@ -82,10 +93,10 @@ class MemorySaver(
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> Optional[bool]:
-        return
+        return self.stack.__exit__(exc_type, exc_value, traceback)
 
     async def __aenter__(self) -> "MemorySaver":
-        return self
+        return self.stack.__enter__()
 
     async def __aexit__(
         self,
@@ -93,7 +104,7 @@ class MemorySaver(
         __exc_value: Optional[BaseException],
         __traceback: Optional[TracebackType],
     ) -> Optional[bool]:
-        return
+        return self.stack.__exit__(__exc_type, __exc_value, __traceback)
 
     def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
         """Get a checkpoint tuple from the in-memory storage.
@@ -478,3 +489,76 @@ class MemorySaver(
         next_v = current_v + 1
         next_h = random.random()
         return f"{next_v:032}.{next_h:016}"
+
+
+class PersistentDict(defaultdict):
+    """Persistent dictionary with an API compatible with shelve and anydbm.
+
+    The dict is kept in memory, so the dictionary operations run as fast as
+    a regular dictionary.
+
+    Write to disk is delayed until close or sync (similar to gdbm's fast mode).
+
+    Input file format is automatically discovered.
+    Output file format is selectable between pickle, json, and csv.
+    All three serialization formats are backed by fast C implementations.
+
+    Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/
+
+    """
+
+    def __init__(self, *args: Any, filename: str, **kwds: Any) -> None:
+        self.flag = "c"  # r=readonly, c=create, or n=new
+        self.mode = None  # None or an octal triple like 0644
+        self.format = "pickle"  # 'csv', 'json', or 'pickle'
+        self.filename = filename
+        super().__init__(*args, **kwds)
+
+    def sync(self) -> None:
+        "Write dict to disk"
+        if self.flag == "r":
+            return
+        tempname = self.filename + ".tmp"
+        fileobj = open(tempname, "wb" if self.format == "pickle" else "w")
+        try:
+            self.dump(fileobj)
+        except Exception:
+            os.remove(tempname)
+            raise
+        finally:
+            fileobj.close()
+        shutil.move(tempname, self.filename)  # atomic commit
+        if self.mode is not None:
+            os.chmod(self.filename, self.mode)
+
+    def close(self) -> None:
+        self.sync()
+        self.clear()
+
+    def __enter__(self) -> "PersistentDict":
+        return self
+
+    def __exit__(self, *exc_info: Any) -> None:
+        self.close()
+
+    def dump(self, fileobj: Any) -> None:
+        if self.format == "pickle":
+            pickle.dump(dict(self), fileobj, 2)
+        else:
+            raise NotImplementedError("Unknown format: " + repr(self.format))
+
+    def load(self) -> None:
+        # try formats from most restrictive to least restrictive
+        if self.flag == "n":
+            return
+        with open(self.filename, "rb" if self.format == "pickle" else "r") as fileobj:
+            for loader in (pickle.load,):
+                fileobj.seek(0)
+                try:
+                    return self.update(loader(fileobj))
+                except EOFError:
+                    return
+                except Exception:
+                    logging.error(f"Failed to load file: {fileobj.name}")
+                    raise
+            raise ValueError("File not in a supported f ormat")

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_store.py

@@ -314,7 +314,7 @@ async def test_cannot_put_empty_namespace() -> None:
     assert store.get(("langgraph", "foo"), "bar") is None
 
     class MockAsyncBatchedStore(AsyncBatchedBaseStore):
-        def __init__(self):
+        def __init__(self) -> None:
             super().__init__()
             self._store = InMemoryStore()
 
@@ -340,13 +340,17 @@ async def test_cannot_put_empty_namespace() -> None:
         await async_store.aput(("langgraph", "foo"), "bar", doc)
 
     await async_store.aput(("foo", "langgraph", "foo"), "bar", doc)
-    assert (await async_store.aget(("foo", "langgraph", "foo"), "bar")).value == doc
+    val = await async_store.aget(("foo", "langgraph", "foo"), "bar")
+    assert val is not None
+    assert val.value == doc
     assert (await async_store.asearch(("foo", "langgraph", "foo")))[0].value == doc
     await async_store.adelete(("foo", "langgraph", "foo"), "bar")
     assert (await async_store.aget(("foo", "langgraph", "foo"), "bar")) is None
 
     await async_store.abatch([PutOp(("valid", "namespace"), "key", doc)])
-    assert (await async_store.aget(("valid", "namespace"), "key")).value == doc
+    val = await async_store.aget(("valid", "namespace"), "key")
+    assert val is not None
+    assert val.value == doc
     assert (await async_store.asearch(("valid", "namespace")))[0].value == doc
     await async_store.adelete(("valid", "namespace"), "key")
     assert (await async_store.aget(("valid", "namespace"), "key")) is None

libs/langgraph/langgraph/pregel/__init__.py

@@ -65,6 +65,7 @@ from langgraph.constants import (
     CONFIG_KEY_STREAM,
     CONFIG_KEY_STREAM_WRITER,
     CONFIG_KEY_TASK_ID,
+    END,
     ERROR,
     INPUT,
     INTERRUPT,
@@ -901,8 +902,8 @@ class Pregel(PregelProtocol):
             checkpoint,
             LoopProtocol(config=config, step=step + 1, stop=step + 2),
         ) as (channels, managed):
-            # no values, just clear all tasks
-            if values is None and as_node is None:
+            # no values as END, just clear all tasks
+            if values is None and as_node == END:
                 if saved is not None:
                     # tasks for this checkpoint
                     next_tasks = prepare_next_tasks(
@@ -955,6 +956,42 @@ class Pregel(PregelProtocol):
                 return patch_checkpoint_map(
                     next_config, saved.metadata if saved else None
                 )
+            # no values, copy checkpoint
+            if values is None and as_node is None:
+                next_checkpoint = create_checkpoint(checkpoint, None, step)
+                # copy checkpoint
+                next_config = checkpointer.put(
+                    checkpoint_config,
+                    next_checkpoint,
+                    {
+                        **checkpoint_metadata,
+                        "source": "update",
+                        "step": step + 1,
+                        "writes": {},
+                        "parents": saved.metadata.get("parents", {}) if saved else {},
+                    },
+                    {},
+                )
+                return patch_checkpoint_map(
+                    next_config, saved.metadata if saved else None
+                )
+            if values is None and as_node == "__copy__":
+                next_checkpoint = create_checkpoint(checkpoint, None, step)
+                # copy checkpoint
+                next_config = checkpointer.put(
+                    saved.parent_config or saved.config if saved else checkpoint_config,
+                    next_checkpoint,
+                    {
+                        **checkpoint_metadata,
+                        "source": "fork",
+                        "step": step + 1,
+                        "parents": saved.metadata.get("parents", {}) if saved else {},
+                    },
+                    {},
+                )
+                return patch_checkpoint_map(
+                    next_config, saved.metadata if saved else None
+                )
             # apply pending writes, if not on specific checkpoint
             if (
                 CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
@@ -1144,7 +1181,7 @@ class Pregel(PregelProtocol):
             managed,
         ):
             # no values, just clear all tasks
-            if values is None and as_node is None:
+            if values is None and as_node == END:
                 if saved is not None:
                     # tasks for this checkpoint
                     next_tasks = prepare_next_tasks(
@@ -1197,6 +1234,42 @@ class Pregel(PregelProtocol):
                 return patch_checkpoint_map(
                     next_config, saved.metadata if saved else None
                 )
+            # no values, copy checkpoint
+            if values is None and as_node is None:
+                next_checkpoint = create_checkpoint(checkpoint, None, step)
+                # copy checkpoint
+                next_config = await checkpointer.aput(
+                    checkpoint_config,
+                    next_checkpoint,
+                    {
+                        **checkpoint_metadata,
+                        "source": "update",
+                        "step": step + 1,
+                        "writes": {},
+                        "parents": saved.metadata.get("parents", {}) if saved else {},
+                    },
+                    {},
+                )
+                return patch_checkpoint_map(
+                    next_config, saved.metadata if saved else None
+                )
+            if values is None and as_node == "__copy__":
+                next_checkpoint = create_checkpoint(checkpoint, None, step)
+                # copy checkpoint
+                next_config = await checkpointer.aput(
+                    saved.parent_config or saved.config if saved else checkpoint_config,
+                    next_checkpoint,
+                    {
+                        **checkpoint_metadata,
+                        "source": "fork",
+                        "step": step + 1,
+                        "parents": saved.metadata.get("parents", {}) if saved else {},
+                    },
+                    {},
+                )
+                return patch_checkpoint_map(
+                    next_config, saved.metadata if saved else None
+                )
             # apply pending writes, if not on specific checkpoint
             if (
                 CONFIG_KEY_CHECKPOINT_ID not in config[CONF]

libs/langgraph/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.2.48"
+version = "0.2.50"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"

libs/langgraph/tests/memory_assert.py

@@ -1,5 +1,8 @@
 import asyncio
+import os
+import tempfile
 from collections import defaultdict
+from functools import partial
 from typing import Any, Optional
 
 from langchain_core.runnables import RunnableConfig
@@ -12,7 +15,7 @@ from langgraph.checkpoint.base import (
     SerializerProtocol,
     copy_checkpoint,
 )
-from langgraph.checkpoint.memory import MemorySaver
+from langgraph.checkpoint.memory import MemorySaver, PersistentDict
 
 
 class NoopSerializer(SerializerProtocol):
@@ -32,9 +35,13 @@ class MemorySaverAssertImmutable(MemorySaver):
         serde: Optional[SerializerProtocol] = None,
         put_sleep: Optional[float] = None,
     ) -> None:
-        super().__init__(serde=serde)
+        _, filename = tempfile.mkstemp()
+        super().__init__(
+            serde=serde, factory=partial(PersistentDict, filename=filename)
+        )
         self.storage_for_copies = defaultdict(lambda: defaultdict(dict))
         self.put_sleep = put_sleep
+        self.stack.callback(os.remove, filename)
 
     def put(
         self,

libs/langgraph/tests/test_pregel.py

@@ -8556,8 +8556,8 @@ def test_dynamic_interrupt(
         parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config,
     )
     # clear the interrupt and next tasks
-    tool_two.update_state(thread1, None)
-    # interrupt is cleared, task will still run next
+    tool_two.update_state(thread1, None, as_node=END)
+    # interrupt and next tasks are cleared
     assert tool_two.get_state(thread1) == StateSnapshot(
         values={"my_key": "value ⛰️", "market": "DE"},
         next=(),
@@ -8575,6 +8575,167 @@ def test_dynamic_interrupt(
     )
 
 
+@pytest.mark.skipif(not FF_SEND_V2, reason="send v2 is not enabled")
+@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
+def test_copy_checkpoint(
+    request: pytest.FixtureRequest, checkpointer_name: str
+) -> None:
+    checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
+
+    class State(TypedDict):
+        my_key: Annotated[str, operator.add]
+        market: str
+
+    def tool_one(s: State) -> State:
+        return {"my_key": " one"}
+
+    tool_two_node_count = 0
+
+    def tool_two_node(s: State) -> State:
+        nonlocal tool_two_node_count
+        tool_two_node_count += 1
+        if s["market"] == "DE":
+            answer = interrupt("Just because...")
+        else:
+            answer = " all good"
+        return {"my_key": answer}
+
+    def start(state: State) -> list[Union[Send, str]]:
+        return ["tool_two", Send("tool_one", state)]
+
+    tool_two_graph = StateGraph(State)
+    tool_two_graph.add_node("tool_two", tool_two_node, retry=RetryPolicy())
+    tool_two_graph.add_node("tool_one", tool_one)
+    tool_two_graph.set_conditional_entry_point(start)
+    tool_two = tool_two_graph.compile()
+
+    tracer = FakeTracer()
+    assert tool_two.invoke(
+        {"my_key": "value", "market": "DE"}, {"callbacks": [tracer]}
+    ) == {
+        "my_key": "value one",
+        "market": "DE",
+    }
+    assert tool_two_node_count == 1, "interrupts aren't retried"
+    assert len(tracer.runs) == 1
+    run = tracer.runs[0]
+    assert run.end_time is not None
+    assert run.error is None
+    assert run.outputs == {"market": "DE", "my_key": "value one"}
+
+    assert tool_two.invoke({"my_key": "value", "market": "US"}) == {
+        "my_key": "value one all good",
+        "market": "US",
+    }
+
+    tool_two = tool_two_graph.compile(checkpointer=checkpointer)
+
+    # missing thread_id
+    with pytest.raises(ValueError, match="thread_id"):
+        tool_two.invoke({"my_key": "value", "market": "DE"})
+
+    # flow: interrupt -> resume with answer
+    thread2 = {"configurable": {"thread_id": "2"}}
+    # stop when about to enter node
+    assert [
+        c for c in tool_two.stream({"my_key": "value ⛰️", "market": "DE"}, thread2)
+    ] == [
+        {
+            "tool_one": {"my_key": " one"},
+        },
+        {
+            "__interrupt__": (
+                Interrupt(
+                    value="Just because...",
+                    resumable=True,
+                    ns=[AnyStr("tool_two:")],
+                ),
+            )
+        },
+    ]
+    # resume with answer
+    assert [c for c in tool_two.stream(Command(resume=" my answer"), thread2)] == [
+        {"tool_two": {"my_key": " my answer"}},
+    ]
+
+    # flow: interrupt -> clear tasks
+    thread1 = {"configurable": {"thread_id": "1"}}
+    # stop when about to enter node
+    assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
+        "my_key": "value ⛰️ one",
+        "market": "DE",
+    }
+    assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [
+        {
+            "parents": {},
+            "source": "loop",
+            "step": 0,
+            "writes": {"tool_one": {"my_key": " one"}},
+            "thread_id": "1",
+        },
+        {
+            "parents": {},
+            "source": "input",
+            "step": -1,
+            "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}},
+            "thread_id": "1",
+        },
+    ]
+    assert tool_two.get_state(thread1) == StateSnapshot(
+        values={"my_key": "value ⛰️ one", "market": "DE"},
+        next=("tool_two",),
+        tasks=(
+            PregelTask(
+                AnyStr(),
+                "tool_two",
+                (PULL, "tool_two"),
+                interrupts=(
+                    Interrupt(
+                        value="Just because...",
+                        resumable=True,
+                        ns=[AnyStr("tool_two:")],
+                    ),
+                ),
+            ),
+        ),
+        config=tool_two.checkpointer.get_tuple(thread1).config,
+        created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"],
+        metadata={
+            "parents": {},
+            "source": "loop",
+            "step": 0,
+            "writes": {"tool_one": {"my_key": " one"}},
+            "thread_id": "1",
+        },
+        parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config,
+    )
+    # clear the interrupt and next tasks
+    tool_two.update_state(thread1, None)
+    # interrupt is cleared, next task is kept
+    assert tool_two.get_state(thread1) == StateSnapshot(
+        values={"my_key": "value ⛰️ one", "market": "DE"},
+        next=("tool_two",),
+        tasks=(
+            PregelTask(
+                AnyStr(),
+                "tool_two",
+                (PULL, "tool_two"),
+                interrupts=(),
+            ),
+        ),
+        config=tool_two.checkpointer.get_tuple(thread1).config,
+        created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"],
+        metadata={
+            "parents": {},
+            "source": "update",
+            "step": 1,
+            "writes": {},
+            "thread_id": "1",
+        },
+        parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config,
+    )
+
+
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_start_branch_then(
     snapshot: SnapshotAssertion, request: pytest.FixtureRequest, checkpointer_name: str

libs/langgraph/tests/test_pregel_async.py

@@ -404,7 +404,7 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
         )
 
         # clear the interrupt and next tasks
-        await tool_two.aupdate_state(thread1, None)
+        await tool_two.aupdate_state(thread1, None, as_node=END)
         # interrupt is cleared, as well as the next tasks
         tup = await tool_two.checkpointer.aget_tuple(thread1)
         assert await tool_two.aget_state(thread1) == StateSnapshot(
@@ -426,6 +426,181 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
         )
 
 
+@pytest.mark.skipif(not FF_SEND_V2, reason="send v2 is not enabled")
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
+async def test_copy_checkpoint(checkpointer_name: str) -> None:
+    class State(TypedDict):
+        my_key: Annotated[str, operator.add]
+        market: str
+
+    def tool_one(s: State) -> State:
+        return {"my_key": " one"}
+
+    tool_two_node_count = 0
+
+    def tool_two_node(s: State) -> State:
+        nonlocal tool_two_node_count
+        tool_two_node_count += 1
+        if s["market"] == "DE":
+            answer = interrupt("Just because...")
+        else:
+            answer = " all good"
+        return {"my_key": answer}
+
+    def start(state: State) -> list[Union[Send, str]]:
+        return ["tool_two", Send("tool_one", state)]
+
+    tool_two_graph = StateGraph(State)
+    tool_two_graph.add_node("tool_two", tool_two_node, retry=RetryPolicy())
+    tool_two_graph.add_node("tool_one", tool_one)
+    tool_two_graph.set_conditional_entry_point(start)
+    tool_two = tool_two_graph.compile()
+
+    tracer = FakeTracer()
+    assert await tool_two.ainvoke(
+        {"my_key": "value", "market": "DE"}, {"callbacks": [tracer]}
+    ) == {
+        "my_key": "value one",
+        "market": "DE",
+    }
+    assert tool_two_node_count == 1, "interrupts aren't retried"
+    assert len(tracer.runs) == 1
+    run = tracer.runs[0]
+    assert run.end_time is not None
+    assert run.error is None
+    assert run.outputs == {"market": "DE", "my_key": "value one"}
+
+    assert await tool_two.ainvoke({"my_key": "value", "market": "US"}) == {
+        "my_key": "value one all good",
+        "market": "US",
+    }
+
+    async with awith_checkpointer(checkpointer_name) as checkpointer:
+        tool_two = tool_two_graph.compile(checkpointer=checkpointer)
+
+        # missing thread_id
+        with pytest.raises(ValueError, match="thread_id"):
+            await tool_two.ainvoke({"my_key": "value", "market": "DE"})
+
+        # flow: interrupt -> resume with answer
+        thread2 = {"configurable": {"thread_id": "2"}}
+        # stop when about to enter node
+        assert [
+            c
+            async for c in tool_two.astream(
+                {"my_key": "value ⛰️", "market": "DE"}, thread2
+            )
+        ] == [
+            {
+                "tool_one": {"my_key": " one"},
+            },
+            {
+                "__interrupt__": (
+                    Interrupt(
+                        value="Just because...",
+                        resumable=True,
+                        ns=[AnyStr("tool_two:")],
+                    ),
+                )
+            },
+        ]
+        # resume with answer
+        assert [
+            c async for c in tool_two.astream(Command(resume=" my answer"), thread2)
+        ] == [
+            {"tool_two": {"my_key": " my answer"}},
+        ]
+
+        # flow: interrupt -> clear tasks
+        thread1 = {"configurable": {"thread_id": "1"}}
+        # stop when about to enter node
+        assert await tool_two.ainvoke(
+            {"my_key": "value ⛰️", "market": "DE"}, thread1
+        ) == {
+            "my_key": "value ⛰️ one",
+            "market": "DE",
+        }
+        assert [c.metadata async for c in tool_two.checkpointer.alist(thread1)] == [
+            {
+                "parents": {},
+                "source": "loop",
+                "step": 0,
+                "writes": {"tool_one": {"my_key": " one"}},
+                "thread_id": "1",
+            },
+            {
+                "parents": {},
+                "source": "input",
+                "step": -1,
+                "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}},
+                "thread_id": "1",
+            },
+        ]
+        tup = await tool_two.checkpointer.aget_tuple(thread1)
+        assert await tool_two.aget_state(thread1) == StateSnapshot(
+            values={"my_key": "value ⛰️ one", "market": "DE"},
+            next=("tool_two",),
+            tasks=(
+                PregelTask(
+                    AnyStr(),
+                    "tool_two",
+                    (PULL, "tool_two"),
+                    interrupts=(
+                        Interrupt(
+                            value="Just because...",
+                            resumable=True,
+                            ns=[AnyStr("tool_two:")],
+                        ),
+                    ),
+                ),
+            ),
+            config=tup.config,
+            created_at=tup.checkpoint["ts"],
+            metadata={
+                "parents": {},
+                "source": "loop",
+                "step": 0,
+                "writes": {"tool_one": {"my_key": " one"}},
+                "thread_id": "1",
+            },
+            parent_config=[
+                c async for c in tool_two.checkpointer.alist(thread1, limit=2)
+            ][-1].config,
+        )
+        # clear the interrupt and next tasks
+        await tool_two.aupdate_state(thread1, None)
+        # interrupt is cleared, next task is kept
+        tup = await tool_two.checkpointer.aget_tuple(thread1)
+        assert await tool_two.aget_state(thread1) == StateSnapshot(
+            values={"my_key": "value ⛰️ one", "market": "DE"},
+            next=("tool_two",),
+            tasks=(
+                PregelTask(
+                    AnyStr(),
+                    "tool_two",
+                    (PULL, "tool_two"),
+                    interrupts=(),
+                ),
+            ),
+            config=tup.config,
+            created_at=tup.checkpoint["ts"],
+            metadata={
+                "parents": {},
+                "source": "update",
+                "step": 1,
+                "writes": {},
+                "thread_id": "1",
+            },
+            parent_config=[
+                c async for c in tool_two.checkpointer.alist(thread1, limit=2)
+            ][-1].config,
+        )
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="Python 3.11+ is required for async contextvars support",

libs/sdk-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@langchain/langgraph-sdk",
-  "version": "0.0.24",
+  "version": "0.0.25",
   "description": "Client library for interacting with the LangGraph API",
   "type": "module",
   "packageManager": "yarn@1.22.19",

libs/sdk-js/src/index.ts

@@ -10,6 +10,7 @@ export type {
   Metadata,
   Run,
   Thread,
+  ThreadTask,
   ThreadState,
   ThreadStatus,
   Cron,

libs/sdk-js/src/schema.ts

@@ -208,8 +208,14 @@ export interface ThreadState<ValuesType = DefaultValues> {
 export interface ThreadTask {
   id: string;
   name: string;
+  result?: unknown;
   error: Optional<string>;
-  interrupts: Array<{ value: unknown; when: "during" }>;
+  interrupts: Array<{
+    value: unknown;
+    when: "during";
+    resumable: boolean;
+    ns?: string[];
+  }>;
   checkpoint: Optional<Checkpoint>;
   state: Optional<ThreadState>;
 }