release(cli): 0.4.36 (#9330)

## Summary

<!-- Write the summary yourself -->

## Notes for reviewers
- Includes [push through the active Docker
context](https://github.com/langchain-ai/langgraph/pull/9321) and
[accept serverless and dedicated deployment
aliases](https://github.com/langchain-ai/langgraph/pull/9325), the two
CLI changes merged since 0.4.35.
- Merging doesn't publish by itself: after it merges, run the `release`
workflow with `working-directory: libs/cli`, as was done for 0.4.35.

## Test Plan
- [x] Only `__version__` in `libs/cli/langgraph_cli/__init__.py` changes
(0.4.35 → 0.4.36), like the 0.4.35 release PR
- [ ] After merge, the `release` workflow for `libs/cli` succeeds and
PyPI shows `langgraph-cli` 0.4.36
- [ ] `uvx --from langgraph-cli==0.4.36 langgraph deploy` pushes on
Docker Desktop without `DOCKER_HOST` set

## Demo

<!-- Drag a screen recording or screenshot here -->

libs/checkpoint-conformance/tests/test_validate_memory.py

@@ -14,8 +14,8 @@ async def memory_checkpointer():
 
 
 @pytest.mark.asyncio
-async def test_validate_memory_base():
-    """InMemorySaver passes all base capability tests."""
+async def test_validate_memory():
+    """InMemorySaver passes the tests of every capability it implements."""
     report = await validate(memory_checkpointer)
     report.print_report()
-    assert report.passed_all_base(), f"Base tests failed: {report.to_dict()}"
+    assert report.passed_all(), f"Capability tests failed: {report.to_dict()}"

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

@@ -27,10 +27,15 @@ from psycopg_pool import ConnectionPool
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
     _DELTA_PAGE_SIZE,
+    PRUNE_DELETE_THREADS_SQL,
+    PRUNE_STRATEGIES,
     BasePostgresSaver,
     _build_delta_stage1_sql,
     _build_delta_stage2_sql,
     _DeltaStage2Row,
+    _execute,
+    _keep_latest,
+    _refuse_delta_threads,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -401,6 +406,73 @@ class PostgresSaver(BasePostgresSaver):
                 (str(thread_id),),
             )
 
+    def prune(
+        self,
+        thread_ids: Sequence[str],
+        *,
+        strategy: str = "keep_latest",
+    ) -> None:
+        """Prune checkpoints for the given threads.
+
+        `"delete"` removes every checkpoint, write and blob of the threads, in
+        one transaction.
+
+        `"keep_latest"` keeps the latest checkpoint of each namespace (the one
+        `get_tuple` loads, plus its parent when it predates v4 and has pending
+        sends), with its pending writes and the blobs it references, and
+        deletes the rest. A kept checkpoint whose parent it deletes is left with
+        no parent. A subgraph call's namespace is deleted with the root
+        checkpoint the call started from, except a `checkpointer=True`
+        subgraph's, which keeps its latest checkpoint. Each thread is pruned in
+        its own transaction, on one snapshot of it, so a run appending to the
+        thread meanwhile is safe; forking or replaying from a checkpoint being
+        deleted is not. A thread that fails because another transaction changed
+        it leaves the threads before it pruned, and retrying is safe.
+
+        `"keep_latest"` refuses threads that use a `DeltaChannel`, since it
+        rebuilds the latest value from older checkpoints: if any requested
+        thread does, nothing is pruned. A thread that starts using one before
+        its turn in the batch is refused then, after the threads before it.
+
+        Args:
+            thread_ids: The thread IDs to prune.
+            strategy: `"keep_latest"` or `"delete"`.
+
+        Raises:
+            TypeError: If `thread_ids` is a single string.
+            ValueError: If `strategy` is unknown.
+            NotImplementedError: If `strategy` is `"keep_latest"` and a thread
+                uses a `DeltaChannel`.
+            psycopg.errors.SerializationFailure: If another transaction, such
+                as a second prune, changes a thread's rows while `"keep_latest"`
+                prunes it.
+            psycopg.errors.DeadlockDetected: If a second prune of a thread runs
+                at the same time.
+        """
+        if isinstance(thread_ids, str):
+            raise TypeError("thread_ids must be a sequence of thread IDs, not a str")
+        if strategy not in PRUNE_STRATEGIES:
+            raise ValueError(f"Unknown prune strategy: {strategy!r}")
+        thread_ids = [str(t) for t in thread_ids]
+        if not thread_ids:
+            return
+        if strategy == "delete":
+            with self._cursor() as cur, cur.connection.transaction():
+                for sql in PRUNE_DELETE_THREADS_SQL:
+                    cur.execute(sql, (thread_ids,))
+            return
+        with self._cursor() as cur:
+            _execute(cur, _refuse_delta_threads(thread_ids))
+        for thread_id in thread_ids:
+            with self._cursor() as cur:
+                if self.pipe:
+                    # The transaction has to start with its isolation level, and
+                    # in pipeline mode it would take in what ran since the last
+                    # sync.
+                    self.pipe.sync()
+                with cur.connection.transaction():
+                    _execute(cur, _keep_latest(thread_id))
+
     @contextmanager
     def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
         """Create a database cursor as a context manager.
@@ -451,7 +523,8 @@ class PostgresSaver(BasePostgresSaver):
         * Stage 1 (paged): dynamic SELECT over `checkpoints` with three
           columns per channel: its version, an `EXISTS` probe for a stored
           blob at that version, and its inline value. Pages newest-first by
-          `checkpoint_id` with a cursor; page size is `_DELTA_PAGE_SIZE`.
+          `checkpoint_id`, starting at the target; page size is
+          `_DELTA_PAGE_SIZE`.
           Stops paging when every channel has found its seed or a page comes
           back short.
 
@@ -468,7 +541,7 @@ class PostgresSaver(BasePostgresSaver):
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
-        if checkpoint_id is None:
+        if not checkpoint_id:
             target = self.get_tuple(config)
             if target is None:
                 return {ch: {"writes": []} for ch in channels}
@@ -476,7 +549,7 @@ class PostgresSaver(BasePostgresSaver):
 
         # Stage 1: paged K-JSONB-lookup scan, walking the parent chain in
         # Python after each page. Stops as soon as every channel has its seed.
-        stage1_sql = _build_delta_stage1_sql(channels, paged=True)
+        stage1_sql = _build_delta_stage1_sql(channels, paged=True, include_cursor=True)
         parent_of: dict[str, str | None] = {}
         ver_by_i_by_cid: list[dict[str, str | None]] = [{} for _ in channels]
         hb_by_i_by_cid: list[dict[str, bool]] = [{} for _ in channels]
@@ -486,7 +559,7 @@ class PostgresSaver(BasePostgresSaver):
         seed_inline_by_ch: dict[str, Any] = {}
         walk_cursor_by_ch: dict[str, str | None] = {}
         seeded: set[str] = set()
-        cursor: str | None = None
+        cursor: str | None = checkpoint_id
 
         with self._cursor() as cur:
             while True:
@@ -495,7 +568,7 @@ class PostgresSaver(BasePostgresSaver):
                     # ver_i, blob channel, blob version, inline_i
                     stage1_params.extend([ch, ch, ch, ch])
                 stage1_params.extend(
-                    [thread_id, checkpoint_ns, cursor, cursor, _DELTA_PAGE_SIZE]
+                    [thread_id, checkpoint_ns, cursor, _DELTA_PAGE_SIZE]
                 )
                 cur.execute(stage1_sql, stage1_params)
                 page = cur.fetchall()
@@ -527,6 +600,7 @@ class PostgresSaver(BasePostgresSaver):
                 if len(seeded) == len(channels) or len(page) < _DELTA_PAGE_SIZE:
                     break
                 cursor = oldest
+                stage1_sql = _build_delta_stage1_sql(channels, paged=True)
 
         # Stage 2: per-channel UNION ALL — one writes branch per channel
         # with non-empty chain, plus one blob branch per seeded channel.

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

@@ -27,10 +27,15 @@ from psycopg_pool import AsyncConnectionPool
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
     _DELTA_PAGE_SIZE,
+    PRUNE_DELETE_THREADS_SQL,
+    PRUNE_STRATEGIES,
     BasePostgresSaver,
+    _aexecute,
     _build_delta_stage1_sql,
     _build_delta_stage2_sql,
     _DeltaStage2Row,
+    _keep_latest,
+    _refuse_delta_threads,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -360,6 +365,55 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 (str(thread_id),),
             )
 
+    async def aprune(
+        self,
+        thread_ids: Sequence[str],
+        *,
+        strategy: str = "keep_latest",
+    ) -> None:
+        """Prune checkpoints for the given threads.
+
+        See `PostgresSaver.prune` for what each strategy keeps.
+
+        Args:
+            thread_ids: The thread IDs to prune.
+            strategy: `"keep_latest"` or `"delete"`.
+
+        Raises:
+            TypeError: If `thread_ids` is a single string.
+            ValueError: If `strategy` is unknown.
+            NotImplementedError: If `strategy` is `"keep_latest"` and a thread
+                uses a `DeltaChannel`.
+            psycopg.errors.SerializationFailure: If another transaction, such
+                as a second prune, changes a thread's rows while `"keep_latest"`
+                prunes it.
+            psycopg.errors.DeadlockDetected: If a second prune of a thread runs
+                at the same time.
+        """
+        if isinstance(thread_ids, str):
+            raise TypeError("thread_ids must be a sequence of thread IDs, not a str")
+        if strategy not in PRUNE_STRATEGIES:
+            raise ValueError(f"Unknown prune strategy: {strategy!r}")
+        thread_ids = [str(t) for t in thread_ids]
+        if not thread_ids:
+            return
+        if strategy == "delete":
+            async with self._cursor() as cur, cur.connection.transaction():
+                for sql in PRUNE_DELETE_THREADS_SQL:
+                    await cur.execute(sql, (thread_ids,))
+            return
+        async with self._cursor() as cur:
+            await _aexecute(cur, _refuse_delta_threads(thread_ids))
+        for thread_id in thread_ids:
+            async with self._cursor() as cur:
+                if self.pipe:
+                    # The transaction has to start with its isolation level, and
+                    # in pipeline mode it would take in what ran since the last
+                    # sync.
+                    await self.pipe.sync()
+                async with cur.connection.transaction():
+                    await _aexecute(cur, _keep_latest(thread_id))
+
     @asynccontextmanager
     async def _cursor(
         self, *, pipeline: bool = False
@@ -417,13 +471,13 @@ class AsyncPostgresSaver(BasePostgresSaver):
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
-        if checkpoint_id is None:
+        if not checkpoint_id:
             target = await self.aget_tuple(config)
             if target is None:
                 return {ch: {"writes": []} for ch in channels}
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
-        stage1_sql = _build_delta_stage1_sql(channels, paged=True)
+        stage1_sql = _build_delta_stage1_sql(channels, paged=True, include_cursor=True)
         parent_of: dict[str, str | None] = {}
         ver_by_i_by_cid: list[dict[str, str | None]] = [{} for _ in channels]
         hb_by_i_by_cid: list[dict[str, bool]] = [{} for _ in channels]
@@ -433,7 +487,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
         seed_inline_by_ch: dict[str, Any] = {}
         walk_cursor_by_ch: dict[str, str | None] = {}
         seeded: set[str] = set()
-        cursor: str | None = None
+        cursor: str | None = checkpoint_id
 
         async with self._cursor() as cur:
             while True:
@@ -442,7 +496,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                     # ver_i, blob channel, blob version, inline_i
                     stage1_params.extend([ch, ch, ch, ch])
                 stage1_params.extend(
-                    [thread_id, checkpoint_ns, cursor, cursor, _DELTA_PAGE_SIZE]
+                    [thread_id, checkpoint_ns, cursor, _DELTA_PAGE_SIZE]
                 )
                 await cur.execute(stage1_sql, stage1_params)
                 page = await cur.fetchall()
@@ -472,6 +526,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 if len(seeded) == len(channels) or len(page) < _DELTA_PAGE_SIZE:
                     break
                 cursor = oldest
+                stage1_sql = _build_delta_stage1_sql(channels, paged=True)
 
         channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
         channels_with_seed = [ch for ch in channels if seed_ver_by_ch[ch] is not None]
@@ -687,6 +742,33 @@ class AsyncPostgresSaver(BasePostgresSaver):
             self.adelete_thread(thread_id), self.loop
         ).result()
 
+    def prune(
+        self,
+        thread_ids: Sequence[str],
+        *,
+        strategy: str = "keep_latest",
+    ) -> None:
+        """Prune checkpoints for the given threads.
+
+        See `PostgresSaver.prune` for what each strategy keeps.
+
+        Args:
+            thread_ids: The thread IDs to prune.
+            strategy: `"keep_latest"` or `"delete"`.
+        """
+        try:
+            if asyncio.get_running_loop() is self.loop:
+                raise asyncio.InvalidStateError(
+                    "Synchronous calls to AsyncPostgresSaver are only allowed from a "
+                    "different thread. From the main thread, use the async interface. "
+                    "For example, use `await checkpointer.aprune(...)`."
+                )
+        except RuntimeError:
+            pass
+        return asyncio.run_coroutine_threadsafe(
+            self.aprune(thread_ids, strategy=strategy), self.loop
+        ).result()
+
     def get_delta_channel_history(
         self, *, config: RunnableConfig, channels: Sequence[str]
     ) -> Mapping[str, DeltaChannelHistory]:

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import random
 import warnings
-from collections.abc import Mapping, Sequence
+from collections.abc import Generator, Mapping, Sequence
 from importlib.metadata import version as get_version
 from typing import Any, TypedDict, cast
 
@@ -17,6 +17,8 @@ from langgraph.checkpoint.base import (
     writes_sort_key,
 )
 from langgraph.checkpoint.serde.types import TASKS
+from psycopg import AsyncCursor, Cursor
+from psycopg.rows import DictRow
 from psycopg.types.json import Jsonb
 
 # Page size for stage-1 paged scan in `get_delta_channel_history`. Internal
@@ -160,6 +162,218 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """
 
 
+PRUNE_STRATEGIES = ("keep_latest", "delete")
+
+PRUNE_DELETE_THREADS_SQL = (
+    "DELETE FROM checkpoints WHERE thread_id = ANY(%s)",
+    "DELETE FROM checkpoint_blobs WHERE thread_id = ANY(%s)",
+    "DELETE FROM checkpoint_writes WHERE thread_id = ANY(%s)",
+)
+
+# The prune statements below never join two sets of a thread's rows: a join is
+# planned from per-thread row estimates, which are worst for the few huge
+# threads worth pruning, and can go quadratic. They look rows up by primary key
+# instead (`LATERAL ... LIMIT 1`, `= ANY`).
+
+# A `DeltaChannel` leaves update counters in checkpoint metadata, or a
+# snapshot: an inline `true` marker with a blob behind it, which a plain `True`
+# value never has. LangGraph.js inlines no values, but marks the threads its
+# DeltaChannel graphs start with `delta_writes_versioned`.
+PRUNE_DELTA_THREADS_SQL = """
+SELECT thread_id FROM checkpoints
+WHERE thread_id = ANY(%s) AND (
+    metadata ? 'counters_since_delta_snapshot'
+    OR metadata ? 'delta_writes_versioned'
+)
+UNION
+SELECT c.thread_id
+FROM checkpoints c
+CROSS JOIN LATERAL jsonb_each(c.checkpoint -> 'channel_values') AS v
+CROSS JOIN LATERAL (
+    SELECT 1 FROM checkpoint_blobs b
+    WHERE b.thread_id = c.thread_id
+    AND b.checkpoint_ns = c.checkpoint_ns
+    AND b.channel = v.key
+    AND b.version = c.checkpoint -> 'channel_versions' ->> v.key
+    LIMIT 1
+) AS snapshot
+WHERE c.thread_id = ANY(%s) AND v.value = 'true'::jsonb
+"""
+
+# Per namespace: the latest checkpoint, the parent it reads its pending sends
+# from when it predates v4, and the root graph's checkpoint its subgraph call
+# started from, with whether that one is saved.
+PRUNE_HEADS_SQL = f"""
+SELECT
+    h.checkpoint_ns,
+    h.checkpoint_id AS head,
+    CASE WHEN (c.checkpoint ->> 'v')::int < 4 AND EXISTS (
+        SELECT 1 FROM checkpoint_writes w
+        WHERE w.thread_id = %s
+        AND w.checkpoint_ns = h.checkpoint_ns
+        AND w.checkpoint_id = c.parent_checkpoint_id
+        AND w.channel = '{TASKS}'
+    ) THEN c.parent_checkpoint_id END AS sends_parent,
+    c.metadata -> 'parents' ->> '' AS root,
+    EXISTS (
+        SELECT 1 FROM checkpoints r
+        WHERE r.thread_id = %s AND r.checkpoint_ns = ''
+        AND r.checkpoint_id = c.metadata -> 'parents' ->> ''
+    ) AS root_saved
+FROM (
+    SELECT DISTINCT ON (checkpoint_ns) checkpoint_ns, checkpoint_id
+    FROM checkpoints
+    WHERE thread_id = %s
+    ORDER BY checkpoint_ns DESC, checkpoint_id DESC
+) AS h
+CROSS JOIN LATERAL (
+    SELECT checkpoint, parent_checkpoint_id, metadata FROM checkpoints
+    WHERE thread_id = %s
+    AND checkpoint_ns = h.checkpoint_ns
+    AND checkpoint_id = h.checkpoint_id
+) AS c
+"""
+
+PRUNE_SNAPSHOT_SQL = "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"
+
+PRUNE_UNKEPT_SQL = """
+SELECT checkpoint_id FROM checkpoints
+WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id <> ALL(%s)
+"""
+
+# A run saves a superstep's writes in the background, where they can commit
+# before their checkpoint does, so they go by the checkpoints deleted.
+PRUNE_WRITES_SQL = """
+DELETE FROM checkpoint_writes
+WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = ANY(%s)
+"""
+
+PRUNE_CHECKPOINTS_SQL = """
+WITH deleted AS (
+    DELETE FROM checkpoints
+    WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id <> ALL(%s)
+    RETURNING checkpoint -> 'channel_versions' AS versions
+)
+SELECT DISTINCT v.key AS channel, v.value AS version
+FROM deleted
+CROSS JOIN LATERAL jsonb_each_text(deleted.versions) AS v
+"""
+
+PRUNE_PARENTS_SQL = """
+UPDATE checkpoints SET parent_checkpoint_id = NULL
+WHERE thread_id = %s AND checkpoint_ns = %s
+AND checkpoint_id = ANY(%s) AND parent_checkpoint_id <> ALL(%s)
+"""
+
+PRUNE_REFERENCED_BLOBS_SQL = """
+SELECT DISTINCT v.key AS channel, v.value AS version
+FROM checkpoints c
+CROSS JOIN LATERAL jsonb_each_text(c.checkpoint -> 'channel_versions') AS v
+WHERE c.thread_id = %s AND c.checkpoint_ns = %s
+"""
+
+PRUNE_BLOBS_SQL = """
+DELETE FROM checkpoint_blobs
+WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s AND version = ANY(%s)
+"""
+
+PRUNE_SUBGRAPH_CALLS_SQL = (
+    "DELETE FROM checkpoints WHERE thread_id = %s AND checkpoint_ns = ANY(%s)",
+    "DELETE FROM checkpoint_writes WHERE thread_id = %s AND checkpoint_ns = ANY(%s)",
+    "DELETE FROM checkpoint_blobs WHERE thread_id = %s AND checkpoint_ns = ANY(%s)",
+)
+
+
+def _garbage_blobs(
+    freed: Sequence[Mapping[str, Any]], referenced: Sequence[Mapping[str, Any]]
+) -> dict[str, list[str]]:
+    """Blobs a deleted checkpoint used and no kept one does, by channel."""
+    kept = {(r["channel"], r["version"]) for r in referenced}
+    garbage: dict[str, list[str]] = {}
+    for r in freed:
+        if (r["channel"], r["version"]) not in kept:
+            garbage.setdefault(r["channel"], []).append(r["version"])
+    return garbage
+
+
+_Statements = Generator[tuple[str, Sequence[Any], bool], list[DictRow], None]
+
+
+def _refuse_delta_threads(thread_ids: Sequence[str]) -> _Statements:
+    rows = yield PRUNE_DELTA_THREADS_SQL, (thread_ids, thread_ids), True
+    if delta := sorted(r["thread_id"] for r in rows):
+        raise NotImplementedError(
+            f"keep_latest can't prune threads that use a DeltaChannel: {', '.join(delta)}"
+        )
+
+
+def _keep_latest(thread_id: str) -> _Statements:
+    """The statements of one thread's keep_latest transaction, for `_execute`
+    and `_aexecute` to run. Each comes with whether its rows are sent back.
+
+    They all read one snapshot, so what a run saves meanwhile is neither seen
+    nor deleted, whatever its checkpoint ID.
+    """
+    yield PRUNE_SNAPSHOT_SQL, (), False
+    heads = yield PRUNE_HEADS_SQL, (thread_id,) * 4, True
+    # Catches a thread that started using a DeltaChannel after the batch check.
+    yield from _refuse_delta_threads([thread_id])
+    kept = {
+        h["checkpoint_ns"]: [i for i in (h["head"], h["sends_parent"]) if i]
+        for h in heads
+    }
+    # A subgraph call's namespace has its task ID after a ":", unlike a
+    # `checkpointer=True` subgraph's, whose state outlives each call. A call
+    # goes with the root checkpoint it started from, but only once that one is
+    # saved: runs save in the background, so a call can commit first.
+    calls = {
+        h["checkpoint_ns"]
+        for h in heads
+        if ":" in h["checkpoint_ns"]
+        and h["root_saved"]
+        and h["root"] not in kept.get("", ())
+    }
+    if calls:
+        for sql in PRUNE_SUBGRAPH_CALLS_SQL:
+            yield sql, (thread_id, list(calls)), False
+    for ns, ids in kept.items():
+        if ns in calls:
+            continue
+        unkept = yield PRUNE_UNKEPT_SQL, (thread_id, ns, ids), True
+        if unkept:
+            deleted = [r["checkpoint_id"] for r in unkept]
+            yield PRUNE_WRITES_SQL, (thread_id, ns, deleted), False
+        freed = yield PRUNE_CHECKPOINTS_SQL, (thread_id, ns, ids), True
+        yield PRUNE_PARENTS_SQL, (thread_id, ns, ids, ids), False
+        if freed:
+            referenced = yield PRUNE_REFERENCED_BLOBS_SQL, (thread_id, ns), True
+            for channel, versions in _garbage_blobs(freed, referenced).items():
+                yield PRUNE_BLOBS_SQL, (thread_id, ns, channel, versions), False
+
+
+# Unprepared, so Postgres plans each statement knowing its arrays: a cached
+# generic plan checks a row against an `= ANY(...)` array one element at a time.
+def _execute(cur: Cursor[DictRow], statements: _Statements) -> None:
+    try:
+        sql, params, fetch = next(statements)
+        while True:
+            cur.execute(sql, params, prepare=False)
+            sql, params, fetch = statements.send(cur.fetchall() if fetch else [])
+    except StopIteration:
+        pass
+
+
+async def _aexecute(cur: AsyncCursor[DictRow], statements: _Statements) -> None:
+    try:
+        sql, params, fetch = next(statements)
+        while True:
+            await cur.execute(sql, params, prepare=False)
+            rows = await cur.fetchall() if fetch else []
+            sql, params, fetch = statements.send(rows)
+    except StopIteration:
+        pass
+
+
 class _DeltaStage2Row(TypedDict, total=False):
     """One row from `_build_delta_stage2_sql` (a UNION ALL of writes and blobs)."""
 
@@ -178,10 +392,13 @@ class _DeltaStage2Row(TypedDict, total=False):
 # `_build_delta_stage2_sql` document their shapes.
 
 
-def _build_delta_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
+def _build_delta_stage1_sql(
+    channels: Sequence[str], *, paged: bool, include_cursor: bool = False
+) -> str:
     """Build stage 1 SQL with K parallel version lookups + seed probes.
 
-    For channels=["messages", "files"] (with `paged=True`) the result is::
+    For channels=["messages", "files"] (with `paged=True, include_cursor=True`)
+    the result is::
 
         SELECT checkpoint_id, parent_checkpoint_id,
                checkpoint -> 'channel_versions' ->> %s AS ver_0,
@@ -197,7 +414,7 @@ def _build_delta_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
                checkpoint -> 'channel_values' -> %s AS inline_1
         FROM checkpoints
         WHERE thread_id = %s AND checkpoint_ns = %s
-          AND (%s::text IS NULL OR checkpoint_id < %s)
+          AND checkpoint_id <= %s
         ORDER BY checkpoint_id DESC
         LIMIT %s
 
@@ -238,9 +455,16 @@ def _build_delta_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
     and uses safe identifiers).
 
     Caller must extend params with `[ch_0 x4, ch_1 x4, ..., thread_id, ns,
-    cursor, cursor, page_size]` when `paged=True` — four per channel: the
-    version lookup, the blob's channel, the version the blob must match, and the
-    inline lookup.
+    cursor, page_size]` when `paged=True` — four per channel: the version
+    lookup, the blob's channel, the version the blob must match, and the inline
+    lookup.
+
+    Pages run newest-first from the target down. The first page passes the
+    target as the cursor with `include_cursor=True`, so it opens with the
+    target's own row, whose parent starts the walk; each later page continues
+    below the oldest row read. A checkpoint's ancestors have smaller ids (uuid6
+    is time-ordered, which `get_tuple` also relies on to find the latest
+    checkpoint), so no row newer than the target is part of its chain.
 
     When `paged=False`, the WHERE has no cursor predicate and there's no
     LIMIT/ORDER BY — kept as a non-public helper for tests/diagnostics.
@@ -264,7 +488,7 @@ def _build_delta_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
     )
     if paged:
         sql += (
-            " AND (%s::text IS NULL OR checkpoint_id < %s)"
+            f" AND checkpoint_id {'<=' if include_cursor else '<'} %s"
             " ORDER BY checkpoint_id DESC LIMIT %s"
         )
     return sql
@@ -413,8 +637,8 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
               materialized at this point),
           (c) the next ancestor cid isn't in `parent_of` yet (waiting for
               a later page; the cursor stays put), or
-          (d) the target's own row isn't in `parent_of` yet (the walk has
-              not started; no cursor is set, so a later page retries).
+          (d) the target's own row isn't in `parent_of` (the target doesn't
+              exist, so the walk never starts).
 
         Mutates `chain_by_ch`, `seed_ver_by_ch`, `seed_inline_by_ch`,
         `walk_cursor_by_ch`, and `seeded` in place.
@@ -422,8 +646,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         for i, ch in enumerate(channels):
             if ch in seeded:
                 continue
-            # Pages start at the thread head, so the target may not have
-            # loaded yet; a `None` cursor would read as "target is a root".
+            # The first page opens with the target's row, so it's missing only
+            # when the target doesn't exist; a `None` cursor would read as
+            # "target is a root".
             if ch not in walk_cursor_by_ch:
                 if target_id not in parent_of:
                     continue

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

@@ -36,6 +36,7 @@ from langgraph.store.base import (
     ensure_embeddings,
     get_text_at_path,
     tokenize_path,
+    validate_op_namespace,
 )
 from psycopg import Capabilities, Connection, Cursor, Pipeline
 from psycopg.rows import DictRow, dict_row
@@ -1386,6 +1387,7 @@ def _group_ops(ops: Iterable[Op]) -> tuple[dict[type, list[tuple[int, Op]]], int
     grouped_ops: dict[type, list[tuple[int, Op]]] = defaultdict(list)
     tot = 0
     for idx, op in enumerate(ops):
+        validate_op_namespace(op)
         grouped_ops[type(op)].append((idx, op))
         tot += 1
     return grouped_ops, tot

libs/checkpoint-postgres/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.2"
+version = "3.2.0"
 description = "Library with a Postgres implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"

libs/checkpoint-postgres/tests/test_async_store.py

@@ -13,8 +13,10 @@ import pytest
 from langchain_core.embeddings import Embeddings
 from langgraph.store.base import (
     GetOp,
+    InvalidNamespaceError,
     Item,
     ListNamespacesOp,
+    MatchCondition,
     PutOp,
     SearchOp,
 )
@@ -871,3 +873,65 @@ async def test_omit_expired_search_pagination(store: AsyncPostgresStore) -> None
     page2 = await store.asearch(ns, limit=2, offset=2)
     assert [i.key for i in page1] == ["a", "b"]
     assert [i.key for i in page2] == ["c"]
+
+
+@pytest.mark.parametrize("namespace", [("foo.bar",), ("foo", ""), ("foo", 1)])
+async def test_abatch_rejects_invalid_namespace_labels(
+    store: AsyncPostgresStore, namespace: tuple
+) -> None:
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    for op in (
+        GetOp(namespace, "key"),
+        GetOp(namespace, "key", refresh_ttl=True),
+        PutOp(namespace, "key", {"changed": True}),
+        PutOp(namespace, "key", None),
+        SearchOp(namespace),
+        ListNamespacesOp((MatchCondition("prefix", namespace),)),
+        ListNamespacesOp((MatchCondition("suffix", namespace),)),
+    ):
+        with pytest.raises(InvalidNamespaceError):
+            await store.abatch([op])
+
+    item = await store.aget(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}
+
+
+async def test_invalid_namespace_only_fails_its_own_call(
+    store: AsyncPostgresStore,
+) -> None:
+    """Concurrent calls share one `abatch`, which fails every op if it raises.
+
+    Labels are checked before an op is queued, so one caller's bad label cannot
+    fail another caller's request.
+    """
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    valid, invalid = await asyncio.gather(
+        store.aget(("foo", "bar"), "key"),
+        store.aget(("foo.bar",), "key"),
+        return_exceptions=True,
+    )
+
+    assert isinstance(valid, Item) and valid.value == {"original": True}
+    assert isinstance(invalid, InvalidNamespaceError)
+
+
+async def test_sync_methods_reject_invalid_namespace_labels(
+    store: AsyncPostgresStore,
+) -> None:
+    """The sync wrappers run off the event loop thread and must validate too."""
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    for call in (
+        lambda: store.get(("foo.bar",), "key"),
+        lambda: store.search(("foo.bar",)),
+        lambda: store.delete(("foo.bar",), "key"),
+        lambda: store.list_namespaces(prefix=("foo.bar",)),
+        lambda: store.batch([GetOp(("foo.bar",), "key")]),
+    ):
+        with pytest.raises(InvalidNamespaceError):
+            await asyncio.to_thread(call)
+
+    item = await store.aget(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}

libs/checkpoint-postgres/tests/test_delta_pagination.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping, Sequence
 from typing import Any
 from uuid import uuid4
 
@@ -14,7 +15,7 @@ from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 from langgraph.checkpoint.postgres import PostgresSaver
 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-from langgraph.checkpoint.postgres.base import _DELTA_PAGE_SIZE
+from langgraph.checkpoint.postgres.base import _DELTA_PAGE_SIZE, BasePostgresSaver
 from tests.conftest import DEFAULT_URI
 
 CHANNEL = "items"
@@ -23,8 +24,8 @@ SEED_STEP = 1
 SEED_VALUE = [10, 20]
 TARGET_STEP = 4
 
-# The real page size is the control; the rest leave the target off the first
-# page (three checkpoints are newer than it).
+# The real page size is the control; the rest split the walk from the target
+# to its seed across pages.
 PAGE_SIZES = [_DELTA_PAGE_SIZE, 3, 2, 1]
 
 
@@ -92,7 +93,7 @@ def _assert_history(entry: DeltaChannelHistory, page_size: int) -> None:
 
 
 @pytest.mark.parametrize("page_size", PAGE_SIZES)
-async def test_async_target_older_than_the_first_page(
+async def test_async_walk_continues_across_pages(
     page_size: int, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     monkeypatch.setattr("langgraph.checkpoint.postgres.aio._DELTA_PAGE_SIZE", page_size)
@@ -106,7 +107,7 @@ async def test_async_target_older_than_the_first_page(
 
 
 @pytest.mark.parametrize("page_size", PAGE_SIZES)
-def test_sync_target_older_than_the_first_page(
+def test_sync_walk_continues_across_pages(
     page_size: int, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     monkeypatch.setattr("langgraph.checkpoint.postgres._DELTA_PAGE_SIZE", page_size)
@@ -119,6 +120,84 @@ def test_sync_target_older_than_the_first_page(
         _assert_history(result[CHANNEL], page_size)
 
 
+def _record_rows_read(monkeypatch: pytest.MonkeyPatch) -> list[str]:
+    read: list[str] = []
+    ingest = BasePostgresSaver._ingest_stage1_page
+
+    def record(rows: Sequence[Mapping[str, Any]], *args: Any) -> str | None:
+        read.extend(row["checkpoint_id"] for row in rows)
+        return ingest(rows, *args)
+
+    monkeypatch.setattr(BasePostgresSaver, "_ingest_stage1_page", staticmethod(record))
+    return read
+
+
+def _ids_from_target_down(configs: list[dict]) -> list[str]:
+    return [c["configurable"]["checkpoint_id"] for c in configs[TARGET_STEP::-1]]
+
+
+async def test_async_walk_reads_nothing_newer_than_the_target(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    read = _record_rows_read(monkeypatch)
+    async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+        await saver.setup()
+        configs = await _abuild_chain(saver)
+        result = await saver.aget_delta_channel_history(
+            config=configs[TARGET_STEP], channels=[CHANNEL]
+        )
+    _assert_history(result[CHANNEL], _DELTA_PAGE_SIZE)
+    assert read == _ids_from_target_down(configs)
+
+
+def test_sync_walk_reads_nothing_newer_than_the_target(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    read = _record_rows_read(monkeypatch)
+    with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+        saver.setup()
+        configs = _build_chain(saver)
+        result = saver.get_delta_channel_history(
+            config=configs[TARGET_STEP], channels=[CHANNEL]
+        )
+    _assert_history(result[CHANNEL], _DELTA_PAGE_SIZE)
+    assert read == _ids_from_target_down(configs)
+
+
+def _empty_checkpoint_id(config: dict) -> dict:
+    return {"configurable": {**config["configurable"], "checkpoint_id": ""}}
+
+
+async def test_async_empty_checkpoint_id_reads_from_the_latest_checkpoint() -> None:
+    async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+        await saver.setup()
+        configs = await _abuild_chain(saver)
+        latest = await saver.aget_delta_channel_history(
+            config=configs[-1], channels=[CHANNEL]
+        )
+
+        result = await saver.aget_delta_channel_history(
+            config=_empty_checkpoint_id(configs[-1]), channels=[CHANNEL]
+        )
+
+    assert latest[CHANNEL]["writes"]
+    assert result == latest
+
+
+def test_sync_empty_checkpoint_id_reads_from_the_latest_checkpoint() -> None:
+    with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+        saver.setup()
+        configs = _build_chain(saver)
+        latest = saver.get_delta_channel_history(config=configs[-1], channels=[CHANNEL])
+
+        result = saver.get_delta_channel_history(
+            config=_empty_checkpoint_id(configs[-1]), channels=[CHANNEL]
+        )
+
+    assert latest[CHANNEL]["writes"]
+    assert result == latest
+
+
 async def test_root_target_has_no_history_and_still_terminates(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

libs/checkpoint-postgres/tests/test_prune.py

@@ -0,0 +1,540 @@
+import asyncio
+from collections.abc import Awaitable, Callable
+from contextlib import asynccontextmanager
+from typing import Any
+from uuid import uuid4
+
+import pytest
+from langgraph.checkpoint.base import empty_checkpoint
+from langgraph.checkpoint.base.id import uuid6
+from langgraph.checkpoint.conformance import validate
+from langgraph.checkpoint.conformance.initializer import checkpointer_test
+from langgraph.checkpoint.serde.types import TASKS, _DeltaSnapshot
+from psycopg import AsyncConnection
+
+from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
+from langgraph.checkpoint.postgres.base import (
+    PRUNE_DELTA_THREADS_SQL,
+    PRUNE_HEADS_SQL,
+    PRUNE_REFERENCED_BLOBS_SQL,
+)
+from tests.conftest import DEFAULT_POSTGRES_URI, DEFAULT_URI
+from tests.test_async import _saver as _async_saver
+from tests.test_sync import _saver as _sync_saver
+
+SAVERS = ["base", "pool", "pipe", "aio_base", "aio_pool", "aio_pipe"]
+
+
+@asynccontextmanager
+async def _saver(name: str):
+    if name.startswith("aio_"):
+        async with _async_saver(name.removeprefix("aio_")) as saver:
+            yield saver
+    else:
+        with _sync_saver(name) as saver:
+            yield saver
+
+
+async def _call(saver: Any, method: str, *args: Any, **kwargs: Any) -> Any:
+    if isinstance(saver, AsyncPostgresSaver):
+        return await getattr(saver, f"a{method}")(*args, **kwargs)
+    return getattr(saver, method)(*args, **kwargs)
+
+
+async def _rows(saver: Any, thread_id: str) -> dict[str, list[tuple]]:
+    queries = {
+        "checkpoints": "SELECT checkpoint_ns, checkpoint_id FROM checkpoints"
+        " WHERE thread_id = %s ORDER BY 1, 2",
+        "writes": "SELECT checkpoint_ns, checkpoint_id, task_id, idx"
+        " FROM checkpoint_writes WHERE thread_id = %s ORDER BY 1, 2, 3, 4",
+        "blobs": "SELECT checkpoint_ns, channel, version FROM checkpoint_blobs"
+        " WHERE thread_id = %s ORDER BY 1, 2, 3",
+    }
+    rows = {}
+    for table, sql in queries.items():
+        if isinstance(saver, AsyncPostgresSaver):
+            async with saver._cursor() as cur:
+                await cur.execute(sql, (thread_id,))
+                rows[table] = [tuple(r.values()) for r in await cur.fetchall()]
+        else:
+            with saver._cursor() as cur:
+                cur.execute(sql, (thread_id,))
+                rows[table] = [tuple(r.values()) for r in cur.fetchall()]
+    return rows
+
+
+async def _put(
+    saver: Any,
+    thread_id: str,
+    parent: dict | None,
+    step: int,
+    values: dict[str, Any],
+    versions: dict[str, str],
+    *,
+    ns: str = "",
+    metadata: dict[str, Any] | None = None,
+    v: int = 4,
+    checkpoint_id: str | None = None,
+) -> dict:
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ns}}
+    if parent is not None:
+        config["configurable"]["checkpoint_id"] = parent["configurable"][
+            "checkpoint_id"
+        ]
+    checkpoint = empty_checkpoint()
+    checkpoint.update(
+        v=v,
+        id=checkpoint_id or str(uuid6(clock_seq=step)),
+        channel_values=values,
+        channel_versions=versions,
+    )
+    new_versions = {k: versions[k] for k in values}
+    return await _call(
+        saver,
+        "put",
+        config,
+        checkpoint,
+        {"source": "loop", "step": step, "parents": {}, **(metadata or {})},
+        new_versions,
+    )
+
+
+async def _three_checkpoints(
+    saver: Any,
+    thread_id: str,
+    ns: str = "",
+    metadata: dict[str, Any] | None = None,
+) -> dict:
+    parent = None
+    steps = [
+        ({"a": [1], "s": ["shared"]}, {"a": "1", "s": "1"}),
+        ({"a": [1, 2]}, {"a": "2", "s": "1"}),
+        ({"a": [1, 2, 3]}, {"a": "3", "s": "1"}),
+    ]
+    for step, (values, versions) in enumerate(steps):
+        parent = await _put(
+            saver, thread_id, parent, step, values, versions, ns=ns, metadata=metadata
+        )
+        await _call(saver, "put_writes", parent, [("a", step)], str(uuid4()))
+    return parent
+
+
+@asynccontextmanager
+async def _second_saver(saver: AsyncPostgresSaver):
+    assert isinstance(saver.conn, AsyncConnection)
+    uri = DEFAULT_POSTGRES_URI + saver.conn.info.dbname
+    async with AsyncPostgresSaver.from_conn_string(uri) as other:
+        yield other
+
+
+def _run_once_after(
+    saver: AsyncPostgresSaver,
+    sql: str,
+    action: Callable[[], Awaitable[Any]],
+    *,
+    nth: int = 1,
+) -> None:
+    seen = 0
+    cursor = saver._cursor
+
+    class Hooked:
+        def __init__(self, cur: Any) -> None:
+            self.cur = cur
+
+        def __getattr__(self, name: str) -> Any:
+            return getattr(self.cur, name)
+
+        async def execute(self, query: Any, params: Any = None, **kwargs: Any) -> None:
+            nonlocal seen
+            await self.cur.execute(query, params, **kwargs)
+            if query is sql:
+                seen += 1
+                if seen == nth:
+                    await action()
+
+    @asynccontextmanager
+    async def hooked(**kwargs: Any):
+        async with cursor(**kwargs) as cur:
+            yield Hooked(cur)
+
+    saver._cursor = hooked
+
+
+async def _delta_thread(saver: Any, thread_id: str, evidence: str) -> None:
+    if evidence == "counters":
+        first = await _put(saver, thread_id, None, 0, {"m": [1]}, {"m": "1"})
+        await _put(
+            saver,
+            thread_id,
+            first,
+            1,
+            {},
+            {"m": "2"},
+            metadata={"counters_since_delta_snapshot": {"m": [1, 1]}},
+        )
+    elif evidence == "snapshot":
+        first = await _put(
+            saver, thread_id, None, 0, {"m": _DeltaSnapshot([1])}, {"m": "1"}
+        )
+        await _put(saver, thread_id, first, 1, {}, {"m": "2"})
+    else:
+        marker = {"delta_writes_versioned": True}
+        first = await _put(
+            saver, thread_id, None, 0, {"m": [1]}, {"m": "1"}, metadata=marker
+        )
+        await _put(
+            saver, thread_id, first, 1, {"m": [1, 2]}, {"m": "2"}, metadata=marker
+        )
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_keeps_each_namespace_head_with_its_writes_and_blobs(
+    name: str,
+) -> None:
+    async with _saver(name) as saver:
+        thread_id, other_id = str(uuid4()), str(uuid4())
+        heads = {
+            ns: await _three_checkpoints(saver, thread_id, ns) for ns in ("", "sub:1")
+        }
+        await _three_checkpoints(saver, other_id)
+        before = {ns: await _call(saver, "get_tuple", h) for ns, h in heads.items()}
+        other_before = await _rows(saver, other_id)
+
+        await _call(saver, "prune", [thread_id])
+
+        rows = await _rows(saver, thread_id)
+        assert rows["checkpoints"] == [
+            (ns, heads[ns]["configurable"]["checkpoint_id"]) for ns in ("", "sub:1")
+        ]
+        assert [(ns, cid) for ns, cid, *_ in rows["writes"]] == rows["checkpoints"]
+        assert rows["blobs"] == [
+            (ns, ch, ver)
+            for ns in ("", "sub:1")
+            for ch, ver in (("a", "3"), ("s", "1"))
+        ]
+        for ns, head in heads.items():
+            assert await _call(saver, "get_tuple", head) == before[ns]._replace(
+                parent_config=None
+            )
+        assert await _rows(saver, other_id) == other_before
+
+
+@pytest.mark.parametrize("evidence", ["counters", "snapshot", "versioned"])
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_refuses_a_batch_with_a_delta_channel_thread(
+    name: str, evidence: str
+) -> None:
+    async with _saver(name) as saver:
+        ordinary_id, delta_id = str(uuid4()), str(uuid4())
+        await _three_checkpoints(saver, ordinary_id)
+        await _delta_thread(saver, delta_id, evidence)
+        before = [await _rows(saver, t) for t in (ordinary_id, delta_id)]
+
+        with pytest.raises(NotImplementedError, match=delta_id):
+            await _call(saver, "prune", [ordinary_id, delta_id])
+
+        assert [await _rows(saver, t) for t in (ordinary_id, delta_id)] == before
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_prunes_a_thread_storing_a_plain_true(name: str) -> None:
+    async with _saver(name) as saver:
+        thread_id = str(uuid4())
+        first = await _put(
+            saver, thread_id, None, 0, {"flag": True, "a": [1]}, {"flag": "1", "a": "1"}
+        )
+        head = await _put(
+            saver, thread_id, first, 1, {"flag": True}, {"flag": "2", "a": "1"}
+        )
+
+        await _call(saver, "prune", [thread_id])
+
+        assert (await _rows(saver, thread_id))["checkpoints"] == [
+            ("", head["configurable"]["checkpoint_id"])
+        ]
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_keeps_the_parent_holding_a_pre_v4_heads_sends(
+    name: str,
+) -> None:
+    async with _saver(name) as saver:
+        thread_id = str(uuid4())
+        first = await _put(saver, thread_id, None, 0, {}, {}, v=1)
+        await _call(saver, "put_writes", first, [(TASKS, "send-1")], str(uuid4()))
+        second = await _put(saver, thread_id, first, 1, {}, {}, v=1)
+        head = await _put(saver, thread_id, second, 2, {}, {}, v=1)
+        await _call(saver, "put_writes", second, [(TASKS, "send-2")], str(uuid4()))
+
+        await _call(saver, "prune", [thread_id])
+
+        assert (await _rows(saver, thread_id))["checkpoints"] == [
+            ("", second["configurable"]["checkpoint_id"]),
+            ("", head["configurable"]["checkpoint_id"]),
+        ]
+        loaded = await _call(saver, "get_tuple", head)
+        assert loaded.checkpoint["channel_values"][TASKS] == ["send-2"]
+        assert loaded.parent_config == second
+        assert (await _call(saver, "get_tuple", second)).parent_config is None
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_delete_removes_every_row_of_the_threads(name: str) -> None:
+    async with _saver(name) as saver:
+        ordinary_id, delta_id, other_id = str(uuid4()), str(uuid4()), str(uuid4())
+        await _three_checkpoints(saver, ordinary_id)
+        await _delta_thread(saver, delta_id, "counters")
+        await _three_checkpoints(saver, other_id)
+        other_before = await _rows(saver, other_id)
+
+        await _call(saver, "prune", [ordinary_id, delta_id], strategy="delete")
+
+        empty = {"checkpoints": [], "writes": [], "blobs": []}
+        assert await _rows(saver, ordinary_id) == empty
+        assert await _rows(saver, delta_id) == empty
+        assert await _rows(saver, other_id) == other_before
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_unknown_strategy_changes_nothing(name: str) -> None:
+    async with _saver(name) as saver:
+        thread_id = str(uuid4())
+        await _three_checkpoints(saver, thread_id)
+        before = await _rows(saver, thread_id)
+
+        with pytest.raises(ValueError, match="keep_all"):
+            await _call(saver, "prune", [thread_id], strategy="keep_all")
+
+        assert await _rows(saver, thread_id) == before
+
+
+async def test_async_saver_prune_runs_from_another_thread_only() -> None:
+    async with _saver("aio_base") as saver:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+
+        with pytest.raises(asyncio.InvalidStateError):
+            saver.prune([thread_id])
+        await asyncio.to_thread(saver.prune, [thread_id])
+
+        assert (await _rows(saver, thread_id))["checkpoints"] == [
+            ("", head["configurable"]["checkpoint_id"])
+        ]
+
+
+async def test_prune_conformance() -> None:
+    @checkpointer_test(name="AsyncPostgresSaver")
+    async def postgres_saver():
+        async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            await saver.setup()
+            yield saver
+
+    report = await validate(postgres_saver, capabilities={"prune"})
+
+    prune = report.results["prune"]
+    if not prune.passed:
+        details = "\n".join(prune.failures)
+        pytest.fail(f"Capability prune failed (passed={prune.passed}):\n{details}")
+
+
+@pytest.mark.parametrize("strategy", ["keep_latest", "delete"])
+@pytest.mark.parametrize("name", ["base", "aio_base"])
+async def test_prune_rejects_a_single_thread_id_string(
+    name: str, strategy: str
+) -> None:
+    async with _saver(name) as saver:
+        for thread_id in ("ab", "a", "b"):
+            await _three_checkpoints(saver, thread_id)
+        before = [await _rows(saver, t) for t in ("ab", "a", "b")]
+
+        with pytest.raises(TypeError, match="not a str"):
+            await _call(saver, "prune", "ab", strategy=strategy)
+
+        assert [await _rows(saver, t) for t in ("ab", "a", "b")] == before
+
+
+@pytest.mark.parametrize(
+    "after", [PRUNE_HEADS_SQL, PRUNE_REFERENCED_BLOBS_SQL], ids=["heads", "blobs"]
+)
+async def test_keep_latest_keeps_a_checkpoint_saved_while_it_runs(after: str) -> None:
+    async with _saver("aio_base") as saver, _second_saver(saver) as other:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+        saved: list[dict] = []
+
+        async def save_a_child() -> None:
+            saved.append(
+                await _put(
+                    other, thread_id, head, 3, {"a": [1, 2, 3, 4]}, {"a": "4", "s": "1"}
+                )
+            )
+
+        _run_once_after(saver, after, save_a_child)
+        await saver.aprune([thread_id])
+
+        loaded = await saver.aget_tuple(saved[0])
+        assert loaded is not None, "the checkpoint saved during the prune was deleted"
+        assert loaded.checkpoint["channel_values"] == {
+            "a": [1, 2, 3, 4],
+            "s": ["shared"],
+        }
+        assert loaded.parent_config == head
+
+
+async def test_keep_latest_refuses_a_thread_that_turns_delta_after_the_batch_check() -> (
+    None
+):
+    async with _saver("aio_base") as saver, _second_saver(saver) as other:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+
+        async def save_a_delta_checkpoint() -> None:
+            await _put(
+                other,
+                thread_id,
+                head,
+                3,
+                {},
+                {"a": "4", "s": "1"},
+                metadata={"counters_since_delta_snapshot": {"a": [1, 1]}},
+            )
+
+        _run_once_after(saver, PRUNE_DELTA_THREADS_SQL, save_a_delta_checkpoint)
+        with pytest.raises(NotImplementedError, match=thread_id):
+            await saver.aprune([thread_id])
+
+        assert len((await _rows(saver, thread_id))["checkpoints"]) == 4
+
+
+async def test_keep_latest_keeps_the_history_of_a_delta_checkpoint_saved_after_the_thread_check() -> (
+    None
+):
+    async with _saver("aio_base") as saver, _second_saver(saver) as other:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+        saved: list[dict] = []
+
+        async def save_a_delta_checkpoint() -> None:
+            saved.append(
+                await _put(
+                    other,
+                    thread_id,
+                    head,
+                    3,
+                    {},
+                    {"a": "4", "s": "1"},
+                    metadata={"counters_since_delta_snapshot": {"a": [1, 1]}},
+                )
+            )
+
+        _run_once_after(saver, PRUNE_DELTA_THREADS_SQL, save_a_delta_checkpoint, nth=2)
+        await saver.aprune([thread_id])
+
+        assert (await _rows(saver, thread_id))["checkpoints"] == [
+            ("", head["configurable"]["checkpoint_id"]),
+            ("", saved[0]["configurable"]["checkpoint_id"]),
+        ]
+        history = await saver.aget_delta_channel_history(
+            config=saved[0], channels=["a"]
+        )
+        assert history["a"]["seed"] == [1, 2, 3]
+        assert [value for _, _, value in history["a"]["writes"]] == [2]
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_deletes_a_subgraph_call_with_the_checkpoint_it_started_from(
+    name: str,
+) -> None:
+    async with _saver(name) as saver:
+        thread_id = str(uuid4())
+        older = await _put(saver, thread_id, None, 0, {}, {})
+        head = await _put(saver, thread_id, older, 1, {}, {})
+        unsaved = str(uuid6(clock_seq=2))
+        started_from = {
+            "finished:1": older["configurable"]["checkpoint_id"],
+            "interrupted:2": head["configurable"]["checkpoint_id"],
+            "running:3": unsaved,
+            "checkpointer_true": older["configurable"]["checkpoint_id"],
+        }
+        for ns, root in started_from.items():
+            await _three_checkpoints(
+                saver, thread_id, ns, metadata={"parents": {"": root}}
+            )
+
+        await _call(saver, "prune", [thread_id])
+
+        rows = await _rows(saver, thread_id)
+        assert {ns for ns, *_ in rows["checkpoints"]} == {
+            "",
+            "interrupted:2",
+            "running:3",
+            "checkpointer_true",
+        }
+        assert {ns for ns, *_ in rows["writes"]} == {
+            "interrupted:2",
+            "running:3",
+            "checkpointer_true",
+        }
+        assert {ns for ns, *_ in rows["blobs"]} == {
+            "interrupted:2",
+            "running:3",
+            "checkpointer_true",
+        }
+
+
+async def test_keep_latest_keeps_a_checkpoint_saved_meanwhile_whatever_its_id() -> None:
+    async with _saver("aio_base") as saver, _second_saver(saver) as other:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+        lower_id = "0" + head["configurable"]["checkpoint_id"][1:]
+        saved: list[dict] = []
+
+        async def save_a_child_with_a_lower_id() -> None:
+            child = await _put(
+                other,
+                thread_id,
+                head,
+                3,
+                {"a": [1, 2, 3, 4]},
+                {"a": "4", "s": "1"},
+                checkpoint_id=lower_id,
+            )
+            await _call(other, "put_writes", child, [("a", 4)], str(uuid4()))
+            saved.append(child)
+
+        _run_once_after(saver, PRUNE_HEADS_SQL, save_a_child_with_a_lower_id)
+        await saver.aprune([thread_id])
+
+        loaded = await saver.aget_tuple(saved[0])
+        assert loaded is not None, "the checkpoint saved during the prune was deleted"
+        assert loaded.checkpoint["channel_values"] == {
+            "a": [1, 2, 3, 4],
+            "s": ["shared"],
+        }
+        assert [value for _, _, value in loaded.pending_writes or []] == [4]
+
+
+@pytest.mark.parametrize("name", SAVERS)
+async def test_keep_latest_keeps_writes_saved_before_their_checkpoint(
+    name: str,
+) -> None:
+    async with _saver(name) as saver:
+        thread_id = str(uuid4())
+        head = await _three_checkpoints(saver, thread_id)
+        next_id = str(uuid6(clock_seq=3))
+        next_config = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": "",
+                "checkpoint_id": next_id,
+            }
+        }
+        await _call(saver, "put_writes", next_config, [("a", 4)], str(uuid4()))
+
+        await _call(saver, "prune", [thread_id])
+        await _put(
+            saver, thread_id, head, 3, {}, {"a": "3", "s": "1"}, checkpoint_id=next_id
+        )
+
+        loaded = await _call(saver, "get_tuple", next_config)
+        assert [value for _, _, value in loaded.pending_writes] == [4]

libs/checkpoint-postgres/tests/test_store.py

@@ -11,6 +11,7 @@ import pytest
 from langchain_core.embeddings import Embeddings
 from langgraph.store.base import (
     GetOp,
+    InvalidNamespaceError,
     Item,
     ListNamespacesOp,
     MatchCondition,
@@ -1164,3 +1165,51 @@ def test_namespace_labels_with_trailing_newline(store) -> None:
     assert set(store.list_namespaces(prefix=["users", "alice"], limit=100)) == {
         ("users", "alice"),
     }
+
+
+@pytest.mark.parametrize("namespace", [("foo.bar",), ("foo", ""), ("foo", 1)])
+@pytest.mark.parametrize(
+    "kind", ["get", "put", "delete", "search", "list_prefix", "list_suffix"]
+)
+def test_batch_rejects_invalid_namespace_labels(
+    store, namespace: tuple, kind: str
+) -> None:
+    """Ops passed straight to `batch` must not reach another namespace.
+
+    Namespaces are stored dot-joined, so `("foo.bar",)` flattens to the same
+    text as `("foo", "bar")`. `BaseStore` methods validate labels themselves,
+    but `batch` takes ops as given.
+    """
+    op = {
+        "get": GetOp(namespace, "key"),
+        "put": PutOp(namespace, "key", {"changed": True}),
+        "delete": PutOp(namespace, "key", None),
+        "search": SearchOp(namespace),
+        "list_prefix": ListNamespacesOp((MatchCondition("prefix", namespace),)),
+        "list_suffix": ListNamespacesOp((MatchCondition("suffix", namespace),)),
+    }[kind]
+    store.put(("foo", "bar"), "key", {"original": True})
+
+    with pytest.raises(InvalidNamespaceError):
+        store.batch([PutOp(("valid",), "key", {}), op])
+
+    item = store.get(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}
+    # The whole batch is rejected before any SQL runs.
+    assert store.get(("valid",), "key") is None
+
+
+def test_batch_allows_empty_search_prefix_and_listing_wildcards(
+    store,
+) -> None:
+    store.put(("foo", "bar"), "key", {"v": 1})
+
+    found, listed = store.batch(
+        [
+            SearchOp(()),
+            ListNamespacesOp((MatchCondition("prefix", ("foo", "*")),)),
+        ]
+    )
+
+    assert [item.namespace for item in found] == [("foo", "bar")]
+    assert listed == [("foo", "bar")]

libs/checkpoint-postgres/uv.lock

@@ -353,7 +353,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.2"
+version = "3.2.0"
 source = { editable = "." }
 dependencies = [
     { name = "langgraph-checkpoint" },

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

@@ -539,7 +539,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         thread_id = str(config["configurable"]["thread_id"])
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
-        if checkpoint_id is None:
+        if not checkpoint_id:
             target = self.get_tuple(config)
             if target is None:
                 return {ch: {"writes": []} for ch in channels}

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

@@ -653,7 +653,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         thread_id = str(config["configurable"]["thread_id"])
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
-        if checkpoint_id is None:
+        if not checkpoint_id:
             target = await self.aget_tuple(config)
             if target is None:
                 return {ch: {"writes": []} for ch in channels}

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

@@ -28,6 +28,7 @@ from langgraph.store.base import (
     ensure_embeddings,
     get_text_at_path,
     tokenize_path,
+    validate_op_namespace,
 )
 
 _AIO_ERROR_MSG = (
@@ -257,6 +258,7 @@ def _group_ops(ops: Iterable[Op]) -> tuple[dict[type, list[tuple[int, Op]]], int
     grouped_ops: dict[type, list[tuple[int, Op]]] = defaultdict(list)
     tot = 0
     for idx, op in enumerate(ops):
+        validate_op_namespace(op)
         grouped_ops[type(op)].append((idx, op))
         tot += 1
     return grouped_ops, tot

libs/checkpoint-sqlite/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.1"
+version = "3.1.2"
 description = "Library with a SQLite implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"

libs/checkpoint-sqlite/tests/test_async_store.py

@@ -9,8 +9,10 @@ from typing import cast
 import pytest
 from langgraph.store.base import (
     GetOp,
+    InvalidNamespaceError,
     Item,
     ListNamespacesOp,
+    MatchCondition,
     PutOp,
     SearchOp,
 )
@@ -745,3 +747,65 @@ async def test_async_namespace_segment_boundary(store: AsyncSqliteStore) -> None
     assert set(await store.alist_namespaces(suffix=["alice"], limit=100)) == {
         ("uid", "users", "alice"),
     }
+
+
+@pytest.mark.parametrize("namespace", [("foo.bar",), ("foo", ""), ("foo", 1)])
+async def test_abatch_rejects_invalid_namespace_labels(
+    store: AsyncSqliteStore, namespace: tuple
+) -> None:
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    for op in (
+        GetOp(namespace, "key"),
+        GetOp(namespace, "key", refresh_ttl=True),
+        PutOp(namespace, "key", {"changed": True}),
+        PutOp(namespace, "key", None),
+        SearchOp(namespace),
+        ListNamespacesOp((MatchCondition("prefix", namespace),)),
+        ListNamespacesOp((MatchCondition("suffix", namespace),)),
+    ):
+        with pytest.raises(InvalidNamespaceError):
+            await store.abatch([op])
+
+    item = await store.aget(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}
+
+
+async def test_invalid_namespace_only_fails_its_own_call(
+    store: AsyncSqliteStore,
+) -> None:
+    """Concurrent calls share one `abatch`, which fails every op if it raises.
+
+    Labels are checked before an op is queued, so one caller's bad label cannot
+    fail another caller's request.
+    """
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    valid, invalid = await asyncio.gather(
+        store.aget(("foo", "bar"), "key"),
+        store.aget(("foo.bar",), "key"),
+        return_exceptions=True,
+    )
+
+    assert isinstance(valid, Item) and valid.value == {"original": True}
+    assert isinstance(invalid, InvalidNamespaceError)
+
+
+async def test_sync_methods_reject_invalid_namespace_labels(
+    store: AsyncSqliteStore,
+) -> None:
+    """The sync wrappers run off the event loop thread and must validate too."""
+    await store.aput(("foo", "bar"), "key", {"original": True})
+
+    for call in (
+        lambda: store.get(("foo.bar",), "key"),
+        lambda: store.search(("foo.bar",)),
+        lambda: store.delete(("foo.bar",), "key"),
+        lambda: store.list_namespaces(prefix=("foo.bar",)),
+        lambda: store.batch([GetOp(("foo.bar",), "key")]),
+    ):
+        with pytest.raises(InvalidNamespaceError):
+            await asyncio.to_thread(call)
+
+    item = await store.aget(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}

libs/checkpoint-sqlite/tests/test_delta_parent_walk.py

@@ -65,6 +65,39 @@ async def test_async_walk_reaches_parent_whatever_the_id_order(
         assert got[CHANNEL] == EXPECTED
 
 
+EMPTY_CHECKPOINT_ID: dict[str, Any] = {
+    "configurable": {**CONFIG["configurable"], "checkpoint_id": ""}
+}
+
+
+def test_sync_empty_checkpoint_id_reads_from_the_latest_checkpoint() -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        root = saver.put(CONFIG, _checkpoint("a-older", {CHANNEL: "seed"}), {}, {})
+        saver.put_writes(root, [(CHANNEL, "write-root")], "task")
+        saver.put(root, _checkpoint("z-newer", {}), {}, {})
+
+        got = saver.get_delta_channel_history(
+            config=EMPTY_CHECKPOINT_ID, channels=[CHANNEL]
+        )
+
+        assert got[CHANNEL] == EXPECTED
+
+
+async def test_async_empty_checkpoint_id_reads_from_the_latest_checkpoint() -> None:
+    async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+        root = await saver.aput(
+            CONFIG, _checkpoint("a-older", {CHANNEL: "seed"}), {}, {}
+        )
+        await saver.aput_writes(root, [(CHANNEL, "write-root")], "task")
+        await saver.aput(root, _checkpoint("z-newer", {}), {}, {})
+
+        got = await saver.aget_delta_channel_history(
+            config=EMPTY_CHECKPOINT_ID, channels=[CHANNEL]
+        )
+
+        assert got[CHANNEL] == EXPECTED
+
+
 def test_walk_reaches_root_of_long_chain_with_descending_ids() -> None:
     steps = 40
     with SqliteSaver.from_conn_string(":memory:") as saver:

libs/checkpoint-sqlite/tests/test_store.py

@@ -14,6 +14,7 @@ import pytest
 from langchain_core.embeddings import Embeddings
 from langgraph.store.base import (
     GetOp,
+    InvalidNamespaceError,
     Item,
     ListNamespacesOp,
     MatchCondition,
@@ -1435,3 +1436,51 @@ def test_list_namespaces_metacharacter_labels(store: SqliteStore) -> None:
         assert set(store.list_namespaces(prefix=[label, "child"], limit=100)) == {
             (label, "child"),
         }
+
+
+@pytest.mark.parametrize("namespace", [("foo.bar",), ("foo", ""), ("foo", 1)])
+@pytest.mark.parametrize(
+    "kind", ["get", "put", "delete", "search", "list_prefix", "list_suffix"]
+)
+def test_batch_rejects_invalid_namespace_labels(
+    store: SqliteStore, namespace: tuple, kind: str
+) -> None:
+    """Ops passed straight to `batch` must not reach another namespace.
+
+    Namespaces are stored dot-joined, so `("foo.bar",)` flattens to the same
+    text as `("foo", "bar")`. `BaseStore` methods validate labels themselves,
+    but `batch` takes ops as given.
+    """
+    op = {
+        "get": GetOp(namespace, "key"),
+        "put": PutOp(namespace, "key", {"changed": True}),
+        "delete": PutOp(namespace, "key", None),
+        "search": SearchOp(namespace),
+        "list_prefix": ListNamespacesOp((MatchCondition("prefix", namespace),)),
+        "list_suffix": ListNamespacesOp((MatchCondition("suffix", namespace),)),
+    }[kind]
+    store.put(("foo", "bar"), "key", {"original": True})
+
+    with pytest.raises(InvalidNamespaceError):
+        store.batch([PutOp(("valid",), "key", {}), op])
+
+    item = store.get(("foo", "bar"), "key")
+    assert item is not None and item.value == {"original": True}
+    # The whole batch is rejected before any SQL runs.
+    assert store.get(("valid",), "key") is None
+
+
+def test_batch_allows_empty_search_prefix_and_listing_wildcards(
+    store: SqliteStore,
+) -> None:
+    store.put(("foo", "bar"), "key", {"v": 1})
+
+    found, listed = store.batch(
+        [
+            SearchOp(()),
+            ListNamespacesOp((MatchCondition("prefix", ("foo", "*")),)),
+        ]
+    )
+
+    assert [item.namespace for item in found] == [("foo", "bar")]
+    assert listed == [("foo", "bar")]

libs/checkpoint-sqlite/uv.lock

@@ -360,7 +360,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.1"
+version = "3.1.2"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },

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

@@ -170,8 +170,8 @@ class InMemorySaver(
             return {}
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-        checkpoint_id = config["configurable"].get("checkpoint_id", "")
         ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
+        checkpoint_id = get_checkpoint_id(config) or max(ns_storage, default="")
 
         chain: list[str] = []
         target_entry = ns_storage.get(checkpoint_id)

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

@@ -771,7 +771,12 @@ class BaseStore(ABC):
 
         Returns:
             The retrieved item or `None` if not found.
+
+        Raises:
+            InvalidNamespaceError: If a namespace label is empty, is not a string,
+                or contains a period (`.`).
         """
+        _validate_namespace_labels(namespace)
         return self.batch(
             [GetOp(namespace, str(key), _ensure_refresh(self.ttl_config, refresh_ttl))]
         )[0]
@@ -801,6 +806,10 @@ class BaseStore(ABC):
         Returns:
             List of items matching the search criteria.
 
+        Raises:
+            InvalidNamespaceError: If a `namespace_prefix` label is empty, is not a
+                string, or contains a period (`.`).
+
         ???+ example "Examples"
 
             Basic filtering:
@@ -840,6 +849,7 @@ class BaseStore(ABC):
                 Natural language search support depends on your store implementation
                 and requires proper embedding configuration.
         """
+        _validate_namespace_labels(namespace_prefix)
         return self.batch(
             [
                 SearchOp(
@@ -887,6 +897,11 @@ class BaseStore(ABC):
                 By default, the expiration timer refreshes on both read operations (get/search)
                 and write operations (put/update), whenever the item is included in the operation.
 
+        Raises:
+            InvalidNamespaceError: If the namespace is empty, its root label is
+                `"langgraph"`, or a label is empty, is not a string, or contains a
+                period (`.`).
+
         Note:
             Indexing support depends on your store implementation.
             If you do not initialize the store with indexing capabilities,
@@ -940,7 +955,12 @@ class BaseStore(ABC):
         Args:
             namespace: Hierarchical path for the item.
             key: Unique identifier within the namespace.
+
+        Raises:
+            InvalidNamespaceError: If a namespace label is empty, is not a string,
+                or contains a period (`.`).
         """
+        _validate_namespace_labels(namespace)
         self.batch([PutOp(namespace, str(key), None, ttl=None)])
 
     def list_namespaces(
@@ -969,6 +989,10 @@ class BaseStore(ABC):
             A list of namespace tuples that match the criteria. Each tuple represents a
                 full namespace path up to `max_depth`.
 
+        Raises:
+            InvalidNamespaceError: If a `prefix` or `suffix` label is empty, is not a
+                string, or contains a period (`.`).
+
         ???+ example "Examples":
 
             Setting `max_depth=3`. Given the namespaces:
@@ -984,6 +1008,8 @@ class BaseStore(ABC):
             # [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
             ```
         """
+        _validate_namespace_labels(prefix or ())
+        _validate_namespace_labels(suffix or ())
         match_conditions = []
         if prefix:
             match_conditions.append(MatchCondition(match_type="prefix", path=prefix))
@@ -1013,7 +1039,12 @@ class BaseStore(ABC):
 
         Returns:
             The retrieved item or `None` if not found.
+
+        Raises:
+            InvalidNamespaceError: If a namespace label is empty, is not a string,
+                or contains a period (`.`).
         """
+        _validate_namespace_labels(namespace)
         return (
             await self.abatch(
                 [
@@ -1052,6 +1083,10 @@ class BaseStore(ABC):
         Returns:
             List of items matching the search criteria.
 
+        Raises:
+            InvalidNamespaceError: If a `namespace_prefix` label is empty, is not a
+                string, or contains a period (`.`).
+
         ???+ example "Examples"
 
             Basic filtering:
@@ -1091,6 +1126,7 @@ class BaseStore(ABC):
                 Natural language search support depends on your store implementation
                 and requires proper embedding configuration.
         """
+        _validate_namespace_labels(namespace_prefix)
         return (
             await self.abatch(
                 [
@@ -1140,6 +1176,11 @@ class BaseStore(ABC):
                 By default, the expiration timer refreshes on both read operations (get/search)
                 and write operations (put/update), whenever the item is included in the operation.
 
+        Raises:
+            InvalidNamespaceError: If the namespace is empty, its root label is
+                `"langgraph"`, or a label is empty, is not a string, or contains a
+                period (`.`).
+
         Note:
             Indexing support depends on your store implementation.
             If you do not initialize the store with indexing capabilities,
@@ -1201,7 +1242,12 @@ class BaseStore(ABC):
         Args:
             namespace: Hierarchical path for the item.
             key: Unique identifier within the namespace.
+
+        Raises:
+            InvalidNamespaceError: If a namespace label is empty, is not a string,
+                or contains a period (`.`).
         """
+        _validate_namespace_labels(namespace)
         await self.abatch([PutOp(namespace, str(key), None)])
 
     async def alist_namespaces(
@@ -1230,6 +1276,10 @@ class BaseStore(ABC):
             A list of namespace tuples that match the criteria. Each tuple represents a
                 full namespace path up to `max_depth`.
 
+        Raises:
+            InvalidNamespaceError: If a `prefix` or `suffix` label is empty, is not a
+                string, or contains a period (`.`).
+
         ???+ example "Examples"
 
             Setting `max_depth=3` with existing namespaces:
@@ -1245,6 +1295,8 @@ class BaseStore(ABC):
             # Returns: [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
             ```
         """
+        _validate_namespace_labels(prefix or ())
+        _validate_namespace_labels(suffix or ())
         match_conditions = []
         if prefix:
             match_conditions.append(MatchCondition(match_type="prefix", path=prefix))
@@ -1263,6 +1315,14 @@ class BaseStore(ABC):
 def _validate_namespace(namespace: tuple[str, ...]) -> None:
     if not namespace:
         raise InvalidNamespaceError("Namespace cannot be empty.")
+    _validate_namespace_labels(namespace)
+    if namespace[0] == "langgraph":
+        raise InvalidNamespaceError(
+            f'Root label for namespace cannot be "langgraph". Got: {namespace}'
+        )
+
+
+def _validate_namespace_labels(namespace: tuple[str, ...]) -> None:
     for label in namespace:
         if not isinstance(label, str):
             raise InvalidNamespaceError(
@@ -1277,10 +1337,27 @@ def _validate_namespace(namespace: tuple[str, ...]) -> None:
             raise InvalidNamespaceError(
                 f"Namespace labels cannot be empty strings. Got {label} in {namespace}"
             )
-    if namespace[0] == "langgraph":
-        raise InvalidNamespaceError(
-            f'Root label for namespace cannot be "langgraph". Got: {namespace}'
-        )
+
+
+def validate_op_namespace(op: Op) -> None:
+    """Validate the namespace labels an op carries before a store executes it.
+
+    `BaseStore` methods check labels before batching, but ops passed directly to
+    `batch`/`abatch` skip those methods. Stores that serialize namespaces as
+    delimited text should call this for every op they execute, so a label such
+    as `"foo.bar"` cannot address the namespace `("foo", "bar")`.
+
+    Raises:
+        InvalidNamespaceError: If a label is empty, is not a string, or contains
+            a period (`.`).
+    """
+    if isinstance(op, (GetOp, PutOp)):
+        _validate_namespace_labels(op.namespace)
+    elif isinstance(op, SearchOp):
+        _validate_namespace_labels(op.namespace_prefix)
+    elif isinstance(op, ListNamespacesOp):
+        for condition in op.match_conditions or ():
+            _validate_namespace_labels(condition.path)
 
 
 def _ensure_refresh(
@@ -1319,4 +1396,5 @@ __all__ = [
     "ensure_embeddings",
     "tokenize_path",
     "get_text_at_path",
+    "validate_op_namespace",
 ]

libs/checkpoint/langgraph/store/base/batch.py

@@ -25,6 +25,7 @@ from langgraph.store.base import (
     _ensure_refresh,
     _ensure_ttl,
     _validate_namespace,
+    _validate_namespace_labels,
 )
 
 F = TypeVar("F", bound=Callable)
@@ -86,6 +87,7 @@ class AsyncBatchedBaseStore(BaseStore):
         *,
         refresh_ttl: bool | None = None,
     ) -> Item | None:
+        _validate_namespace_labels(namespace)
         self._ensure_task()
         fut = self._loop.create_future()
         self._aqueue.put_nowait(
@@ -111,6 +113,7 @@ class AsyncBatchedBaseStore(BaseStore):
         offset: int = 0,
         refresh_ttl: bool | None = None,
     ) -> list[SearchItem]:
+        _validate_namespace_labels(namespace_prefix)
         self._ensure_task()
         fut = self._loop.create_future()
         self._aqueue.put_nowait(
@@ -155,6 +158,7 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace: tuple[str, ...],
         key: str,
     ) -> None:
+        _validate_namespace_labels(namespace)
         self._ensure_task()
         fut = self._loop.create_future()
         self._aqueue.put_nowait((fut, PutOp(namespace, key, None)))
@@ -169,6 +173,8 @@ class AsyncBatchedBaseStore(BaseStore):
         limit: int = 100,
         offset: int = 0,
     ) -> list[tuple[str, ...]]:
+        _validate_namespace_labels(prefix or ())
+        _validate_namespace_labels(suffix or ())
         self._ensure_task()
         fut = self._loop.create_future()
         match_conditions = []

libs/checkpoint/tests/test_conformance_delta.py

@@ -1,37 +0,0 @@
-"""Run delta-channel conformance capabilities against InMemorySaver."""
-
-from __future__ import annotations
-
-import pytest
-
-conformance = pytest.importorskip(
-    "langgraph.checkpoint.conformance",
-    reason="langgraph-checkpoint-conformance not installed",
-)
-
-
-@pytest.mark.asyncio
-async def test_delta_channel_conformance():
-    # Imported inside the test: the module-level importorskip above is what
-    # makes these safe, so they cannot move to the top of the file.
-    from langgraph.checkpoint.conformance import validate  # noqa: PLC0415
-    from langgraph.checkpoint.conformance.initializer import (  # noqa: PLC0415
-        checkpointer_test,
-    )
-
-    from langgraph.checkpoint.memory import InMemorySaver  # noqa: PLC0415
-
-    @checkpointer_test(name="InMemorySaver")
-    async def mem_saver():
-        yield InMemorySaver()
-
-    report = await validate(
-        mem_saver,
-        capabilities={
-            "delta_channel_history",
-        },
-    )
-    for cap, result in report.results.items():
-        if result.passed is False:
-            details = "\n".join(result.failures or [])
-            pytest.fail(f"Capability {cap} failed:\n{details}")

libs/checkpoint/tests/test_memory.py

@@ -420,6 +420,19 @@ class TestInMemorySaverDeltaChannel:
         assert "seed" not in result
         assert result["writes"] == []
 
+    def test_get_channel_writes_without_checkpoint_id_reads_the_latest(self) -> None:
+        saver = InMemorySaver()
+        thread: RunnableConfig = {
+            "configurable": {"thread_id": "t1", "checkpoint_ns": ""}
+        }
+        parent = saver.put(thread, empty_checkpoint(), {}, {})
+        saver.put_writes(parent, [("messages", "hi")], "task1")
+        saver.put(parent, empty_checkpoint(), {}, {})
+
+        result = saver.get_delta_channel_history(config=thread, channels=["messages"])
+
+        assert result == {"messages": {"writes": [("task1", "messages", "hi")]}}
+
 
 class TestBaseFallbackGetChannelWrites:
     """Exercises the `BaseCheckpointSaver.get_delta_channel_history` default

libs/checkpoint/tests/test_store.py

@@ -13,10 +13,14 @@ from langgraph.store.base import (
     GetOp,
     InvalidNamespaceError,
     Item,
+    ListNamespacesOp,
+    MatchCondition,
     Op,
     PutOp,
     Result,
+    SearchOp,
     get_text_at_path,
+    validate_op_namespace,
 )
 from langgraph.store.base.batch import AsyncBatchedBaseStore
 from langgraph.store.memory import InMemoryStore
@@ -528,6 +532,127 @@ async def test_cannot_put_empty_namespace() -> None:
     assert (await async_store.aget(("valid", "namespace"), "key")) is None
 
 
+INVALID_NAMESPACES = [("foo.bar",), ("foo", ""), (123,)]
+NAMESPACE_METHODS = ["get", "delete", "search", "prefix", "suffix"]
+
+
+@pytest.mark.parametrize("namespace", INVALID_NAMESPACES)
+@pytest.mark.parametrize("method", NAMESPACE_METHODS)
+def test_rejects_invalid_namespace_labels(
+    mocker: MockerFixture, namespace: tuple, method: str
+) -> None:
+    store = InMemoryStore()
+    batch = mocker.spy(InMemoryStore, "batch")
+    call = {
+        "get": lambda: store.get(namespace, "key"),
+        "delete": lambda: store.delete(namespace, "key"),
+        "search": lambda: store.search(namespace),
+        "prefix": lambda: store.list_namespaces(prefix=namespace),
+        "suffix": lambda: store.list_namespaces(suffix=namespace),
+    }[method]
+
+    with pytest.raises(InvalidNamespaceError):
+        call()
+
+    batch.assert_not_called()
+
+
+@pytest.mark.parametrize("batched", [False, True])
+@pytest.mark.parametrize("namespace", INVALID_NAMESPACES)
+@pytest.mark.parametrize("method", NAMESPACE_METHODS)
+async def test_async_rejects_invalid_namespace_labels(
+    mocker: MockerFixture, batched: bool, namespace: tuple, method: str
+) -> None:
+    # The batched store must reject before queueing: a failure inside the
+    # shared `abatch` would fail every op queued alongside this one.
+    store = MockAsyncBatchedStore() if batched else InMemoryStore()
+    # `MockAsyncBatchedStore` dispatches through `InMemoryStore.batch`.
+    batch = mocker.spy(InMemoryStore, "batch")
+    abatch = mocker.spy(InMemoryStore, "abatch")
+    call = {
+        "get": lambda: store.aget(namespace, "key"),
+        "delete": lambda: store.adelete(namespace, "key"),
+        "search": lambda: store.asearch(namespace),
+        "prefix": lambda: store.alist_namespaces(prefix=namespace),
+        "suffix": lambda: store.alist_namespaces(suffix=namespace),
+    }[method]
+
+    with pytest.raises(InvalidNamespaceError):
+        await call()
+
+    batch.assert_not_called()
+    abatch.assert_not_called()
+
+
+def test_search_and_listing_keep_empty_prefixes_and_wildcards() -> None:
+    store = InMemoryStore()
+    store.put(("tenant", "a_%"), "key", {"v": 1})
+    store.put(("tenant", "b", "child"), "key", {"v": 1})
+
+    assert len(store.search(())) == 2
+    assert [item.namespace for item in store.search(("tenant", "a_%"))] == [
+        ("tenant", "a_%")
+    ]
+    assert sorted(store.list_namespaces(prefix=("tenant", "*"), suffix=("*",))) == [
+        ("tenant", "a_%"),
+        ("tenant", "b", "child"),
+    ]
+
+
+@pytest.mark.parametrize("batched", [False, True])
+async def test_async_search_and_listing_keep_empty_prefixes_and_wildcards(
+    batched: bool,
+) -> None:
+    store = MockAsyncBatchedStore() if batched else InMemoryStore()
+    await store.aput(("tenant", "a_%"), "key", {"v": 1})
+    await store.aput(("tenant", "b", "child"), "key", {"v": 1})
+
+    assert len(await store.asearch(())) == 2
+    assert [item.namespace for item in await store.asearch(("tenant", "a_%"))] == [
+        ("tenant", "a_%")
+    ]
+    assert sorted(
+        await store.alist_namespaces(prefix=("tenant", "*"), suffix=("*",))
+    ) == [("tenant", "a_%"), ("tenant", "b", "child")]
+
+
+@pytest.mark.parametrize("namespace", INVALID_NAMESPACES)
+@pytest.mark.parametrize(
+    "kind", ["get", "put", "delete", "search", "list_prefix", "list_suffix"]
+)
+def test_validate_op_namespace_rejects_invalid_labels(
+    namespace: tuple, kind: str
+) -> None:
+    op = {
+        "get": GetOp(namespace, "key"),
+        "put": PutOp(namespace, "key", {"v": 1}),
+        "delete": PutOp(namespace, "key", None),
+        "search": SearchOp(namespace),
+        "list_prefix": ListNamespacesOp((MatchCondition("prefix", namespace),)),
+        "list_suffix": ListNamespacesOp((MatchCondition("suffix", namespace),)),
+    }[kind]
+
+    with pytest.raises(InvalidNamespaceError):
+        validate_op_namespace(op)
+
+
+def test_validate_op_namespace_allows_empty_prefix_and_wildcards() -> None:
+    for op in (
+        SearchOp(()),
+        ListNamespacesOp(),
+        ListNamespacesOp(
+            (
+                MatchCondition("prefix", ("tenant", "*")),
+                MatchCondition("suffix", ("*",)),
+            )
+        ),
+        GetOp(("tenant", "a_%"), "key"),
+        # Write-only rules belong to `put`, not to op validation.
+        PutOp(("langgraph", "x"), "key", {"v": 1}),
+    ):
+        validate_op_namespace(op)
+
+
 async def test_async_batch_store_deduplication(mocker: MockerFixture) -> None:
     abatch = mocker.spy(InMemoryStore, "batch")
     store = MockAsyncBatchedStore()

libs/cli/js-examples/yarn.lock

@@ -2829,11 +2829,11 @@ graceful-fs@^4.2.11:
   integrity sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==
 
 handlebars@^4.7.9:
-  version "4.7.9"
-  resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.9.tgz#6f139082ab58dc4e5a0e51efe7db5ae890d56a0f"
-  integrity sha512-4E71E0rpOaQuJR2A3xDZ+GM1HyWYv1clR58tC8emQNeQe3RH7MAzSbat+V0wG78LQBo6m6bzSG/L4pBuCsgnUQ==
+  version "4.7.10"
+  resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.10.tgz#12419ea1b41391b3ffcf003f2164dcbce090c417"
+  integrity sha512-P5VJMVM7qgBn6vjXMw8WG9uVI+ncf2pi72j4de4yz5ZULLj2RGqLYaKOYGsgyrViQ0tePOVlN1tDCCXXtFqXKg==
   dependencies:
-    minimist "^1.2.5"
+    minimist "^1.2.8"
     neo-async "^2.6.2"
     source-map "^0.6.1"
     wordwrap "^1.0.0"
@@ -3920,7 +3920,7 @@ minimatch@^9.0.4:
   dependencies:
     brace-expansion "^2.0.2"
 
-minimist@^1.2.0, minimist@^1.2.5, minimist@^1.2.6:
+minimist@^1.2.0, minimist@^1.2.6, minimist@^1.2.8:
   version "1.2.8"
   resolved "https://registry.yarnpkg.com/minimist/-/minimist-1.2.8.tgz#c1a464e7693302e082a075cee0c057741ac4772c"
   integrity sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.33"
+__version__ = "0.4.36"

libs/cli/langgraph_cli/deploy.py

@@ -231,15 +231,6 @@ class RequestedPlacement:
     def requested(self) -> bool:
         return self.listener_id is not None or self.k8s_namespace is not None
 
-    def ensure_not_requested(self, deployment_id: str) -> None:
-        if self.requested:
-            raise click.UsageError(
-                "Listener and namespace are fixed when a deployment is created. "
-                f"Deployment {deployment_id} already exists, so drop --listener-id "
-                "and --k8s-namespace, or create a new deployment with a different "
-                "--name."
-            )
-
     def on(self, listener: Listener) -> Placement:
         return OnListener(listener.id, self._namespace(listener))
 
@@ -1055,15 +1046,28 @@ def _print_deployment_result(
 # ---------------------------------------------------------------------------
 
 
+def _strip_url_scheme(url: str) -> str:
+    """Return a registry reference without a trailing slash or URL scheme."""
+    reference = url.rstrip("/")
+    if "://" in reference:
+        reference = reference.split("//", 1)[1]
+    return reference
+
+
+# Artifact Registry's username for access tokens. Control planes whose registry
+# needs another one return it as `username`.
+_DEFAULT_REGISTRY_USERNAME = "oauth2accesstoken"
+
+
 @contextmanager
-def _docker_config_for_token(registry_host: str, token: str):
+def _docker_config_for_token(registry_host: str, token: str, username: str):
     """Create a temporary Docker config with only the push token.
 
     Yields the path to a temporary config directory that can be passed
     to `docker --config <path>` so that system credential helpers
     (e.g. gcloud) don't interfere with the push token.
     """
-    auth_b64 = base64.b64encode(f"oauth2accesstoken:{token}".encode()).decode()
+    auth_b64 = base64.b64encode(f"{username}:{token}".encode()).decode()
     config_data = {"auths": {registry_host: {"auth": auth_b64}}}
     with tempfile.TemporaryDirectory() as tmpdir:
         with open(os.path.join(tmpdir, "config.json"), "w") as f:
@@ -1201,10 +1205,37 @@ class DockerBuildCommand:
             return cls(("docker", "build"), ())
         flags: tuple[str, ...] = ("--platform", _DEPLOYMENT_PLATFORM, "--load")
         if not verbose:
-            flags += ("--progress=quiet",)
+            flags += ("--progress=plain",)
         return cls(("docker", "buildx", "build"), flags)
 
 
+def _active_docker_host(runner: CommandRunner) -> str | None:
+    """Return the daemon endpoint of the user's active Docker context.
+
+    None when `DOCKER_HOST` is set, which Docker honors with any config
+    directory. Also None when the lookup fails (for example, `DOCKER_CONTEXT`
+    names a context that doesn't exist) or the context has no Docker endpoint
+    (Docker prints `<no value>`); that keeps Docker's own default in place.
+    """
+    if os.environ.get("DOCKER_HOST"):
+        return None
+    try:
+        stdout, _ = runner.run(
+            subp_exec(
+                "docker",
+                "context",
+                "inspect",
+                "--format",
+                "{{.Endpoints.docker.Host}}",
+                collect=True,
+            )
+        )
+    except (click.exceptions.Exit, OSError):
+        return None
+    host = (stdout or "").strip()
+    return host if "://" in host else None
+
+
 def _docker_argv(docker_config_dir: str | None) -> tuple[str, ...]:
     if docker_config_dir is None:
         return ("docker",)
@@ -1238,17 +1269,13 @@ def _push_image(
     runner: CommandRunner,
     image: str,
     *,
-    docker_config_dir: str | None,
+    docker: tuple[str, ...],
     verbose: bool,
 ) -> None:
     for attempt in range(1, _PUSH_ATTEMPTS + 1):
         try:
             with Progress(message="Pushing...", elapsed=not verbose):
-                runner.run(
-                    subp_exec(
-                        *_docker_argv(docker_config_dir), "push", image, verbose=verbose
-                    )
-                )
+                runner.run(subp_exec(*docker, "push", image, verbose=verbose))
             return
         except click.exceptions.Exit:
             if attempt == _PUSH_ATTEMPTS:
@@ -1317,21 +1344,27 @@ def _run_local_build(
             raise click.ClickException(
                 "Push token response missing token or registry_url"
             )
+        username = push_data.get("username") or _DEFAULT_REGISTRY_USERNAME
         step += 1
 
-        normalized_registry = registry_url.rstrip("/")
-        if "://" in normalized_registry:
-            normalized_registry = normalized_registry.split("//", 1)[1]
-        repo_seed = image_name or name or spec.config.parent.name
-        remote_image = str(
-            ImageReference(
-                f"{normalized_registry}/{normalize_name(repo_seed)}",
-                tag,
+        if push_data.get("repository_url"):
+            # The control plane chose the exact repository (an ECR repository
+            # is a single image name, so nothing may be appended to it).
+            repository = _strip_url_scheme(push_data["repository_url"])
+        else:
+            repo_seed = image_name or name or spec.config.parent.name
+            repository = (
+                f"{_strip_url_scheme(registry_url)}/{normalize_name(repo_seed)}"
             )
-        )
-        registry_host = normalized_registry.split("/")[0]
+        remote_image = str(ImageReference(repository, tag))
+        registry_host = repository.split("/")[0]
 
-        with _docker_config_for_token(registry_host, deployment_token) as cfg:
+        with _docker_config_for_token(registry_host, deployment_token, username) as cfg:
+            # A config directory has no current context, so Docker would fall
+            # back to /var/run/docker.sock; keep using the daemon the build used.
+            docker_host = _active_docker_host(runner)
+            host_args = ("--host", docker_host) if docker_host else ()
+            docker = (*_docker_argv(cfg), *host_args)
             _log_deploy_step(step, f"Logging into {registry_host}")
             token_input = (
                 deployment_token
@@ -1340,10 +1373,10 @@ def _run_local_build(
             )
             runner.run(
                 subp_exec(
-                    *_docker_argv(cfg),
+                    *docker,
                     "login",
                     "-u",
-                    "oauth2accesstoken",
+                    username,
                     "--password-stdin",
                     registry_host,
                     input=token_input,
@@ -1356,7 +1389,7 @@ def _run_local_build(
             runner.run(
                 subp_exec("docker", "tag", image_to_push, remote_image, verbose=verbose)
             )
-            _push_image(runner, remote_image, docker_config_dir=cfg, verbose=verbose)
+            _push_image(runner, remote_image, docker=docker, verbose=verbose)
         step += 1
 
         resolved_image = _resolve_pushed_image_digest(
@@ -1647,10 +1680,72 @@ class PublishedImage:
 ImageSource = BuildAndPush | PublishedImage
 
 
+def _resource_spec(image_pull_secrets: tuple[str, ...]) -> Mapping[str, object]:
+    if not image_pull_secrets:
+        return _OPERATOR_DEFAULT_RESOURCE_SPEC
+    return {
+        "image_pull_secrets": [{"name": name} for name in image_pull_secrets],
+    }
+
+
+# Mirrors the control plane's own LocalObjectReference.name validation (a
+# Kubernetes DNS subdomain), so a bad secret name is rejected here instead of
+# after a build and push.
+_K8S_SECRET_NAME_RE = re.compile(
+    r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+)
+
+
+@dataclass(frozen=True, slots=True)
+class CreateOnlySettings:
+    """Options for an external-image deployment that only apply when
+    creating one, never when updating an existing deployment."""
+
+    placement: RequestedPlacement
+    image_pull_secrets: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        for name in self.image_pull_secrets:
+            if len(name) > 253 or not _K8S_SECRET_NAME_RE.match(name):
+                raise click.UsageError(
+                    f"--image-pull-secret {name!r} is not a valid Kubernetes "
+                    "secret name (lowercase alphanumeric characters, '-', or "
+                    "'.' only, max 253 characters)."
+                )
+
+    @property
+    def requested(self) -> bool:
+        return self.placement.requested or bool(self.image_pull_secrets)
+
+    def ensure_creatable(self, push_to: str | None, image_uri: str | None) -> None:
+        if push_to is None and image_uri is None and self.requested:
+            raise click.UsageError(
+                "--listener-id, --k8s-namespace, and --image-pull-secret only "
+                "apply when creating a deployment with --push-to or --image-uri."
+            )
+
+    def ensure_not_existing(self, deployment_id: str) -> None:
+        if self.placement.requested:
+            raise click.UsageError(
+                "Listener and namespace are fixed when a deployment is "
+                f"created. Deployment {deployment_id} already exists, so "
+                "drop --listener-id and --k8s-namespace, or create a new "
+                "deployment with a different --name."
+            )
+        if self.image_pull_secrets:
+            raise click.UsageError(
+                "langgraph deploy does not update image pull secrets on an "
+                f"existing deployment. Deployment {deployment_id} already "
+                "exists, so drop --image-pull-secret and update resource_spec "
+                "through the API or the LangSmith UI instead, or create a "
+                "new deployment with a different --name."
+            )
+
+
 @dataclass(frozen=True, slots=True)
 class CustomerRegistrySource:
     image: ImageSource
-    requested_placement: RequestedPlacement
+    create_only: CreateOnlySettings
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1672,7 +1767,7 @@ class CustomerRegistrySource:
         self, ctx: DeployContext, existing: ExistingDeployment, step: int
     ) -> DeployOutcome:
         _ensure_customer_registry_source(existing)
-        self.requested_placement.ensure_not_requested(existing.id)
+        self.create_only.ensure_not_existing(existing.id)
         image_uri, step = self._publish(ctx, step)
         _log_deploy_step(step, f"Updating deployment {existing.id}")
         updated = ctx.client.update_deployment(
@@ -1687,7 +1782,7 @@ class CustomerRegistrySource:
         )
 
     def _resolve_placement(self, ctx: DeployContext) -> Placement:
-        requested = self.requested_placement
+        requested = self.create_only.placement
         if requested.listener_id is not None:
             return requested.on(_requested_listener(ctx.client, requested.listener_id))
         if not (ctx.endpoints.is_cloud or requested.requested):
@@ -1716,7 +1811,9 @@ class CustomerRegistrySource:
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
                 source_config={
-                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
+                    "resource_spec": _resource_spec(
+                        self.create_only.image_pull_secrets
+                    ),
                     **placement.source_config(),
                 },
                 source_revision_config={"image_uri": image_uri},
@@ -1757,7 +1854,7 @@ class CustomerRegistrySource:
                 _build_image(runner, ctx.spec, image, verbose=ctx.verbose)
             step += 1
             _log_deploy_step(step, f"Pushing image {image}")
-            _push_image(runner, image, docker_config_dir=None, verbose=ctx.verbose)
+            _push_image(runner, image, docker=_docker_argv(None), verbose=ctx.verbose)
             step += 1
             digest = _resolve_pushed_image_digest(
                 runner, remote_image=image, docker_config_dir=None, verbose=ctx.verbose
@@ -1810,19 +1907,13 @@ def _select_source(
     tag: str | None,
     remote_build_flag: bool | None,
     placement: RequestedPlacement,
+    image_pull_secrets: tuple[str, ...],
     selector: DeploymentSelector,
 ) -> DeploymentSource:
-    if push_to is None and image_uri is None and placement.requested:
-        raise click.UsageError(
-            "--listener-id and --k8s-namespace only apply when creating a "
-            "deployment with --push-to or --image-uri."
-        )
-    if placement.requested and isinstance(selector, ById):
-        raise click.UsageError(
-            "Listener and namespace are fixed when a deployment is created, so "
-            "they cannot be set for an existing --deployment-id. Drop them, or "
-            "create a new deployment with --name."
-        )
+    create_only = CreateOnlySettings(placement, image_pull_secrets)
+    create_only.ensure_creatable(push_to, image_uri)
+    if create_only.requested and isinstance(selector, ById):
+        create_only.ensure_not_existing(selector.deployment_id)
     if image_uri is not None:
         if push_to is not None:
             raise click.UsageError("--image-uri cannot be combined with --push-to.")
@@ -1834,7 +1925,7 @@ def _select_source(
             raise click.UsageError("--image-uri cannot be combined with --remote.")
         return CustomerRegistrySource(
             image=PublishedImage(_validate_image_uri(image_uri)),
-            requested_placement=placement,
+            create_only=create_only,
         )
     if push_to is not None:
         if remote_build_flag is True:
@@ -1844,7 +1935,7 @@ def _select_source(
             _require_local_docker()
         return CustomerRegistrySource(
             image=BuildAndPush(reference=reference, prebuilt_image=image),
-            requested_placement=placement,
+            create_only=create_only,
         )
     if image and remote_build_flag is True:
         raise click.UsageError("--image cannot be combined with --remote builds.")
@@ -1874,7 +1965,10 @@ def _create_host_backend_client(
     host_url: str | None,
     api_key: str | None,
     env_vars: dict[str, str] | None = None,
+    *,
+    no_input: bool | None = None,
 ) -> HostBackendClient:
+    effective_no_input = _no_input if no_input is None else no_input
     if env_vars is None:
         env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG)
     resolved_api_key = api_key
@@ -1889,7 +1983,7 @@ def _create_host_backend_client(
                 resolved_api_key = val
                 break
     if not resolved_api_key:
-        if _no_input:
+        if effective_no_input:
             raise click.ClickException(
                 "No LangSmith API key found. Set LANGSMITH_API_KEY in the "
                 "environment or .env file."
@@ -1914,6 +2008,8 @@ def _create_host_backend_client(
 def _call_host_backend_with_optional_tenant(
     client: HostBackendClient,
     operation: Callable[[HostBackendClient], _T],
+    *,
+    no_input: bool | None = None,
 ) -> _T:
     """Run *operation*, prompting for a workspace ID on org-scoped 403s.
 
@@ -1922,6 +2018,7 @@ def _call_host_backend_with_optional_tenant(
     in-place so all subsequent calls through the same instance are
     tenant-aware.
     """
+    effective_no_input = _no_input if no_input is None else no_input
     prompted_for_tenant = False
 
     while True:
@@ -1933,7 +2030,7 @@ def _call_host_backend_with_optional_tenant(
                 and err.status_code == 403
                 and "requires workspace specification" in err.message
             ):
-                if _no_input:
+                if effective_no_input:
                     raise click.ClickException(
                         "API key is org-scoped and requires a workspace ID. "
                         "Set LANGSMITH_TENANT_ID in your .env file or "
@@ -2116,11 +2213,13 @@ def _deploy_base_options(
             ),
             click.option(
                 "--deployment-type",
-                type=click.Choice(["dev", "prod"]),
+                type=click.Choice(["serverless", "dedicated", "dev", "prod"]),
                 default="dev",
                 show_default=True,
                 help=(
-                    "Deployment type (used when creating a new deployment). "
+                    "Deployment type (used when creating a new deployment): "
+                    "serverless (scale to zero) or dedicated (prod); "
+                    "legacy dev and prod are also accepted. "
                     "Ignored with --push-to."
                 ),
             ),
@@ -2182,6 +2281,18 @@ def _deploy_base_options(
                     "creating a deployment with --push-to or --image-uri."
                 ),
             ),
+            click.option(
+                "--image-pull-secret",
+                "image_pull_secrets",
+                multiple=True,
+                help=(
+                    "Name of a Kubernetes secret (type kubernetes.io/"
+                    "dockerconfigjson) in the deployment's namespace to use for "
+                    "pulling the image from a private registry. Repeat to give "
+                    "more than one. Only used when creating a deployment with "
+                    "--push-to or --image-uri."
+                ),
+            ),
             click.option(
                 "--config",
                 "-c",
@@ -2295,6 +2406,7 @@ def _deploy_cmd(
     image_uri: str | None,
     listener_id: str | None,
     k8s_namespace: str | None,
+    image_pull_secrets: tuple[str, ...],
     tag: str | None,
     base_image: str | None,
     install_command: str | None,
@@ -2371,6 +2483,7 @@ def _deploy_cmd(
         tag=tag,
         remote_build_flag=remote_build_flag,
         placement=RequestedPlacement(listener_id, k8s_namespace),
+        image_pull_secrets=image_pull_secrets,
         selector=selector,
     )
 
@@ -2397,7 +2510,9 @@ def _deploy_cmd(
             ),
             verbose=verbose,
             selector=selector,
-            deployment_type=deployment_type,
+            deployment_type={"serverless": "dev_zero", "dedicated": "prod"}.get(
+                deployment_type, deployment_type
+            ),
             secrets=secrets,
             tracked_packages=tracked_packages,
         )
@@ -2518,6 +2633,13 @@ def deploy_listeners() -> None:
 
 @OPT_HOST_API_KEY
 @OPT_HOST_URL
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    default=False,
+    help="Emit the listeners as a JSON array instead of a table.",
+)
 @deploy_listeners.command(
     "list",
     help=(
@@ -2526,11 +2648,16 @@ def deploy_listeners() -> None:
         "--listener-id <id>` to deploy through it."
     ),
 )
-def deploy_listeners_list(api_key: str | None, host_url: str | None) -> None:
-    client = _create_host_backend_client(host_url, api_key)
+def deploy_listeners_list(
+    api_key: str | None, host_url: str | None, json_output: bool
+) -> None:
+    client = _create_host_backend_client(host_url, api_key, no_input=json_output)
     listeners = _call_host_backend_with_optional_tenant(
-        client, lambda c: c.list_listeners()
+        client, lambda c: c.list_listeners(), no_input=json_output
     )
+    if json_output:
+        click.echo(json_mod.dumps(listeners))
+        return
     if not listeners:
         click.echo("No listeners found for this workspace.")
         return

libs/cli/langgraph_cli/host_backend.py

@@ -249,9 +249,12 @@ class HostBackendClient:
         return self._request("DELETE", f"/v2/deployments/{deployment_id}")
 
     def request_push_token(self, deployment_id: str) -> dict[str, Any]:
+        # Tells the control plane this client logs in with the returned
+        # `username` and pushes to `repository_url`. LangSmith AWS requires it.
         return self._request(
             "POST",
             f"/v2/deployments/{deployment_id}/push-token",
+            {"supports_registry_username": True},
         )
 
     def request_upload_url(self, deployment_id: str) -> dict[str, Any]:

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -14,6 +14,7 @@ import langgraph_cli.deploy as deploy_module
 from langgraph_cli.cli import cli, prepare_args_and_stdin
 from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config
 from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version
+from langgraph_cli.host_backend import HostBackendError
 from langgraph_cli.util import clean_empty_lines
 
 FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
@@ -535,6 +536,121 @@ def test_deploy_listeners_list_command_no_results(monkeypatch) -> None:
     assert result.output.strip() == "No listeners found for this workspace."
 
 
+def test_deploy_listeners_list_command_json(monkeypatch) -> None:
+    runner = CliRunner()
+
+    listeners = [
+        {
+            "id": "listener-1",
+            "compute_id": "prod-cluster",
+            "compute_config": {"k8s_namespaces": ["agents"]},
+        },
+        {
+            "id": "listener-2",
+            "compute_id": "multi-cluster",
+            "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+        },
+    ]
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            return listeners
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.output) == listeners
+
+
+def test_deploy_listeners_list_command_json_no_results(monkeypatch) -> None:
+    runner = CliRunner()
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            return []
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.output) == []
+
+
+def test_deploy_listeners_list_command_json_org_scoped_403(monkeypatch) -> None:
+    """With --json, an org-scoped 403 must raise cleanly instead of
+    prompting for a workspace ID, which would write to stdout and hang
+    waiting for input that a scripted invocation never provides."""
+    runner = CliRunner()
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            raise HostBackendError("requires workspace specification", status_code=403)
+
+        def set_tenant(self, tenant_id: str) -> None:
+            raise AssertionError("must not retry with a prompted tenant with --json")
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    prompt_calls = []
+    monkeypatch.setattr(
+        click, "prompt", lambda *a, **k: prompt_calls.append((a, k)) or "unreachable"
+    )
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert prompt_calls == [], "must not prompt interactively with --json"
+    assert result.exit_code != 0
+    assert "LANGSMITH_TENANT_ID" in result.output
+
+
 def test_deploy_revisions_list_command(monkeypatch) -> None:
     runner = CliRunner()
     captured: dict[str, str] = {}

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -52,6 +52,8 @@ ARCHIVE = ("/tmp/src.tgz", 2048, "langgraph.json")
 OBJECT_PATH = "tarballs/src.tgz"
 PLATFORM_FORMAT = "{{.Os}}/{{.Architecture}}"
 DIGESTS_FORMAT = "{{json .RepoDigests}}"
+CONTEXT_HOST_FORMAT = "{{.Endpoints.docker.Host}}"
+CONTEXT_HOST = "unix:///Users/dev/.docker/run/docker.sock"
 NOT_A_CLI_DEPLOYMENT = (
     "push token is only available for 'internal_docker' source deployments"
 )
@@ -93,6 +95,7 @@ class ControlPlaneDouble:
     timeline: list[str]
     existing_deployments: list[dict] = field(default_factory=list)
     push_token_status: int = 200
+    push_token_extra: dict = field(default_factory=dict)
     create_error: str | None = None
     listeners: list[dict] = field(default_factory=list)
     listeners_by_id: dict[str, dict] = field(default_factory=dict)
@@ -147,7 +150,12 @@ class ControlPlaneDouble:
             if self.push_token_status != 200:
                 return httpx.Response(self.push_token_status, text=NOT_A_CLI_DEPLOYMENT)
             return httpx.Response(
-                200, json={"token": PUSH_TOKEN, "registry_url": REGISTRY_URL}
+                200,
+                json={
+                    "token": PUSH_TOKEN,
+                    "registry_url": REGISTRY_URL,
+                    **self.push_token_extra,
+                },
             )
         if path.endswith("/upload-url"):
             return httpx.Response(
@@ -186,6 +194,8 @@ class DockerCommand:
 class DockerDouble:
     timeline: list[str]
     failing_pushes: int = 0
+    context_lookup_fails: bool = False
+    context_host: str = CONTEXT_HOST
     builds: list[dict] = field(default_factory=list)
     commands: list[DockerCommand] = field(default_factory=list)
 
@@ -229,6 +239,10 @@ class DockerDouble:
         if "push" in args and self.failing_pushes > 0:
             self.failing_pushes -= 1
             raise click.exceptions.Exit(1)
+        if CONTEXT_HOST_FORMAT in args:
+            if self.context_lookup_fails:
+                raise click.exceptions.Exit(1)
+            return f"{self.context_host}\n", None
         if PLATFORM_FORMAT in args:
             return "linux/amd64\n", None
         if DIGESTS_FORMAT in args:
@@ -238,6 +252,8 @@ class DockerDouble:
 
     @staticmethod
     def _verb(args: tuple[str, ...]) -> str:
+        if CONTEXT_HOST_FORMAT in args:
+            return "inspect-context"
         if PLATFORM_FORMAT in args:
             return "inspect-platform"
         if DIGESTS_FORMAT in args:
@@ -286,7 +302,12 @@ def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployPro
         json.dumps({"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]})
     )
     monkeypatch.chdir(tmp_path)
-    for name in ("LANGSMITH_TENANT_ID", "LANGSMITH_ENDPOINT", "LANGGRAPH_HOST_URL"):
+    for name in (
+        "LANGSMITH_TENANT_ID",
+        "LANGSMITH_ENDPOINT",
+        "LANGGRAPH_HOST_URL",
+        "DOCKER_HOST",
+    ):
         monkeypatch.delenv(name, raising=False)
 
     timeline: list[str] = []
@@ -334,6 +355,7 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
         CREATE_DEPLOYMENT,
         "docker build",
         _push_token(CREATED_ID),
+        "docker inspect-context",
         "docker login",
         "docker tag",
         "docker push",
@@ -343,6 +365,27 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
     assert "Deployment updated" in result.output
 
 
+@pytest.mark.parametrize(
+    ("deployment_type", "wire_type"),
+    [
+        ("serverless", "dev_zero"),
+        ("dedicated", "prod"),
+        ("dev", "dev"),
+        ("prod", "prod"),
+    ],
+)
+@pytest.mark.parametrize("build_mode", ["--no-remote", "--remote"])
+def test_deployment_type_aliases(
+    deploy_project: DeployProject, deployment_type: str, wire_type: str, build_mode: str
+) -> None:
+    result = deploy_project.run(build_mode, "--deployment-type", deployment_type)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "deployment_type": wire_type
+    }
+
+
 def test_first_local_deploy_creates_an_internal_docker_deployment(
     deploy_project: DeployProject,
 ) -> None:
@@ -363,7 +406,7 @@ def test_first_local_deploy_creates_an_internal_docker_deployment(
         pytest.param(
             "arm64",
             ("docker", "buildx", "build"),
-            ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+            ("--platform", "linux/amd64", "--load", "--progress=plain"),
             id="apple_silicon_cross_builds_for_linux_amd64",
         ),
         pytest.param(
@@ -400,7 +443,7 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
 
     login = deploy_project.docker.command("login")
     assert login.args[:2] == ("docker", "--config")
-    assert login.args[3:] == (
+    assert login.args[-5:] == (
         "login",
         "-u",
         "oauth2accesstoken",
@@ -410,6 +453,95 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
     assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"
 
 
+def test_local_deploy_logs_in_and_pushes_through_the_active_docker_context(
+    deploy_project: DeployProject,
+) -> None:
+    """The temporary config has no current context, so without the active
+    context's endpoint Docker would push through /var/run/docker.sock."""
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        args = deploy_project.docker.command(verb).args
+        assert args[3:5] == ("--host", CONTEXT_HOST), verb
+
+
+def test_local_deploy_leaves_an_explicit_docker_host_to_docker(
+    deploy_project: DeployProject, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("DOCKER_HOST", "tcp://127.0.0.1:2375")
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        assert "--host" not in deploy_project.docker.command(verb).args, verb
+
+
+def test_local_deploy_keeps_dockers_default_when_the_context_lookup_fails(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_lookup_fails = True
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
+def test_local_deploy_ignores_a_context_without_a_docker_endpoint(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_host = "<no value>"
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
+def test_local_deploy_tells_the_control_plane_it_honors_registry_usernames(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    assert deploy_project.control_plane.bodies[_push_token(CREATED_ID)] == {
+        "supports_registry_username": True
+    }
+
+
+ECR_REPOSITORY = "123456789012.dkr.ecr.us-east-2.amazonaws.com/lsd/agent-1"
+ECR_PUSH_TOKEN_FIELDS = {"username": "AWS", "repository_url": ECR_REPOSITORY}
+
+
+def test_local_deploy_logs_in_with_the_username_the_control_plane_returns(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
+
+    deploy_project.run("--no-remote")
+
+    login = deploy_project.docker.command("login")
+    assert login.args[-5:] == (
+        "login",
+        "-u",
+        "AWS",
+        "--password-stdin",
+        "123456789012.dkr.ecr.us-east-2.amazonaws.com",
+    )
+
+
+def test_local_deploy_pushes_to_exactly_the_returned_repository(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
+
+    deploy_project.run("--no-remote")
+
+    tag = deploy_project.docker.command("tag")
+    assert tag.args[-1] == f"{ECR_REPOSITORY}:latest"
+
+
 def test_local_deploy_tags_the_build_into_the_token_registry(
     deploy_project: DeployProject,
 ) -> None:
@@ -458,6 +590,7 @@ def test_prebuilt_image_is_validated_and_pushed_without_a_build(
     assert deploy_project.docker.builds == []
     assert deploy_project.docker.verbs() == [
         "docker inspect-platform",
+        "docker inspect-context",
         "docker login",
         "docker tag",
         "docker push",
@@ -918,6 +1051,100 @@ def test_self_hosted_control_plane_places_when_asked(
     }
 
 
+def test_image_pull_secret_is_included_in_resource_spec_on_create(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--image-pull-secret",
+        "regcred",
+        "--image-pull-secret",
+        "other-secret",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {
+            "image_pull_secrets": [
+                {"name": "regcred"},
+                {"name": "other-secret"},
+            ],
+        },
+    }
+
+
+def test_image_pull_secret_works_with_image_uri_too(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {"image_pull_secrets": [{"name": "regcred"}]},
+    }
+
+
+def test_image_pull_secret_without_push_to_or_image_uri_is_refused(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--image-pull-secret", "regcred")
+
+    assert result.exit_code != 0
+    assert "only apply when creating a deployment with --push-to" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_an_invalid_image_pull_secret_name_is_rejected_before_any_docker_work(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, "--image-pull-secret", "Not_Valid!"
+    )
+
+    assert result.exit_code != 0
+    assert "not a valid Kubernetes" in result.output
+    assert deploy_project.docker.verbs() == []
+    assert deploy_project.timeline == []
+
+
+def test_image_pull_secret_cannot_update_an_existing_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--deployment-id",
+        "dep-1",
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--image-pull-secret",
+        "regcred",
+    )
+
+    assert result.exit_code != 0
+    assert "--image-pull-secret" in result.output
+    assert "dep-1" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_image_pull_secret_cannot_update_a_deployment_found_by_name(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run(
+        "--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
+    )
+
+    assert result.exit_code != 0
+    assert "--image-pull-secret" in result.output
+    assert deploy_project.docker.verbs() == []
+    assert _patch("dep-ext") not in deploy_project.timeline
+
+
 def test_updating_a_deployment_never_looks_up_listeners(
     deploy_project: DeployProject,
 ) -> None:

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -16,6 +16,7 @@ from langgraph_cli.deploy import (
     BuildAndPush,
     ById,
     ByName,
+    CreateOnlySettings,
     CustomerRegistrySource,
     DockerBuildCommand,
     ExistingDeployment,
@@ -34,6 +35,7 @@ from langgraph_cli.deploy import (
     _parse_env_from_config,
     _resolve_env_path,
     _resolve_pushed_image_digest,
+    _resource_spec,
     _select_source,
     _validate_prebuilt_image,
     find_deployment_by_name,
@@ -46,7 +48,9 @@ from langgraph_cli.image_reference import ImageReference
 
 class TestDockerConfigForToken:
     def test_creates_config_json(self):
-        with _docker_config_for_token("us-docker.pkg.dev", "my-token") as cfg:
+        with _docker_config_for_token(
+            "us-docker.pkg.dev", "my-token", "oauth2accesstoken"
+        ) as cfg:
             config_path = os.path.join(cfg, "config.json")
             assert os.path.isfile(config_path)
             with open(config_path) as f:
@@ -54,13 +58,26 @@ class TestDockerConfigForToken:
             expected_auth = base64.b64encode(b"oauth2accesstoken:my-token").decode()
             assert data == {"auths": {"us-docker.pkg.dev": {"auth": expected_auth}}}
 
+    def test_uses_the_given_username(self):
+        with _docker_config_for_token(
+            "123456789012.dkr.ecr.us-east-2.amazonaws.com", "ecr-password", "AWS"
+        ) as cfg:
+            with open(os.path.join(cfg, "config.json")) as f:
+                data = json.load(f)
+        expected_auth = base64.b64encode(b"AWS:ecr-password").decode()
+        assert data["auths"]["123456789012.dkr.ecr.us-east-2.amazonaws.com"] == {
+            "auth": expected_auth
+        }
+
     def test_tempdir_cleaned_up(self):
-        with _docker_config_for_token("registry.example.com", "tok") as cfg:
+        with _docker_config_for_token(
+            "registry.example.com", "tok", "oauth2accesstoken"
+        ) as cfg:
             assert os.path.isdir(cfg)
         assert not os.path.exists(cfg)
 
     def test_different_registries(self):
-        with _docker_config_for_token("gcr.io", "token123") as cfg:
+        with _docker_config_for_token("gcr.io", "token123", "oauth2accesstoken") as cfg:
             with open(os.path.join(cfg, "config.json")) as f:
                 data = json.load(f)
             assert "gcr.io" in data["auths"]
@@ -593,9 +610,9 @@ class TestDockerBuildCommand:
                 False,
                 DockerBuildCommand(
                     ("docker", "buildx", "build"),
-                    ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+                    ("--platform", "linux/amd64", "--load", "--progress=plain"),
                 ),
-                id="other_hosts_cross_build_quietly",
+                id="other_hosts_capture_plain_build_logs",
             ),
             pytest.param(
                 "arm64",
@@ -621,6 +638,7 @@ class TestSelectSource:
         "tag": None,
         "remote_build_flag": None,
         "placement": RequestedPlacement(),
+        "image_pull_secrets": (),
         "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
@@ -636,7 +654,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -648,7 +666,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "v2"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -660,7 +678,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "v3"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -672,7 +690,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image="app:dev",
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
             ),
@@ -687,16 +705,35 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement("listener-1", "agents")
+                    ),
                 ),
                 id="push_to_carries_the_requested_placement",
             ),
+            pytest.param(
+                {
+                    "push_to": REPOSITORY,
+                    "image_pull_secrets": ("regcred", "other-secret"),
+                },
+                True,
+                CustomerRegistrySource(
+                    image=BuildAndPush(
+                        reference=ImageReference(REPOSITORY, "latest"),
+                        prebuilt_image=None,
+                    ),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement(), ("regcred", "other-secret")
+                    ),
+                ),
+                id="push_to_carries_the_image_pull_secrets",
+            ),
             pytest.param(
                 {"image_uri": f"{REPOSITORY}@sha256:abc123"},
                 True,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="image_uri_selects_the_published_image_source_by_digest",
             ),
@@ -705,7 +742,7 @@ class TestSelectSource:
                 False,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="image_uri_needs_no_local_docker_and_is_trimmed",
             ),
@@ -717,7 +754,9 @@ class TestSelectSource:
                 False,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement("listener-1", "agents")
+                    ),
                 ),
                 id="image_uri_carries_the_requested_placement",
             ),
@@ -806,6 +845,11 @@ class TestSelectSource:
                 "only apply when creating a deployment with --push-to",
                 id="namespace_without_push_to",
             ),
+            pytest.param(
+                {"image_pull_secrets": ("regcred",)},
+                "only apply when creating a deployment with --push-to",
+                id="image_pull_secret_without_push_to",
+            ),
             pytest.param(
                 {"image_uri": REPOSITORY, "push_to": REPOSITORY},
                 "--image-uri cannot be combined with --push-to.",
@@ -1243,7 +1287,7 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
 
     monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
 
-    with pytest.raises(click.UsageError, match="--deployment-id"):
+    with pytest.raises(click.UsageError, match="dep-1"):
         _select_source(
             push_to="registry.example.com/app",
             image=None,
@@ -1252,10 +1296,77 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
             tag=None,
             remote_build_flag=None,
             placement=RequestedPlacement(listener_id="listener-1"),
+            image_pull_secrets=(),
             selector=ById("dep-1"),
         )
 
 
+def test_a_deployment_id_with_image_pull_secrets_is_refused_without_probing_docker(
+    monkeypatch,
+):
+    def explode() -> tuple[bool, str | None]:
+        raise AssertionError("docker must not be probed for an argv-only conflict")
+
+    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
+
+    with pytest.raises(click.UsageError, match="dep-1"):
+        _select_source(
+            push_to="registry.example.com/app",
+            image=None,
+            image_uri=None,
+            image_name=None,
+            tag=None,
+            remote_build_flag=None,
+            placement=RequestedPlacement(),
+            image_pull_secrets=("regcred",),
+            selector=ById("dep-1"),
+        )
+
+
+class TestResourceSpec:
+    def test_no_secrets_keeps_the_operator_default(self):
+        assert _resource_spec(()) == {}
+
+    def test_secrets_become_local_object_references_in_order(self):
+        assert _resource_spec(("regcred", "other-secret")) == {
+            "image_pull_secrets": [
+                {"name": "regcred"},
+                {"name": "other-secret"},
+            ],
+        }
+
+
+class TestCreateOnlySettings:
+    @pytest.mark.parametrize(
+        "name",
+        [
+            pytest.param("Bad_Name", id="uppercase_and_underscore"),
+            pytest.param("", id="empty"),
+            pytest.param("-leading-hyphen", id="leading_hyphen"),
+            pytest.param("a" * 254, id="too_long"),
+        ],
+    )
+    def test_an_invalid_secret_name_is_rejected_up_front(self, name):
+        with pytest.raises(click.UsageError, match="not a valid Kubernetes"):
+            CreateOnlySettings(RequestedPlacement(), (name,))
+
+    def test_a_valid_secret_name_is_accepted(self):
+        CreateOnlySettings(RequestedPlacement(), ("regcred-1.example",))
+
+    def test_ensure_not_existing_mentions_the_api_and_ui_for_pull_secrets(self):
+        settings = CreateOnlySettings(RequestedPlacement(), ("regcred",))
+        with pytest.raises(click.UsageError, match="API or the LangSmith UI"):
+            settings.ensure_not_existing("dep-1")
+
+    def test_ensure_not_existing_keeps_the_placement_message_unchanged(self):
+        settings = CreateOnlySettings(RequestedPlacement(listener_id="listener-1"))
+        with pytest.raises(
+            click.UsageError,
+            match="Listener and namespace are fixed when a deployment is created",
+        ):
+            settings.ensure_not_existing("dep-1")
+
+
 class TestPlacementOnAKnownListener:
     @pytest.mark.parametrize(
         ("request_", "listener", "expected"),

libs/langgraph/langgraph/_internal/_constants.py

@@ -74,8 +74,6 @@ CONFIG_KEY_RUNTIME = sys.intern("__pregel_runtime")
 # holds a `Runtime` instance with context, store, stream writer, etc.
 CONFIG_KEY_RESUME_MAP = sys.intern("__pregel_resume_map")
 # holds a mapping of task ns -> resume value for resuming tasks
-CONFIG_KEY_RESUME_MAP_SENT = sys.intern("__pregel_resume_map_sent")
-# holds the set of resume map ids already forwarded to a remote graph in this run
 CONFIG_KEY_STREAM_MESSAGES_V2 = sys.intern("__pregel_stream_messages_v2")
 # when True, attach StreamMessagesHandlerV2 so content-block (v2) events
 # flow through stream_mode="messages"; set by StreamingHandler only.
@@ -136,7 +134,6 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     CONFIG_KEY_RESUME_MAP,
-    CONFIG_KEY_RESUME_MAP_SENT,
     CONFIG_KEY_STREAM_MESSAGES_V2,
     # other constants
     PUSH,

libs/langgraph/langgraph/channels/binop.py

@@ -2,6 +2,7 @@ import collections.abc
 from collections.abc import Callable, Sequence
 from typing import Any, Generic
 
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import NotRequired, Required, Self
 
 from langgraph._internal._constants import OVERWRITE
@@ -116,6 +117,9 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
     def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ, self.operator)
         empty.key = self.key
+        # A thread written while this field was a DeltaChannel stores snapshots.
+        if isinstance(checkpoint, _DeltaSnapshot):
+            checkpoint = checkpoint.value
         if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty

libs/langgraph/langgraph/graph/message.py

@@ -24,6 +24,9 @@ from typing_extensions import TypedDict, deprecated
 
 from langgraph._internal._constants import CONF, CONFIG_KEY_SEND, NS_SEP
 from langgraph.graph.state import StateGraph
+from langgraph.pregel._messages import (
+    _messages_delta_reducer as _messages_delta_reducer,
+)
 from langgraph.warnings import LangGraphDeprecatedSinceV10
 
 __all__ = (
@@ -244,71 +247,6 @@ def add_messages(
     return merged
 
 
-def _messages_delta_reducer(
-    state: list[AnyMessage], writes: list[list[AnyMessage]]
-) -> list[AnyMessage]:
-    """**Experimental.** Batch reducer for use with `DeltaChannel`.
-
-    Processes all writes in one pass — dedup by ID, `RemoveMessage`
-    tombstoning — without calling `add_messages`.
-
-    This reducer is batching-invariant, as required by `DeltaChannel`:
-    `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.
-
-    Raw dict / string / tuple inputs are coerced to typed `BaseMessage`
-    objects so that HTTP-driven graphs work without a separate coercion
-    step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`,
-    unknown-id `RemoveMessage` errors, missing-id UUID assignment, and
-    `BaseMessageChunk` conversion are not handled here.
-
-    Example::
-
-        from typing import Annotated
-        from langgraph.channels.delta import DeltaChannel
-        from langgraph.graph.message import _messages_delta_reducer
-
-        class State(TypedDict):
-            messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-    """
-
-    # Each write is either a list of message-likes or a single message-like
-    # (BaseMessage / dict / str / tuple). Only lists flatten; everything
-    # else is one message.
-    flat: list[Any] = []
-    for w in writes:
-        if isinstance(w, list):
-            flat.extend(w)
-        else:
-            flat.append(w)
-    # Steady state: the reducer's own output is already typed, so skip
-    # `convert_to_messages` on state when the first element is a BaseMessage.
-    # Only raw input (initial dicts, deserialized blobs) hits the slow path.
-    if state and isinstance(state[0], BaseMessage):
-        state_msgs = state
-    else:
-        state_msgs = cast("list[AnyMessage]", convert_to_messages(state))
-    msgs = cast("list[AnyMessage]", convert_to_messages(flat))
-
-    index: dict[str, int] = {
-        m.id: i for i, m in enumerate(state_msgs) if m.id is not None
-    }
-    result: list[AnyMessage | None] = list(state_msgs)
-    for msg in msgs:
-        mid = msg.id
-        if mid is None:
-            result.append(msg)
-        elif isinstance(msg, RemoveMessage):
-            if mid in index:
-                result[index[mid]] = None
-                del index[mid]
-        elif mid in index:
-            result[index[mid]] = msg
-        else:
-            index[mid] = len(result)
-            result.append(msg)
-    return [m for m in result if m is not None]
-
-
 @deprecated(
     "MessageGraph is deprecated in langgraph 1.0.0, to be removed in 2.0.0. Please use StateGraph with a `messages` key instead.",
     category=None,

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -1,10 +1,10 @@
 from __future__ import annotations
 
 import uuid
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from datetime import datetime, timezone
 from inspect import signature
-from typing import Any, Literal, cast
+from typing import Any, Literal
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -12,6 +12,7 @@ from langgraph.checkpoint.base import (
     ChannelVersions,
     Checkpoint,
     CheckpointTuple,
+    DeltaChannelHistory,
     PendingWrite,
 )
 from langgraph.checkpoint.base.id import uuid6
@@ -32,6 +33,7 @@ from langgraph._internal._constants import (
 )
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
+from langgraph.channels.binop import BinaryOperatorAggregate, _get_overwrite
 from langgraph.channels.delta import DeltaChannel
 from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
 
@@ -152,6 +154,23 @@ def delta_channels_with_pending_writes(
     }
 
 
+def delta_channels_overwritten(
+    specs: Mapping[str, Any], writes: Iterable[tuple[str, Any]]
+) -> set[str]:
+    """Return the names of the DeltaChannels that `writes` set with an `Overwrite`.
+
+    `update_state` saves a full snapshot of these channels in the checkpoint it
+    creates, like the loop does when a node returns an `Overwrite`. Otherwise,
+    reading the channel later starts from an older snapshot and replays the
+    writes the `Overwrite` threw away.
+    """
+    return {
+        ch
+        for ch, value in writes
+        if isinstance(specs.get(ch), DeltaChannel) and _get_overwrite(value)[0]
+    }
+
+
 def checkpoint_superseded(
     saver: BaseCheckpointSaver, config: RunnableConfig, saved: CheckpointTuple
 ) -> bool:
@@ -256,6 +275,7 @@ def create_checkpoint(
     get_next_version: GetNextVersion | None = None,
     channels_to_snapshot: set[str] | None = None,
     stored_versions: ChannelVersions | None = None,
+    trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
 ) -> Checkpoint:
     """Build a new Checkpoint from the previous one and live channel state.
 
@@ -314,7 +334,9 @@ def create_checkpoint(
         id=id or str(uuid6(clock_seq=step)),
         channel_values=values,
         channel_versions=channel_versions,
-        versions_seen=_mark_bumps_seen(checkpoint["versions_seen"], bumped),
+        versions_seen=_mark_bumps_seen(
+            checkpoint["versions_seen"], bumped, trigger_to_nodes or {}
+        ),
         updated_channels=None if updated_channels is None else sorted(updated_channels),
     )
 
@@ -322,19 +344,27 @@ def create_checkpoint(
 def _mark_bumps_seen(
     versions_seen: dict[str, ChannelVersions],
     bumped: Mapping[str, tuple[Any, Any]],
+    trigger_to_nodes: Mapping[str, Sequence[str]],
 ) -> dict[str, ChannelVersions]:
     """Advance whoever had seen a bumped channel's old version to the new one.
 
     A bump that only stores a snapshot is not a write. Left unseen, it would
-    re-fire `interrupt_before` and rerun the channel's subscribers. For each
-    entry it advances, `SNAPSHOT_BUMPS` keeps the new version and the one the
-    node really read, so `versions_seen_without_bumps` can put the read back.
+    re-fire `interrupt_before` and rerun the channel's subscribers. A channel
+    bumped from no version was never written, so it also goes to the
+    subscribers that never ran: they have no entry, and would start on the bump.
+    For each entry it advances, `SNAPSHOT_BUMPS` keeps the new version and the
+    one the node really read, so `versions_seen_without_bumps` can put the read
+    back.
     """
     if not bumped:
         return versions_seen
     out = dict(versions_seen)
+    for k, (old, _) in bumped.items():
+        if old is None:
+            for node in trigger_to_nodes.get(k, ()):
+                out.setdefault(node, {})
     marks = dict(versions_seen.get(SNAPSHOT_BUMPS, {}))
-    for node, seen in versions_seen.items():
+    for node, seen in out.items():
         if node == SNAPSHOT_BUMPS:
             continue
         for k, (old, new) in bumped.items():
@@ -380,23 +410,39 @@ def versions_seen_without_bumps(
 def _delta_channels_to_replay(
     specs: Mapping[str, BaseChannel], checkpoint: Checkpoint
 ) -> list[str]:
-    """DeltaChannels whose value at `checkpoint` the ancestor walk rebuilds.
+    """Channels whose value at `checkpoint` the ancestor walk rebuilds.
 
     A `_DeltaSnapshot` blob or a plain value (migration) resolves directly via
     `from_checkpoint`, so only a channel with nothing stored here needs the
     walk. A channel with no version was never written, so it is empty without
     one; a walk for it would find no snapshot to stop at and read every
     ancestor, every time the thread is loaded.
+
+    A `BinaryOperatorAggregate` stores its value at every version, so one with
+    nothing stored was written as a `DeltaChannel` before its annotation
+    changed, and is rebuilt the same way.
     """
     return [
         k
         for k, spec in specs.items()
-        if isinstance(spec, DeltaChannel)
+        if isinstance(spec, (DeltaChannel, BinaryOperatorAggregate))
         and k in checkpoint["channel_versions"]
         and checkpoint["channel_values"].get(k, MISSING) is MISSING
     ]
 
 
+def _replay(spec: BaseChannel, history: DeltaChannelHistory) -> BaseChannel:
+    ch = spec.from_checkpoint(history.get("seed", MISSING))
+    if isinstance(ch, DeltaChannel):
+        ch.replay_writes(history["writes"])
+    else:
+        # One update per write: the writes span many supersteps, and a single
+        # call would reject a second Overwrite and drop the writes after one.
+        for _, _, value in history["writes"]:
+            ch.update([value])
+    return ch
+
+
 def _require_saver_for_history(
     delta_channels: list[str],
     saver: BaseCheckpointSaver | None,
@@ -436,7 +482,7 @@ def channels_from_checkpoint(
 
     delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
     _require_saver_for_history(delta_channels, saver, config)
-    histories: Mapping[str, Any] = {}
+    histories: Mapping[str, DeltaChannelHistory] = {}
     if delta_channels and saver is not None and config is not None:
         histories = saver.get_delta_channel_history(
             config=config, channels=delta_channels
@@ -446,11 +492,7 @@ def channels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
-            delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
-            replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
-            replay_ch.replay_writes(history["writes"])
-            ch = replay_ch
+            ch = _replay(spec, histories[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch
@@ -475,7 +517,7 @@ async def achannels_from_checkpoint(
 
     delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
     _require_saver_for_history(delta_channels, saver, config)
-    histories: Mapping[str, Any] = {}
+    histories: Mapping[str, DeltaChannelHistory] = {}
     if delta_channels and saver is not None and config is not None:
         histories = await saver.aget_delta_channel_history(
             config=config, channels=delta_channels
@@ -485,11 +527,7 @@ async def achannels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
-            delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
-            replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
-            replay_ch.replay_writes(history["writes"])
-            ch = replay_ch
+            ch = _replay(spec, histories[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch

libs/langgraph/langgraph/pregel/_loop.py

@@ -19,6 +19,7 @@ from typing import (
     TypeVar,
     cast,
 )
+from uuid import UUID, uuid5
 
 from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
@@ -43,7 +44,6 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_REPLAY_STATE,
     CONFIG_KEY_RESUME_MAP,
-    CONFIG_KEY_RESUME_MAP_SENT,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_RUNTIME,
     CONFIG_KEY_SCRATCHPAD,
@@ -161,6 +161,12 @@ def DuplexStream(*streams: StreamProtocol) -> StreamProtocol:
     return StreamProtocol(__call__, {mode for s in streams for mode in s.modes})
 
 
+def _cacheable(writes: WritesT) -> bool:
+    # A cache hit skips the node, so a cached error would skip it without
+    # raising: only a node that finished goes to the cache.
+    return not any(c in (INTERRUPT, ERROR) for c, _ in writes)
+
+
 class PregelLoop:
     config: RunnableConfig
     store: BaseStore | None
@@ -192,6 +198,7 @@ class PregelLoop:
         Callable[
             [
                 concurrent.futures.Future | None,
+                Sequence[Any],
                 RunnableConfig,
                 Checkpoint,
                 str,
@@ -205,11 +212,13 @@ class PregelLoop:
     submit: Submit
     channels: Mapping[str, BaseChannel]
     # Futures from `checkpointer.put_writes` calls that produced delta-channel
-    # writes. `_checkpointer_put_after_previous` drains this list (swap to a
-    # local `futs` then reset to `[]` and wait/gather) before putting the
-    # next checkpoint, so a checkpoint never becomes durable before the
-    # writes that produced it. Initialised to `[]` in both sync and async
-    # `__enter__`; stays `None` only when no checkpointer.
+    # writes. `_put_checkpoint` hands this list to the save it submits, which
+    # waits for them first, so a checkpoint never becomes durable before the
+    # writes that produced it. If a write or the previous save failed, the
+    # save fails too: a DeltaChannel is rebuilt from its writes along the
+    # parent chain, so a checkpoint saved past either gap reads back short
+    # for good. Initialised to `[]` in both sync and async `__enter__`;
+    # stays `None` only when no checkpointer.
     _delta_write_futs: list[Any] | None = None
 
     # Same pattern as `_delta_write_futs` but for error-handler writes.
@@ -248,8 +257,10 @@ class PregelLoop:
     _resume_loaded_writes: Sequence[PendingWrite] = ()
     # Tasks `_reapply_writes_to_succeeded_nodes` handed loaded writes back to.
     _reapplied_task_ids: set[str]
-    # An update or fork checkpoint addressed by `checkpoint_id` that the thread
-    # has moved past, so it may already have children.
+    # Tasks that wrote during this run, the only ones exit mode saves at exit.
+    _written_task_ids: set[str]
+    # A checkpoint addressed by `checkpoint_id` that the thread has moved past,
+    # so it may already have children.
     _addressed_checkpoint_superseded: bool = False
 
     # The checkpoint_config that points at the parent loaded at `__enter__`
@@ -267,6 +278,9 @@ class PregelLoop:
     # `_put_exit_delta_writes` uses this to decide between anchoring on
     # the existing parent (True) or creating a lazy stub (False).
     _has_persisted_parent: bool = False
+    # True iff `__enter__` loaded the thread's latest checkpoint, not one a
+    # `checkpoint_id` addressed, so nothing has been built on it yet.
+    _loaded_latest: bool = False
 
     managed: ManagedValueMapping
     checkpoint: Checkpoint
@@ -337,7 +351,6 @@ class PregelLoop:
         self.interrupt_before = interrupt_before
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
-        self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF]
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
@@ -395,6 +408,14 @@ class PregelLoop:
             else ()
         )
         self.prev_checkpoint_config = None
+        # Whether the run points this loop at a checkpoint: a truthy
+        # `checkpoint_id`, this graph's `checkpoint_map` entry, or a replaying
+        # parent. Every task config carries `checkpoint_id: None`, so a key
+        # check would count every subgraph.
+        self._addressed = bool(
+            self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            or self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
+        )
         runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
         self.control = runtime.control if isinstance(runtime, Runtime) else None
 
@@ -437,13 +458,19 @@ class PregelLoop:
             return None
         return self._graph_lifecycle_events.popleft()
 
-    def put_writes(self, task_id: str, writes: WritesT) -> None:
+    def put_writes(
+        self, task_id: str, writes: WritesT, *, cached: bool = False
+    ) -> None:
         """Put writes for a task, to be read by the next tick."""
         if not writes:
             return
+        self._written_task_ids.add(task_id)
         # deduplicate writes to special channels, last write wins
         if all(w[0] in WRITES_IDX_MAP for w in writes):
             writes = list({w[0]: w for w in writes}.values())
+        # Before the writes are saved, kept in `checkpoint_pending_writes` or
+        # applied, so all of them hold the same message ids.
+        ensure_message_ids(self.specs, writes)
         if task_id == NULL_TASK_ID:
             # writes for the null task are accumulated
             self.checkpoint_pending_writes = [
@@ -451,6 +478,11 @@ class PregelLoop:
                 for w in self.checkpoint_pending_writes
                 if w[0] != task_id or w[1] not in WRITES_IDX_MAP
             ]
+            # Savers key a regular write by its position in the list, and the
+            # special ones dropped above won't be there next time, so they go
+            # last: a resume saved ahead of an update would move it to a new
+            # slot once the list is saved again, and it would apply twice.
+            writes = sorted(writes, key=lambda w: w[0] in WRITES_IDX_MAP)
             writes_to_save: WritesT = [
                 w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
             ] + list(writes)
@@ -480,14 +512,6 @@ class PregelLoop:
 
         # save writes
         self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
-        # Assign stable IDs to any id=None BaseMessages in DeltaChannel writes
-        # before the background thread serialises them. Without this, reducers
-        # that assign IDs inside apply_writes() race with serialisation and
-        # store id=None, causing get_state() replays to produce a different UUID
-        # on every call.
-        for c, v in writes_to_save:
-            if isinstance(self.specs.get(c), DeltaChannel):
-                ensure_message_ids(v)
         if self.durability != "exit" and self.checkpointer_put_writes is not None:
             config = patch_configurable(
                 self.checkpoint_config,
@@ -530,7 +554,7 @@ class PregelLoop:
                 self._error_handler_write_futs.append(fut)
         # output writes
         if hasattr(self, "tasks"):
-            self.output_writes(task_id, writes)
+            self.output_writes(task_id, writes, cached=cached)
 
     def _put_pending_writes(self) -> None:
         if self.checkpointer_put_writes is None:
@@ -547,10 +571,12 @@ class PregelLoop:
                 CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
             },
         )
-        # group by task id
+        # Writes loaded with the checkpoint are stored on it already. Saving them
+        # again can file them under new indexes, and then they apply twice.
         by_task = defaultdict(list)
         for task_id, channel, value in self.checkpoint_pending_writes:
-            by_task[task_id].append((channel, value))
+            if task_id in self._written_task_ids:
+                by_task[task_id].append((channel, value))
         # submit writes to checkpointer
         for task_id, writes in by_task.items():
             if self.checkpointer_put_writes_accepts_task_path and hasattr(
@@ -679,10 +705,6 @@ class PregelLoop:
             self.status = "done"
             return False
 
-        if self.control is not None and self.control.drain_requested:
-            self.status = "draining"
-            return False
-
         # if there are pending writes from a previous loop, apply them
         if self._reapplies_pending_writes and self.checkpoint_pending_writes:
             self._reapply_writes_to_succeeded_nodes(self.tasks)
@@ -695,6 +717,13 @@ class PregelLoop:
             self.status = "interrupt_before"
             raise GraphInterrupt()
 
+        # stop before running the next tasks if a drain was requested. after
+        # the breakpoint check: a resume passes the next step's breakpoints,
+        # so draining in front of one would skip it
+        if self.control is not None and self.control.drain_requested:
+            self.status = "draining"
+            return False
+
         # produce debug output
         self._emit("tasks", map_debug_tasks, self.tasks.values())
 
@@ -892,6 +921,7 @@ class PregelLoop:
             if tid != NULL_TASK_ID and isinstance(self.specs.get(ch), DeltaChannel)
         }
         self._exit_command_writes = []
+        self._written_task_ids = set()
         # Resuming from a previous checkpoint requires two things:
         # 1. A prior checkpoint exists (channel_versions is non-empty)
         # 2. The input signals continuation (not a fresh run with new input)
@@ -919,7 +949,7 @@ class PregelLoop:
         # writes so that interrupt() calls re-fire instead of returning
         # stale values. But if we're actively resuming, keep them —
         # multi-interrupt scenarios need previously resolved values preserved.
-        is_time_traveling = self.is_replaying and (
+        is_time_traveling = self._addressed and (
             # Time-travel to a subgraph checkpoint: the parent sets
             # RESUMING=True (it can't distinguish time-travel from resume),
             # so we check if this subgraph's own ns is in checkpoint_map.
@@ -938,6 +968,10 @@ class PregelLoop:
                 or configurable.get(CONFIG_KEY_RESUMING, False)
             )
         )
+        # Finished tasks rerun only on time travel, or on a resume at a
+        # checkpoint the thread has moved past. A resume addressed at the
+        # thread's latest checkpoint hands their writes back, as without an id.
+        self.is_replaying = is_time_traveling or self._addressed_checkpoint_superseded
         if is_time_traveling:
             self.checkpoint_pending_writes = [
                 w for w in self.checkpoint_pending_writes if w[1] != RESUME
@@ -989,7 +1023,6 @@ class PregelLoop:
                     and all(is_xxh3_128_hexdigest(k) for k in resume)
                 ):
                     self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
-                    self.config[CONF][CONFIG_KEY_RESUME_MAP_SENT] = set()
                 else:
                     if len(self._pending_interrupts()) > 1:
                         raise RuntimeError(
@@ -1007,6 +1040,8 @@ class PregelLoop:
             # save writes
             for tid, ws in writes.items():
                 if forks:
+                    # These skip `put_writes`, which gives messages their ids.
+                    ensure_message_ids(self.specs, ws)
                     carried.extend((tid, c, v) for c, v in ws)
                 else:
                     self.put_writes(tid, ws)
@@ -1071,6 +1106,9 @@ class PregelLoop:
             )
         # map inputs to channel updates
         elif input_writes := deque(map_input(input_keys, self.input)):
+            # Before the input is applied, so this run reads the message ids
+            # that are saved with it.
+            ensure_message_ids(self.specs, input_writes)
             # discard any unfinished tasks from previous checkpoint
             discard_tasks = prepare_next_tasks(
                 self.checkpoint,
@@ -1116,16 +1154,26 @@ class PregelLoop:
                         self._exit_delta_writes.append(
                             (self.step, NULL_TASK_ID, "", c, v)
                         )
-            # Persist delta-channel input writes so sub-freq inputs are
-            # recoverable via ancestor walk (mirrors the Command input path).
+            # A DeltaChannel reads its input from the writes stored on the
+            # checkpoint this run starts from, under a task id of their own:
+            # readers apply a checkpoint's NULL_TASK_ID writes as its own state.
+            # A new thread has no checkpoint to store them on, and one a
+            # `checkpoint_id` addressed may have children that would read them,
+            # so then the input checkpoint snapshots the channel instead.
             if self.durability != "exit":
                 delta_input = [
                     (c, v)
                     for c, v in input_writes
                     if isinstance(self.specs.get(c), DeltaChannel)
                 ]
-                if delta_input:
-                    self.put_writes(NULL_TASK_ID, delta_input)
+                if delta_input and self._has_persisted_parent and self._loaded_latest:
+                    self.put_writes(
+                        str(uuid5(UUID(self.checkpoint["id"]), INPUT)), delta_input
+                    )
+                else:
+                    self._delta_channels_forced_snapshot.update(
+                        c for c, _ in delta_input
+                    )
             # save input checkpoint
             self.updated_channels = updated_channels
             self._put_checkpoint({"source": "input"})
@@ -1139,11 +1187,11 @@ class PregelLoop:
             # checkpoint ID since the fork was created after the subgraph's
             # checkpoints from the original execution.
             #
-            # Only gate on is_time_traveling (not is_replaying). When the
+            # Only gate on is_time_traveling (not _addressed). When the
             # client resumes with an explicit checkpoint_id that happens to
             # point at the current head (e.g. LangGraph Studio sending
             # `checkpoint: {checkpoint_id}` alongside Command(resume=...)),
-            # is_replaying is True but is_time_traveling is False. In that
+            # the run is addressed but not time traveling. In that
             # case subgraphs should load their latest checkpoint normally,
             # not go through ReplayState's before-bound lookup which would
             # miss subgraph checkpoints created during processing of the
@@ -1255,6 +1303,7 @@ class PregelLoop:
             else None,
             channels_to_snapshot=channels_to_snapshot,
             stored_versions=self.checkpoint_previous_versions,
+            trigger_to_nodes=self.trigger_to_nodes,
         )
         for k in channels_to_snapshot:
             new_counters[k] = (0, 0)
@@ -1301,12 +1350,17 @@ class PregelLoop:
             )
             self.checkpoint_previous_versions = channel_versions
 
+            # Take this checkpoint's writes now: saves run in the background
+            # and can start out of order, so a save that took them itself
+            # could get another checkpoint's writes.
+            delta_write_futs, self._delta_write_futs = self._delta_write_futs, []
             # save it, without blocking
             # if there's a previous checkpoint save in progress, wait for it
             # ensuring checkpointers receive checkpoints in order
             self._put_checkpoint_fut = self.submit(
                 self._checkpointer_put_after_previous,
                 getattr(self, "_put_checkpoint_fut", None),
+                delta_write_futs,
                 self.checkpoint_config,
                 copy_checkpoint(self.checkpoint),
                 self.checkpoint_metadata,
@@ -1339,6 +1393,14 @@ class PregelLoop:
         ):
             return
 
+        # A checkpoint a `checkpoint_id` addressed may already have children,
+        # which read the writes stored on it, so then the channels this run
+        # wrote snapshot instead.
+        if self._has_persisted_parent and not self._loaded_latest:
+            self._delta_channels_forced_snapshot.update(
+                w[3] for w in self._exit_delta_writes
+            )
+
         counters = dict(
             self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
         )
@@ -1376,6 +1438,7 @@ class PregelLoop:
             self._put_checkpoint_fut = self.submit(
                 self._checkpointer_put_after_previous,
                 getattr(self, "_put_checkpoint_fut", None),
+                (),
                 stub_put_config,
                 stub_cp,
                 {"step": -2},
@@ -1643,21 +1706,19 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     def _checkpointer_put_after_previous(
         self,
         prev: concurrent.futures.Future | None,
+        delta_write_futs: Sequence[concurrent.futures.Future],
         config: RunnableConfig,
         checkpoint: Checkpoint,
         metadata: CheckpointMetadata,
         new_versions: ChannelVersions,
     ) -> RunnableConfig:
-        if self._delta_write_futs:
-            futs, self._delta_write_futs = self._delta_write_futs, []
-            concurrent.futures.wait(futs)
-        try:
-            if prev is not None:
-                prev.result()
-        finally:
-            cast(BaseCheckpointSaver, self.checkpointer).put(
-                config, checkpoint, metadata, new_versions
-            )
+        for fut in delta_write_futs:
+            fut.result()
+        if prev is not None:
+            prev.result()
+        cast(BaseCheckpointSaver, self.checkpointer).put(
+            config, checkpoint, metadata, new_versions
+        )
 
     def match_cached_writes(self) -> Sequence[PregelExecutableTask]:
         if self.cache is None:
@@ -1679,7 +1740,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     ) -> PregelExecutableTask | None:
         if pushed := super().accept_push(task, write_idx, call):
             for task in self.match_cached_writes():
-                self.output_writes(task.id, task.writes, cached=True)
+                self.put_writes(task.id, task.writes, cached=True)
         return pushed
 
     def schedule_error_handler(
@@ -1716,16 +1777,18 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         if self._reapplies_pending_writes:
             self._reapply_writes_to_succeeded_nodes({handler_task.id: handler_task})
         for task in self.match_cached_writes():
-            self.output_writes(task.id, task.writes, cached=True)
+            self.put_writes(task.id, task.writes, cached=True)
         return handler_task
 
-    def put_writes(self, task_id: str, writes: WritesT) -> None:
+    def put_writes(
+        self, task_id: str, writes: WritesT, *, cached: bool = False
+    ) -> None:
         """Put writes for a task, to be read by the next tick."""
-        super().put_writes(task_id, writes)
-        if not writes or self.cache is None or not hasattr(self, "tasks"):
+        super().put_writes(task_id, writes, cached=cached)
+        if cached or not writes or self.cache is None or not hasattr(self, "tasks"):
             return
         task = self.tasks.get(task_id)
-        if task is None or task.cache_key is None:
+        if task is None or task.cache_key is None or not _cacheable(writes):
             return
         self.submit(
             self.cache.set,
@@ -1748,7 +1811,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                 self._addressed_checkpoint_superseded = checkpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )
@@ -1769,6 +1832,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # Normal case: fetch the most recent checkpoint for this
             # graph/thread. Returns None on first invocation.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
+            self._loaded_latest = True
 
         # Capture before the synthetic-empty fallback below overwrites `saved`.
         # `_put_exit_delta_writes` uses this on first run (no persisted parent)
@@ -1898,23 +1962,19 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     async def _checkpointer_put_after_previous(
         self,
         prev: asyncio.Task | None,
+        delta_write_futs: Sequence[asyncio.Future],
         config: RunnableConfig,
         checkpoint: Checkpoint,
         metadata: CheckpointMetadata,
         new_versions: ChannelVersions,
     ) -> RunnableConfig:
-        # Drain DeltaChannel write futures before committing the checkpoint so
-        # ancestor walks never see a checkpoint without its backing writes.
-        if self._delta_write_futs:
-            futs, self._delta_write_futs = self._delta_write_futs, []
-            await asyncio.gather(*futs)
-        try:
-            if prev is not None:
-                await prev
-        finally:
-            await cast(BaseCheckpointSaver, self.checkpointer).aput(
-                config, checkpoint, metadata, new_versions
-            )
+        if delta_write_futs:
+            await asyncio.gather(*delta_write_futs)
+        if prev is not None:
+            await prev
+        await cast(BaseCheckpointSaver, self.checkpointer).aput(
+            config, checkpoint, metadata, new_versions
+        )
 
     async def amatch_cached_writes(self) -> Sequence[PregelExecutableTask]:
         if self.cache is None:
@@ -1936,7 +1996,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     ) -> PregelExecutableTask | None:
         if pushed := super().accept_push(task, write_idx, call):
             for task in await self.amatch_cached_writes():
-                self.output_writes(task.id, task.writes, cached=True)
+                self.put_writes(task.id, task.writes, cached=True)
         return pushed
 
     async def aschedule_error_handler(
@@ -1973,19 +2033,18 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         if self._reapplies_pending_writes:
             self._reapply_writes_to_succeeded_nodes({handler_task.id: handler_task})
         for task in await self.amatch_cached_writes():
-            self.output_writes(task.id, task.writes, cached=True)
+            self.put_writes(task.id, task.writes, cached=True)
         return handler_task
 
-    def put_writes(self, task_id: str, writes: WritesT) -> None:
+    def put_writes(
+        self, task_id: str, writes: WritesT, *, cached: bool = False
+    ) -> None:
         """Put writes for a task, to be read by the next tick."""
-        super().put_writes(task_id, writes)
-        if not writes or self.cache is None or not hasattr(self, "tasks"):
+        super().put_writes(task_id, writes, cached=cached)
+        if cached or not writes or self.cache is None or not hasattr(self, "tasks"):
             return
         task = self.tasks.get(task_id)
-        if task is None or task.cache_key is None:
-            return
-        if writes[0][0] in (INTERRUPT, ERROR):
-            # only cache successful tasks
+        if task is None or task.cache_key is None or not _cacheable(writes):
             return
         self.submit(
             self.cache.aset,
@@ -2008,7 +2067,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                 self._addressed_checkpoint_superseded = await acheckpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )
@@ -2029,6 +2088,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # Normal case: fetch the most recent checkpoint for this
             # graph/thread. Returns None on first invocation.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
+            self._loaded_latest = True
 
         # Capture before the synthetic-empty fallback below overwrites `saved`.
         # `_put_exit_delta_writes` uses this on first run (no persisted parent)

libs/langgraph/langgraph/pregel/_messages.py

@@ -1,6 +1,13 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Callable, Iterator, Sequence
+from collections.abc import (
+    AsyncIterator,
+    Callable,
+    Iterator,
+    Mapping,
+    MutableSequence,
+    Sequence,
+)
 from dataclasses import fields, is_dataclass
 from typing import (
     Any,
@@ -10,16 +17,18 @@ from typing import (
 from uuid import UUID, uuid4
 
 from langchain_core.callbacks import BaseCallbackHandler
-from langchain_core.messages import BaseMessage, ToolMessage
+from langchain_core.messages import AnyMessage, BaseMessage, RemoveMessage, ToolMessage
 from langchain_core.messages.utils import convert_to_messages
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
 from pydantic import BaseModel
 
 from langgraph._internal._config import filter_to_user_tags
 from langgraph._internal._constants import NS_SEP
+from langgraph.channels.binop import _get_overwrite
+from langgraph.channels.delta import DeltaChannel
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
 from langgraph.pregel.protocol import StreamChunk
-from langgraph.types import Command
+from langgraph.types import Command, Overwrite
 
 try:
     from langchain_core.tracers._streaming import _StreamingCallbackHandler
@@ -408,6 +417,74 @@ class StreamMessagesHandlerV2(StreamMessagesHandler, _V2StreamingCallbackHandler
             self.stream((meta[0], "messages", (event, v2_meta)))
 
 
+# Defined here rather than in `langgraph.graph.message`, which re-exports it:
+# `ensure_message_ids` has to recognize it, and importing that module from the
+# loop would be an import cycle.
+def _messages_delta_reducer(
+    state: list[AnyMessage], writes: list[list[AnyMessage]]
+) -> list[AnyMessage]:
+    """**Experimental.** Batch reducer for use with `DeltaChannel`.
+
+    Processes all writes in one pass — dedup by ID, `RemoveMessage`
+    tombstoning — without calling `add_messages`.
+
+    This reducer is batching-invariant, as required by `DeltaChannel`:
+    `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.
+
+    Raw dict / string / tuple inputs are coerced to typed `BaseMessage`
+    objects so that HTTP-driven graphs work without a separate coercion
+    step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`,
+    unknown-id `RemoveMessage` errors, missing-id UUID assignment, and
+    `BaseMessageChunk` conversion are not handled here.
+
+    Example::
+
+        from typing import Annotated
+        from langgraph.channels.delta import DeltaChannel
+        from langgraph.graph.message import _messages_delta_reducer
+
+        class State(TypedDict):
+            messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+    """
+
+    # Each write is either a list of message-likes or a single message-like
+    # (BaseMessage / dict / str / tuple). Only lists flatten; everything
+    # else is one message.
+    flat: list[Any] = []
+    for w in writes:
+        if isinstance(w, list):
+            flat.extend(w)
+        else:
+            flat.append(w)
+    # Steady state: the reducer's own output is already typed, so skip
+    # `convert_to_messages` on state when the first element is a BaseMessage.
+    # Only raw input (initial dicts, deserialized blobs) hits the slow path.
+    if state and isinstance(state[0], BaseMessage):
+        state_msgs = state
+    else:
+        state_msgs = cast("list[AnyMessage]", convert_to_messages(state))
+    msgs = cast("list[AnyMessage]", convert_to_messages(flat))
+
+    index: dict[str, int] = {
+        m.id: i for i, m in enumerate(state_msgs) if m.id is not None
+    }
+    result: list[AnyMessage | None] = list(state_msgs)
+    for msg in msgs:
+        mid = msg.id
+        if mid is None:
+            result.append(msg)
+        elif isinstance(msg, RemoveMessage):
+            if mid in index:
+                result[index[mid]] = None
+                del index[mid]
+        elif mid in index:
+            result[index[mid]] = msg
+        else:
+            index[mid] = len(result)
+            result.append(msg)
+    return [m for m in result if m is not None]
+
+
 # Known role values (OpenAI-style) and type values (LangChain serialisation)
 # that identify a dict as a message. Checked before coercing to BaseMessage so
 # we don't accidentally touch unrelated dicts that happen to have a "role" key.
@@ -423,39 +500,103 @@ def _is_message_dict(item: dict) -> bool:
     return item.get("role") in _MESSAGE_ROLES or item.get("type") in _MESSAGE_TYPES
 
 
-def ensure_message_ids(value: Any) -> None:
-    """Coerce message-like write values to typed BaseMessages with stable IDs.
+def ensure_message_ids(
+    channels: Mapping[str, Any], writes: Sequence[tuple[str, Any]]
+) -> None:
+    """Give the messages written to DeltaChannels stable ids.
 
-    Called in put_writes() before DeltaChannel writes are submitted to the
-    checkpointer. Without this the checkpoint may store raw dicts or id=None
-    BaseMessages; every get_state() replay then produces a different UUID and
-    the same message appears with a different ID in each LangSmith trace.
+    A `DeltaChannel` replays its saved writes on every read, and its reducer
+    may not assign missing ids (`_messages_delta_reducer` doesn't), so a
+    message saved without an id comes back without one, or with a new one on
+    every read, and nothing can target it by id, for example `RemoveMessage`.
 
-    Handles three input shapes:
-    - BaseMessage objects: assign a UUID if id is None.
-    - Dicts with a known "role" (OpenAI-style) or "type" (LangChain format) at
-      the root level: stamp "id" into the dict in-place. The reducer's
-      convert_to_messages call will forward the id to the resulting BaseMessage.
-    - Lists of the above: apply the same logic to each element, replacing dict
-      items with coerced BaseMessages so the shared list reference seen by
-      checkpoint_pending_writes and the background thread both get typed messages.
+    In a channel whose reducer is `_messages_delta_reducer`, every message-like
+    it accepts (a message, a dict, a string, a `(role, content)` tuple; alone,
+    in a list or in an `Overwrite`) becomes a message with an id. Other
+    DeltaChannels may hold other data, so there only message objects get ids,
+    plus message dicts outside an `Overwrite`: one in a list becomes a message.
+    Anything that can't become a message is left as it is, so it fails where
+    it always has: a message dict in a list raises here, anything else when
+    the reducer reads it. An empty id counts as missing.
 
-    Mutating synchronously here (before the background thread is submitted) is
-    safe: the serialised bytes always reflect the post-coercion state.
+    A write that changes is replaced in `writes`: lists and dicts the caller
+    passed in are never changed, only the ids of message objects.
     """
-    if isinstance(value, BaseMessage):
-        if value.id is None:
-            value.id = str(uuid4())
-    elif isinstance(value, dict) and _is_message_dict(value):
-        if not value.get("id"):
-            value["id"] = str(uuid4())
-    elif isinstance(value, list):
-        for i, item in enumerate(value):
-            if isinstance(item, BaseMessage):
-                if item.id is None:
-                    item.id = str(uuid4())
-            elif isinstance(item, dict) and _is_message_dict(item):
-                msg = convert_to_messages([item])[0]
-                if msg.id is None:
-                    msg.id = str(uuid4())
-                value[i] = msg
+    for i, (ch, value) in enumerate(writes):
+        spec = channels.get(ch)
+        if not isinstance(spec, DeltaChannel):
+            continue
+        if spec.reducer is _messages_delta_reducer:
+            new_value = _with_message_ids(value)
+        else:
+            new_value = _with_base_message_ids(value)
+        if new_value is not value:
+            cast(MutableSequence[tuple[str, Any]], writes)[i] = (ch, new_value)
+
+
+def _with_message_ids(value: Any) -> Any:
+    is_overwrite, overwrite_value = _get_overwrite(value)
+    if not is_overwrite:
+        if isinstance(value, list):
+            return _map_messages(value, _list_item_with_id)
+        return _message_with_id(value)
+    if isinstance(overwrite_value, (list, tuple)):
+        items = _map_messages(overwrite_value, _message_with_id)
+        if items is not overwrite_value:
+            return Overwrite(items)
+    return value
+
+
+def _with_base_message_ids(value: Any) -> Any:
+    is_overwrite, overwrite_value = _get_overwrite(value)
+    if is_overwrite:
+        if isinstance(overwrite_value, (list, tuple)):
+            for item in overwrite_value:
+                _stamp_message_id(item)
+        return value
+    if isinstance(value, list):
+        return _map_messages(value, _base_list_item_with_id)
+    if isinstance(value, dict) and _is_message_dict(value) and not value.get("id"):
+        return {**value, "id": str(uuid4())}
+    _stamp_message_id(value)
+    return value
+
+
+def _map_messages(items: list | tuple, fn: Callable[[Any], Any]) -> list | tuple:
+    mapped = [fn(item) for item in items]
+    if all(new is old for new, old in zip(mapped, items)):
+        return items
+    return mapped if isinstance(items, list) else tuple(mapped)
+
+
+def _list_item_with_id(item: Any) -> Any:
+    if isinstance(item, dict) and _is_message_dict(item):
+        return _message_with_id(item, strict=True)
+    return _message_with_id(item)
+
+
+def _base_list_item_with_id(item: Any) -> Any:
+    if isinstance(item, dict) and _is_message_dict(item):
+        return _message_with_id(item, strict=True)
+    _stamp_message_id(item)
+    return item
+
+
+def _message_with_id(item: Any, *, strict: bool = False) -> Any:
+    if isinstance(item, BaseMessage):
+        _stamp_message_id(item)
+        return item
+    try:
+        message = convert_to_messages([item])[0]
+    except Exception:
+        if strict:
+            raise
+        return item
+    if not message.id:
+        message.id = str(uuid4())
+    return message
+
+
+def _stamp_message_id(item: Any) -> None:
+    if isinstance(item, BaseMessage) and item.id is None:
+        item.id = str(uuid4())

libs/langgraph/langgraph/pregel/_retry.py

@@ -7,7 +7,7 @@ import sys
 import threading
 import time
 import weakref
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Callable, Sequence
 from contextlib import suppress
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
@@ -686,8 +686,6 @@ async def arun_with_retry(
     task: PregelExecutableTask,
     retry_policy: Sequence[RetryPolicy] | None,
     stream: bool = False,
-    match_cached_writes: Callable[[], Awaitable[Sequence[PregelExecutableTask]]]
-    | None = None,
     configurable: dict[str, Any] | None = None,
 ) -> None:
     """Run a task asynchronously with retries."""
@@ -711,11 +709,6 @@ async def arun_with_retry(
                 )
             },
         )
-    if match_cached_writes is not None and task.cache_key is not None:
-        for t in await match_cached_writes():
-            if t is task:
-                # if the task is already cached, return
-                return
     while True:
         runtime = config.get(CONF, {}).get(CONFIG_KEY_RUNTIME)
         if isinstance(runtime, Runtime):

libs/langgraph/langgraph/pregel/main.py

@@ -137,6 +137,7 @@ from langgraph.pregel._checkpoint import (
     copy_checkpoint,
     create_checkpoint,
     create_checkpoint_plan_for_update_state_api,
+    delta_channels_overwritten,
     delta_channels_with_pending_writes,
     empty_checkpoint,
     get_updated_channels_from_tasks,
@@ -152,6 +153,7 @@ from langgraph.pregel._loop import (
 from langgraph.pregel._messages import (
     StreamMessagesHandler,
     StreamMessagesHandlerV2,
+    ensure_message_ids,
 )
 from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
 from langgraph.pregel._retry import RetryPolicy
@@ -1766,6 +1768,7 @@ class Pregel(
                     get_next_version=checkpointer.get_next_version,
                     channels_to_snapshot=channels_to_snapshot,
                     stored_versions=checkpoint_previous_versions,
+                    trigger_to_nodes=self.trigger_to_nodes,
                 )
                 next_config = checkpointer.put(
                     checkpoint_config,
@@ -1788,6 +1791,22 @@ class Pregel(
                     )
 
                 if input_writes := deque(map_input(self.input_channels, values)):
+                    _store_or_fork_delta_writes(
+                        checkpointer,
+                        config,
+                        saved,
+                        checkpoint_config,
+                        self.channels,
+                        [
+                            (
+                                str(uuid5(UUID(checkpoint["id"]), INPUT)),
+                                input_writes,
+                                None,
+                            )
+                        ],
+                        fork_pending,
+                        is_first=is_first,
+                    )
                     updated_channels = apply_writes(
                         checkpoint,
                         channels,
@@ -1795,6 +1814,9 @@ class Pregel(
                         checkpointer.get_next_version,
                         self.trigger_to_nodes,
                     )
+                    fork_pending |= delta_channels_overwritten(
+                        self.channels, input_writes
+                    )
 
                     # apply input write to channels
                     next_step = (
@@ -1822,6 +1844,7 @@ class Pregel(
                         get_next_version=checkpointer.get_next_version,
                         channels_to_snapshot=channels_to_snapshot,
                         stored_versions=checkpoint_previous_versions,
+                        trigger_to_nodes=self.trigger_to_nodes,
                     )
                     next_config = checkpointer.put(
                         checkpoint_config,
@@ -1833,13 +1856,6 @@ class Pregel(
                         ),
                     )
 
-                    # store the writes
-                    checkpointer.put_writes(
-                        next_config,
-                        input_writes,
-                        str(uuid5(UUID(checkpoint["id"]), INPUT)),
-                    )
-
                     return patch_checkpoint_map(
                         next_config, saved.metadata if saved else None
                     )
@@ -2033,30 +2049,22 @@ class Pregel(
                     ),
                 )
             updated_channels = get_updated_channels_from_tasks(run_tasks)
-            # The base's other children replay whatever is stored on it, so an
-            # edit of an older checkpoint stores none of its writes there: the
-            # checkpoint written here carries them, its delta channels
-            # snapshotted. Later supersteps address the checkpoint just written.
-            if (
-                is_first
-                and saved is not None
-                and checkpoint_superseded(checkpointer, config, saved)
-            ):
-                fork_pending.update(
-                    ch
-                    for ch in updated_channels
-                    if isinstance(self.channels.get(ch), DeltaChannel)
-                )
-            elif saved is not None:
-                for task_id, task in zip(run_task_ids, run_tasks):
-                    channel_writes = [w for w in task.writes if w[0] != PUSH]
-                    if channel_writes:
-                        checkpointer.put_writes(
-                            checkpoint_config,
-                            channel_writes,
-                            task_id,
-                            **_task_path_kwarg(checkpointer.put_writes, task),
-                        )
+            fork_pending |= delta_channels_overwritten(
+                self.channels, (w for t in run_tasks for w in t.writes)
+            )
+            _store_or_fork_delta_writes(
+                checkpointer,
+                config,
+                saved,
+                checkpoint_config,
+                self.channels,
+                [
+                    (task_id, task.writes, task)
+                    for task_id, task in zip(run_task_ids, run_tasks)
+                ],
+                fork_pending,
+                is_first=is_first,
+            )
             apply_writes(
                 checkpoint,
                 channels,
@@ -2086,6 +2094,7 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
                 stored_versions=checkpoint_previous_versions,
+                trigger_to_nodes=self.trigger_to_nodes,
             )
             next_config = checkpointer.put(
                 checkpoint_config,
@@ -2258,6 +2267,7 @@ class Pregel(
                     get_next_version=checkpointer.get_next_version,
                     channels_to_snapshot=channels_to_snapshot,
                     stored_versions=checkpoint_previous_versions,
+                    trigger_to_nodes=self.trigger_to_nodes,
                 )
                 next_config = await checkpointer.aput(
                     checkpoint_config,
@@ -2280,6 +2290,22 @@ class Pregel(
                     )
 
                 if input_writes := deque(map_input(self.input_channels, values)):
+                    await _astore_or_fork_delta_writes(
+                        checkpointer,
+                        config,
+                        saved,
+                        checkpoint_config,
+                        self.channels,
+                        [
+                            (
+                                str(uuid5(UUID(checkpoint["id"]), INPUT)),
+                                input_writes,
+                                None,
+                            )
+                        ],
+                        fork_pending,
+                        is_first=is_first,
+                    )
                     updated_channels = apply_writes(
                         checkpoint,
                         channels,
@@ -2287,6 +2313,9 @@ class Pregel(
                         checkpointer.get_next_version,
                         self.trigger_to_nodes,
                     )
+                    fork_pending |= delta_channels_overwritten(
+                        self.channels, input_writes
+                    )
 
                     # apply input write to channels
                     next_step = (
@@ -2314,6 +2343,7 @@ class Pregel(
                         get_next_version=checkpointer.get_next_version,
                         channels_to_snapshot=channels_to_snapshot,
                         stored_versions=checkpoint_previous_versions,
+                        trigger_to_nodes=self.trigger_to_nodes,
                     )
                     next_config = await checkpointer.aput(
                         checkpoint_config,
@@ -2325,13 +2355,6 @@ class Pregel(
                         ),
                     )
 
-                    # store the writes
-                    await checkpointer.aput_writes(
-                        next_config,
-                        input_writes,
-                        str(uuid5(UUID(checkpoint["id"]), INPUT)),
-                    )
-
                     return patch_checkpoint_map(
                         next_config, saved.metadata if saved else None
                     )
@@ -2524,30 +2547,22 @@ class Pregel(
                     ),
                 )
             updated_channels = get_updated_channels_from_tasks(run_tasks)
-            # The base's other children replay whatever is stored on it, so an
-            # edit of an older checkpoint stores none of its writes there: the
-            # checkpoint written here carries them, its delta channels
-            # snapshotted. Later supersteps address the checkpoint just written.
-            if (
-                is_first
-                and saved is not None
-                and await acheckpoint_superseded(checkpointer, config, saved)
-            ):
-                fork_pending.update(
-                    ch
-                    for ch in updated_channels
-                    if isinstance(self.channels.get(ch), DeltaChannel)
-                )
-            elif saved is not None:
-                for task_id, task in zip(run_task_ids, run_tasks):
-                    channel_writes = [w for w in task.writes if w[0] != PUSH]
-                    if channel_writes:
-                        await checkpointer.aput_writes(
-                            checkpoint_config,
-                            channel_writes,
-                            task_id,
-                            **_task_path_kwarg(checkpointer.aput_writes, task),
-                        )
+            fork_pending |= delta_channels_overwritten(
+                self.channels, (w for t in run_tasks for w in t.writes)
+            )
+            await _astore_or_fork_delta_writes(
+                checkpointer,
+                config,
+                saved,
+                checkpoint_config,
+                self.channels,
+                [
+                    (task_id, task.writes, task)
+                    for task_id, task in zip(run_task_ids, run_tasks)
+                ],
+                fork_pending,
+                is_first=is_first,
+            )
             apply_writes(
                 checkpoint,
                 channels,
@@ -2577,6 +2592,7 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
                 stored_versions=checkpoint_previous_versions,
+                trigger_to_nodes=self.trigger_to_nodes,
             )
             next_config = await checkpointer.aput(
                 checkpoint_config,
@@ -3038,7 +3054,7 @@ class Pregel(
                 # with channel updates applied only at the transition between steps.
                 while loop.tick():
                     for task in loop.match_cached_writes():
-                        loop.output_writes(task.id, task.writes, cached=True)
+                        loop.put_writes(task.id, task.writes, cached=True)
                     for _ in runner.tick(
                         [t for t in loop.tasks.values() if not t.writes],
                         timeout=self.step_timeout,
@@ -3509,7 +3525,7 @@ class Pregel(
                 try:
                     while loop.tick():
                         for task in await loop.amatch_cached_writes():
-                            loop.output_writes(task.id, task.writes, cached=True)
+                            loop.put_writes(task.id, task.writes, cached=True)
                         async for _ in runner.atick(
                             [t for t in loop.tasks.values() if not t.writes],
                             timeout=self.step_timeout,
@@ -4255,6 +4271,87 @@ def _task_path_kwarg(put_writes: Callable[..., Any], task: PregelTaskWrites) ->
     return {"task_path": task_path_str(task.path)}
 
 
+_UpdateWrites = Sequence[tuple[str, Sequence[tuple[str, Any]], PregelTaskWrites | None]]
+
+
+def _store_or_fork_delta_writes(
+    checkpointer: BaseCheckpointSaver,
+    config: RunnableConfig,
+    saved: CheckpointTuple | None,
+    checkpoint_config: RunnableConfig,
+    channels: Mapping[str, BaseChannel | ManagedValueSpec],
+    writes: _UpdateWrites,
+    fork_pending: set[str],
+    *,
+    is_first: bool,
+) -> None:
+    """Save an update's writes where its DeltaChannels will read them.
+
+    A DeltaChannel rebuilds its value from the writes saved on a checkpoint's
+    ancestors, so the writes go on the checkpoint the update builds on. If the
+    thread already moved past that checkpoint, its other children would read
+    them too, so the new checkpoint snapshots those channels instead. `writes`
+    holds `(task_id, writes, task)` per task; `task` is `None` for input. A
+    task's `PUSH` writes are left out: the update saves them on the checkpoint
+    it creates. The messages in `writes` get their ids first, in place, as the
+    loop's `put_writes` does, so pass the writes the update then applies.
+    """
+    for _, task_writes, _ in writes:
+        ensure_message_ids(channels, task_writes)
+    if saved is None:
+        return
+    if is_first and checkpoint_superseded(checkpointer, config, saved):
+        fork_pending.update(
+            ch
+            for _, task_writes, _ in writes
+            for ch, _ in task_writes
+            if isinstance(channels.get(ch), DeltaChannel)
+        )
+        return
+    for task_id, task_writes, task in writes:
+        if channel_writes := [w for w in task_writes if w[0] != PUSH]:
+            checkpointer.put_writes(
+                checkpoint_config,
+                channel_writes,
+                task_id,
+                **(_task_path_kwarg(checkpointer.put_writes, task) if task else {}),
+            )
+
+
+async def _astore_or_fork_delta_writes(
+    checkpointer: BaseCheckpointSaver,
+    config: RunnableConfig,
+    saved: CheckpointTuple | None,
+    checkpoint_config: RunnableConfig,
+    channels: Mapping[str, BaseChannel | ManagedValueSpec],
+    writes: _UpdateWrites,
+    fork_pending: set[str],
+    *,
+    is_first: bool,
+) -> None:
+    """Async `_store_or_fork_delta_writes`."""
+    for _, task_writes, _ in writes:
+        ensure_message_ids(channels, task_writes)
+    if saved is None:
+        return
+    if is_first and await acheckpoint_superseded(checkpointer, config, saved):
+        fork_pending.update(
+            ch
+            for _, task_writes, _ in writes
+            for ch, _ in task_writes
+            if isinstance(channels.get(ch), DeltaChannel)
+        )
+        return
+    for task_id, task_writes, task in writes:
+        if channel_writes := [w for w in task_writes if w[0] != PUSH]:
+            await checkpointer.aput_writes(
+                checkpoint_config,
+                channel_writes,
+                task_id,
+                **(_task_path_kwarg(checkpointer.aput_writes, task) if task else {}),
+            )
+
+
 def _update_task_id(checkpoint_id: str, i: int) -> str:
     """Task id for the `i`th update of a superstep that has no task to reuse.
 

libs/langgraph/langgraph/pregel/remote.py

@@ -29,12 +29,10 @@ from langgraph_sdk.client import (
     get_client,
     get_sync_client,
 )
-from langgraph_sdk.errors import NotFoundError
 from langgraph_sdk.schema import (
     Checkpoint,
     Context,
     QueryParamTypes,
-    Thread,
     ThreadState,
 )
 from langgraph_sdk.schema import (
@@ -51,10 +49,6 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
-    CONFIG_KEY_RESUME_MAP,
-    CONFIG_KEY_RESUME_MAP_SENT,
-    CONFIG_KEY_RESUMING,
-    CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_STREAM,
     CONFIG_KEY_TASK_ID,
     INTERRUPT,
@@ -87,9 +81,6 @@ _CONF_DROPLIST = frozenset(
         CONFIG_KEY_CHECKPOINT_ID,
         CONFIG_KEY_CHECKPOINT_NS,
         CONFIG_KEY_TASK_ID,
-        CONFIG_KEY_RESUME_MAP,
-        CONFIG_KEY_RESUME_MAP_SENT,
-        CONFIG_KEY_RESUMING,
     ),
 )
 
@@ -442,43 +433,6 @@ class RemoteGraph(PregelProtocol):
 
         return sanitized
 
-    @staticmethod
-    def _is_resuming(config: RunnableConfig) -> bool:
-        """Whether a parent graph is re-running this node to resume or retry it."""
-        conf = config.get(CONF, {})
-        return CONFIG_KEY_SCRATCHPAD in conf and bool(conf.get(CONFIG_KEY_RESUMING))
-
-    @staticmethod
-    def _resume_command(
-        config: RunnableConfig, thread: Thread | None
-    ) -> CommandSDK | None:
-        """Build a command that resumes an interrupted remote thread."""
-        if thread is None or thread["status"] != "interrupted":
-            return None
-        pending = [
-            Interrupt(**i)
-            for ints in (thread.get("interrupts") or {}).values()
-            for i in ints
-        ]
-        if not pending:
-            return None
-        conf = config[CONF]
-        if resume_map := conf.get(CONFIG_KEY_RESUME_MAP):
-            sent: set[str] = conf.get(CONFIG_KEY_RESUME_MAP_SENT, set())
-            # Do not forward the whole map: it can hold answers for other remotes.
-            # Send each answer once per run: a later interrupt in the same remote
-            # node reuses the id, so a retry would apply the old answer to it.
-            if scoped := {
-                i.id: resume_map[i.id]
-                for i in pending
-                if i.id in resume_map and i.id not in sent
-            }:
-                sent.update(scoped)
-                return {"resume": scoped}
-        elif (resume := conf[CONFIG_KEY_SCRATCHPAD].get_null_resume(True)) is not None:
-            return {"resume": resume}
-        raise GraphInterrupt(pending)
-
     def get_state(
         self,
         config: RunnableConfig,
@@ -840,21 +794,13 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
-        command: CommandSDK | None = None
         if isinstance(input, Command):
-            command = cast(CommandSDK, asdict(input))
-        elif thread_id is not None and self._is_resuming(merged_config):
-            try:
-                thread = sync_client.threads.get(thread_id, headers=headers)
-            except NotFoundError:
-                thread = None
-            command = self._resume_command(merged_config, thread)
-        if command is not None:
+            command: CommandSDK | None = cast(CommandSDK, asdict(input))
             input = None
+        else:
+            command = None
+        thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
 
-        interrupted = False
-        interrupts: dict[str, Interrupt] = {}
         for chunk in sync_client.runs.stream(
             thread_id=thread_id,
             assistant_id=self.assistant_id,
@@ -886,24 +832,16 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
-            is_interrupt = (
-                chunk.event.startswith("updates")
-                and isinstance(chunk.data, dict)
-                and INTERRUPT in chunk.data
-            )
-            if interrupted and not is_interrupt and not chunk.event.startswith("error"):
-                continue
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
-            # collect interrupts or raise errors
-            if is_interrupt:
-                if caller_ns:
-                    interrupted = True
-                    for i in chunk.data[INTERRUPT]:
-                        item = Interrupt(**i)
-                        interrupts.setdefault(item.id, item)
-                    continue
+            # raise interrupt or errors
+            if chunk.event.startswith("updates"):
+                if isinstance(chunk.data, dict) and INTERRUPT in chunk.data:
+                    if caller_ns:
+                        raise GraphInterrupt(
+                            [Interrupt(**i) for i in chunk.data[INTERRUPT]]
+                        )
             elif chunk.event.startswith("error"):
                 raise RemoteException(chunk.data)
             # filter for what was actually requested
@@ -936,8 +874,6 @@ class RemoteGraph(PregelProtocol):
                 yield chunk.data
             else:
                 yield chunk
-        if interrupted:
-            raise GraphInterrupt(list(interrupts.values()))
 
     @overload
     def astream(
@@ -1013,21 +949,13 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
-        command: CommandSDK | None = None
         if isinstance(input, Command):
-            command = cast(CommandSDK, asdict(input))
-        elif thread_id is not None and self._is_resuming(merged_config):
-            try:
-                thread = await client.threads.get(thread_id, headers=headers)
-            except NotFoundError:
-                thread = None
-            command = self._resume_command(merged_config, thread)
-        if command is not None:
+            command: CommandSDK | None = cast(CommandSDK, asdict(input))
             input = None
+        else:
+            command = None
+        thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
 
-        interrupted = False
-        interrupts: dict[str, Interrupt] = {}
         async for chunk in client.runs.stream(
             thread_id=thread_id,
             assistant_id=self.assistant_id,
@@ -1059,24 +987,16 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
-            is_interrupt = (
-                chunk.event.startswith("updates")
-                and isinstance(chunk.data, dict)
-                and INTERRUPT in chunk.data
-            )
-            if interrupted and not is_interrupt and not chunk.event.startswith("error"):
-                continue
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
-            # collect interrupts or raise errors
-            if is_interrupt:
-                if caller_ns:
-                    interrupted = True
-                    for i in chunk.data[INTERRUPT]:
-                        item = Interrupt(**i)
-                        interrupts.setdefault(item.id, item)
-                    continue
+            # raise interrupt or errors
+            if chunk.event.startswith("updates"):
+                if isinstance(chunk.data, dict) and INTERRUPT in chunk.data:
+                    if caller_ns:
+                        raise GraphInterrupt(
+                            [Interrupt(**i) for i in chunk.data[INTERRUPT]]
+                        )
             elif chunk.event.startswith("error"):
                 raise RemoteException(chunk.data)
             # filter for what was actually requested
@@ -1109,8 +1029,6 @@ class RemoteGraph(PregelProtocol):
                 yield chunk.data
             else:
                 yield chunk
-        if interrupted:
-            raise GraphInterrupt(list(interrupts.values()))
 
     def stream_events(
         self,

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.14"
+version = "1.2.15"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/test_delta_channel_exit_mode.py

@@ -521,10 +521,11 @@ def test_addressed_resume_keeps_a_rerun_tasks_write_to_a_new_channel(
     graph = builder.compile(checkpointer=sync_checkpointer)
     config = {"configurable": {"thread_id": "t"}}
     graph.invoke(_both("in"), config, durability=durability)
-    head = graph.get_state(config).config
+    paused = graph.get_state(config).config
+    graph.update_state(config, None)
 
     live = graph.invoke(
-        Command(resume="yes", update={"flag": True}), head, durability=durability
+        Command(resume="yes", update={"flag": True}), paused, durability=durability
     )
 
     state = graph.get_state(config)
@@ -625,3 +626,33 @@ def test_exit_resume_retried_after_its_final_checkpoint_fails_reruns_the_resumed
     state = graph.get_state(config)
     assert state.values["log"] == state.values["plain"]
     assert sorted(state.values["plain"]) == ["ask", "done", "in"]
+
+
+class _TurnState(_ResumeState):
+    turn: int
+
+
+@pytest.mark.parametrize("exit_runs", [1, 2])
+def test_exit_runs_from_a_checkpoint_the_thread_moved_past_leave_its_newer_branch_alone(
+    sync_checkpointer: BaseCheckpointSaver, exit_runs: int
+) -> None:
+    builder = StateGraph(_TurnState)
+    builder.add_node("a", lambda state: _both(f"a{state['turn']}"))
+    builder.add_node("b", lambda state: _both(f"b{state['turn']}"))
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"turn": 1, **_both("in1")}, config)
+    end_of_turn_1 = graph.get_state(config).config
+    graph.invoke({"turn": 2, **_both("in2")}, config)
+
+    for turn in range(3, 3 + exit_runs):
+        graph.invoke(
+            {"turn": turn, **_both(f"in{turn}")}, end_of_turn_1, durability="exit"
+        )
+
+    history = list(graph.get_state_history(config))
+    assert [s.values.get("log", []) for s in history] == [
+        s.values.get("plain", []) for s in history
+    ]

libs/langgraph/tests/test_delta_channel_id_stability.py

@@ -16,7 +16,11 @@ from langgraph.checkpoint.memory import InMemorySaver
 from typing_extensions import TypedDict
 
 from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.last_value import LastValue
 from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.types import Command, Durability, Overwrite
 
 pytestmark = pytest.mark.anyio
 
@@ -34,18 +38,19 @@ def _append_reducer(
     return result
 
 
-def _build_graph(checkpointer: Any) -> Any:
+def _build_graph(
+    checkpointer: Any,
+    *,
+    reducer: Any = _append_reducer,
+    write: Any = lambda: [AIMessage(content="reply", id="ai-1")],
+) -> Any:
     State = TypedDict(  # noqa: UP013
         "State",
-        {
-            "messages": Annotated[
-                list, DeltaChannel(_append_reducer, snapshot_frequency=50)
-            ]
-        },
+        {"messages": Annotated[list, DeltaChannel(reducer, snapshot_frequency=50)]},
     )  # type: ignore[call-overload]
 
     def agent(state: dict) -> dict:
-        return {"messages": [AIMessage(content="reply", id="ai-1")]}
+        return {"messages": write()}
 
     return (
         StateGraph(State)
@@ -138,3 +143,193 @@ def test_delta_channel_dict_style_message_gets_stable_id() -> None:
     assert len(set(ids)) == 1, (
         f"dict-style HumanMessage id must be stable across get_state() calls; got {ids}"
     )
+
+
+@pytest.mark.parametrize("container", [list, tuple])
+def test_delta_channel_message_in_an_overwrite_gets_stable_id(container: type) -> None:
+    graph = _build_graph(InMemorySaver())
+    config = {"configurable": {"thread_id": "overwrite-id-stability"}}
+
+    graph.invoke(
+        {"messages": Overwrite(container([HumanMessage(content="hello")]))}, config
+    )
+
+    ids = [
+        next(
+            m.id
+            for m in graph.get_state(config).values["messages"]
+            if isinstance(m, HumanMessage)
+        )
+        for _ in range(3)
+    ]
+
+    assert ids[0] is not None, "a message inside an Overwrite should get a UUID"
+    assert len(set(ids)) == 1, (
+        f"its id must be stable across get_state() calls; got {ids}"
+    )
+
+
+# Writes of "hello" in every shape `_messages_delta_reducer` accepts.
+MESSAGE_SHAPES = [
+    pytest.param(lambda: [HumanMessage("hello")], id="message"),
+    pytest.param(lambda: HumanMessage("hello"), id="bare message"),
+    pytest.param(lambda: [{"role": "user", "content": "hello"}], id="dict"),
+    pytest.param(lambda: {"type": "human", "content": "hello"}, id="bare dict"),
+    pytest.param(
+        lambda: [{"role": "user", "content": "hello", "id": ""}], id="dict, empty id"
+    ),
+    pytest.param(lambda: [("user", "hello")], id="tuple"),
+    pytest.param(lambda: ("user", "hello"), id="bare tuple"),
+    pytest.param(lambda: ["hello"], id="string"),
+    pytest.param(lambda: "hello", id="bare string"),
+    pytest.param(lambda: Overwrite(["hello"]), id="Overwrite"),
+    pytest.param(lambda: Overwrite(("hello",)), id="Overwrite of a tuple"),
+    pytest.param(lambda: {"__overwrite__": [("user", "hello")]}, id="Overwrite dict"),
+]
+
+
+def _hello_ids(graph: Any, config: dict) -> list[str | None]:
+    return [
+        next(
+            m.id
+            for m in graph.get_state(config).values["messages"]
+            if m.content == "hello"
+        )
+        for _ in range(2)
+    ]
+
+
+@pytest.mark.parametrize("write", MESSAGE_SHAPES)
+def test_input_message_of_any_shape_gets_stable_id(write: Any) -> None:
+    graph = _build_graph(InMemorySaver(), reducer=_messages_delta_reducer)
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke({"messages": write()}, config)
+
+    first, second = _hello_ids(graph, config)
+    assert first, "the message should get a UUID"
+    assert first == second, "its id must be the same on every read"
+
+
+@pytest.mark.parametrize("write", MESSAGE_SHAPES)
+def test_node_message_of_any_shape_gets_stable_id(write: Any) -> None:
+    graph = _build_graph(InMemorySaver(), reducer=_messages_delta_reducer, write=write)
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke({"messages": []}, config)
+
+    first, second = _hello_ids(graph, config)
+    assert first, "the message should get a UUID"
+    assert first == second, "its id must be the same on every read"
+
+
+def test_a_list_written_twice_adds_its_messages_twice() -> None:
+    turn = [("user", "hello")]
+    graph = _build_graph(InMemorySaver(), reducer=_messages_delta_reducer)
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke({"messages": turn}, config)
+    graph.invoke({"messages": turn}, config)
+
+    hellos = [
+        m.id for m in graph.get_state(config).values["messages"] if m.content == "hello"
+    ]
+    assert len(set(hellos)) == 2, f"each write should add a message; got {hellos}"
+    assert turn == [("user", "hello")], "the caller's list should be left as it was"
+
+
+def test_a_forked_command_update_gets_a_stable_id() -> None:
+    graph = _build_graph(InMemorySaver(), reducer=_messages_delta_reducer)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("a", id="a")]}, config)
+    before_agent = next(
+        s.config for s in graph.get_state_history(config) if s.next == ("agent",)
+    )
+
+    graph.invoke(Command(update={"messages": "hello"}), before_agent)
+
+    first, second = _hello_ids(graph, config)
+    assert first, "the message should get a UUID"
+    assert first == second, "its id must be the same on every read"
+
+
+def test_a_message_dict_the_reducer_rejects_fails_the_run_before_it_is_saved() -> None:
+    graph = _build_graph(
+        InMemorySaver(),
+        reducer=_messages_delta_reducer,
+        write=lambda: [{"role": "tool", "content": "x"}],
+    )
+    config = {"configurable": {"thread_id": "t"}}
+
+    with pytest.raises(KeyError):
+        graph.invoke({"messages": [HumanMessage("a", id="a")]}, config)
+
+    assert [m.id for m in graph.get_state(config).values["messages"]] == ["a"]
+
+
+def _pregel_graph(seen: list[list[str | None]]) -> Any:
+    def look(state: dict) -> list:
+        seen.append([m.id for m in state["messages"]])
+        return []
+
+    node = (
+        NodeBuilder()
+        .subscribe_to("go", read=False)
+        .read_from("messages")
+        .do(look)
+        .write_to("messages")
+    )
+    return Pregel(
+        nodes={"look": node},
+        channels={
+            "messages": DeltaChannel(_messages_delta_reducer),
+            "go": LastValue(int),
+        },
+        input_channels=["messages", "go"],
+        output_channels=["messages"],
+        checkpointer=InMemorySaver(),
+    )
+
+
+def test_pregel_input_message_gets_stable_id_on_a_new_thread(
+    durability: Durability,
+) -> None:
+    graph = _pregel_graph([])
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke(
+        {"messages": [HumanMessage("hello")], "go": 1}, config, durability=durability
+    )
+
+    first, second = _hello_ids(graph, config)
+    assert first, "the message should get a UUID"
+    assert first == second, "its id must be the same on every read"
+
+
+def test_a_node_reads_the_pregel_input_message_ids_that_are_saved() -> None:
+    seen: list[list[str | None]] = []
+    graph = _pregel_graph(seen)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("first", id="first")], "go": 1}, config)
+
+    graph.invoke({"messages": [{"role": "user", "content": "hello"}], "go": 1}, config)
+
+    saved = [m.id for m in graph.get_state(config).values["messages"]]
+    assert None not in saved
+    assert seen[-1] == saved, "the node must see the ids that are saved"
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        pytest.param({"type": "tool", "name": "search"}, id="not a message"),
+        pytest.param({"role": "user", "content": "x"}, id="message-shaped"),
+    ],
+)
+def test_other_delta_channels_keep_the_dicts_in_an_overwrite(value: dict) -> None:
+    graph = _build_graph(InMemorySaver(), write=lambda: Overwrite([dict(value)]))
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke({"messages": []}, config)
+
+    assert graph.get_state(config).values["messages"] == [value]

libs/langgraph/tests/test_delta_channel_node_cache.py

@@ -0,0 +1,122 @@
+"""A node served from the node cache saves its writes like a node that ran,
+without writing them back to the cache, and a node that fails isn't cached."""
+
+import operator
+from typing import Annotated, Any
+
+import pytest
+from langgraph.cache.memory import InMemoryCache
+from langgraph.checkpoint.memory import InMemorySaver
+from typing_extensions import TypedDict
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import START, StateGraph
+from langgraph.types import CachePolicy, Durability
+
+pytestmark = pytest.mark.anyio
+
+INPUT = {"log": [], "plain": []}
+
+
+def _append(current: list, writes: list) -> list:
+    return [*current, *(item for write in writes for item in write)]
+
+
+class _State(TypedDict):
+    log: Annotated[list, DeltaChannel(_append)]
+    plain: Annotated[list, operator.add]
+
+
+class _CountsSets(InMemoryCache):
+    sets = 0
+
+    def set(self, keys: Any) -> None:
+        self.sets += 1
+        super().set(keys)
+
+
+def _a_then_cached_b_then_c(runs: list[str], cache: InMemoryCache) -> Any:
+    def node(name: str) -> Any:
+        def run(state: _State) -> dict:
+            runs.append(name)
+            return {"log": [name], "plain": [name]}
+
+        return run
+
+    builder = StateGraph(_State)
+    builder.add_node("a", node("a"))
+    builder.add_node("b", node("b"), cache_policy=CachePolicy())
+    builder.add_node("c", node("c"))
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("b", "c")
+    return builder.compile(checkpointer=InMemorySaver(), cache=cache)
+
+
+def test_a_cache_hit_saves_its_writes_without_caching_them_again(
+    durability: Durability,
+) -> None:
+    runs: list[str] = []
+    cache = _CountsSets()
+    graph = _a_then_cached_b_then_c(runs, cache)
+    graph.invoke(INPUT, {"configurable": {"thread_id": "1"}}, durability=durability)
+    config = {"configurable": {"thread_id": "2"}}
+
+    graph.invoke(INPUT, config, durability=durability)
+
+    assert runs == ["a", "b", "c", "a", "c"]
+    assert cache.sets == 1, "the cache hit was written back to the cache"
+    for state in graph.get_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+
+async def test_a_cache_hit_saves_its_writes_without_caching_them_again_async(
+    durability: Durability,
+) -> None:
+    runs: list[str] = []
+    cache = _CountsSets()
+    graph = _a_then_cached_b_then_c(runs, cache)
+    await graph.ainvoke(
+        INPUT, {"configurable": {"thread_id": "1"}}, durability=durability
+    )
+    config = {"configurable": {"thread_id": "2"}}
+
+    await graph.ainvoke(INPUT, config, durability=durability)
+
+    assert runs == ["a", "b", "c", "a", "c"]
+    assert cache.sets == 1, "the cache hit was written back to the cache"
+    async for state in graph.aget_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+
+def _cached_node_that_fails(runs: list[str]) -> Any:
+    def fail(state: _State) -> dict:
+        runs.append("b")
+        raise ValueError("b failed")
+
+    builder = StateGraph(_State)
+    builder.add_node("b", fail, cache_policy=CachePolicy())
+    builder.add_edge(START, "b")
+    return builder.compile(checkpointer=InMemorySaver(), cache=InMemoryCache())
+
+
+def test_a_node_that_fails_is_not_cached() -> None:
+    runs: list[str] = []
+    graph = _cached_node_that_fails(runs)
+
+    for thread in ("1", "2"):
+        with pytest.raises(ValueError, match="b failed"):
+            graph.invoke(INPUT, {"configurable": {"thread_id": thread}})
+
+    assert runs == ["b", "b"]
+
+
+async def test_a_node_that_fails_is_not_cached_async() -> None:
+    runs: list[str] = []
+    graph = _cached_node_that_fails(runs)
+
+    for thread in ("1", "2"):
+        with pytest.raises(ValueError, match="b failed"):
+            await graph.ainvoke(INPUT, {"configurable": {"thread_id": thread}})
+
+    assert runs == ["b", "b"]

libs/langgraph/tests/test_delta_channel_run_input.py

@@ -0,0 +1,103 @@
+"""A run's input to a DeltaChannel input channel reads back on the checkpoints
+built from it, and on no others."""
+
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+
+from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.last_value import LastValue
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.types import Durability
+
+pytestmark = pytest.mark.anyio
+
+
+def _sorted_extend(current: list, writes: list) -> list:
+    return sorted([*current, *(item for write in writes for item in write)])
+
+
+def _delta_input_graph() -> Pregel:
+    node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log", "plain")
+    return Pregel(
+        nodes={"n": node},
+        channels={
+            "log": DeltaChannel(_sorted_extend),
+            "plain": BinaryOperatorAggregate(list, lambda a, b: sorted(a + b)),
+            "go": LastValue(int),
+        },
+        input_channels=["log", "plain", "go"],
+        output_channels=["log", "plain"],
+        checkpointer=InMemorySaver(),
+    )
+
+
+def test_each_run_input_reads_back_on_its_own_checkpoints(
+    durability: Durability,
+) -> None:
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke({"log": [0], "plain": [0], "go": 1}, config, durability=durability)
+    graph.invoke({"log": [5], "plain": [5], "go": 1}, config, durability=durability)
+
+    for state in graph.get_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+
+async def test_each_run_input_reads_back_on_its_own_checkpoints_async(
+    durability: Durability,
+) -> None:
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+
+    await graph.ainvoke(
+        {"log": [0], "plain": [0], "go": 1}, config, durability=durability
+    )
+    await graph.ainvoke(
+        {"log": [5], "plain": [5], "go": 1}, config, durability=durability
+    )
+
+    async for state in graph.aget_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+
+OTHER_BRANCH_INPUTS = pytest.mark.parametrize(
+    "other_branch_input",
+    [{"go": 1}, {"log": [5], "plain": [5], "go": 1}],
+    ids=["other-branch-without-delta-input", "other-branch-with-delta-input"],
+)
+
+
+@OTHER_BRANCH_INPUTS
+def test_run_input_from_an_older_checkpoint_stays_out_of_its_other_branch(
+    durability: Durability, other_branch_input: dict
+) -> None:
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"go": 1}, config, durability=durability)
+    older = graph.get_state(config).config
+    graph.invoke(other_branch_input, config, durability=durability)
+
+    graph.invoke({"log": [7], "plain": [7], "go": 1}, older, durability=durability)
+
+    for state in graph.get_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+
+@OTHER_BRANCH_INPUTS
+async def test_run_input_from_an_older_checkpoint_stays_out_of_its_other_branch_async(
+    durability: Durability, other_branch_input: dict
+) -> None:
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"go": 1}, config, durability=durability)
+    older = (await graph.aget_state(config)).config
+    await graph.ainvoke(other_branch_input, config, durability=durability)
+
+    await graph.ainvoke(
+        {"log": [7], "plain": [7], "go": 1}, older, durability=durability
+    )
+
+    async for state in graph.aget_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])

libs/langgraph/tests/test_delta_channel_seal_subscribers.py

@@ -0,0 +1,78 @@
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.last_value import LastValue
+from langgraph.pregel import NodeBuilder, Pregel
+
+pytestmark = pytest.mark.anyio
+
+CONFIG = {"configurable": {"thread_id": "t"}}
+
+
+def _extend(current: list, writes: list) -> list:
+    return [*current, *(item for write in writes for item in write)]
+
+
+def _graph(reads: list) -> Pregel:
+    writer = NodeBuilder().subscribe_only("a").do(lambda _: [1]).write_to("d")
+    reader = NodeBuilder().subscribe_only("d").do(lambda d: reads.append(list(d)))
+    return Pregel(
+        nodes={"writer": writer, "reader": reader},
+        channels={"a": LastValue(str), "d": DeltaChannel(_extend)},
+        input_channels="a",
+        output_channels=["d"],
+        checkpointer=InMemorySaver(),
+    )
+
+
+def test_an_input_update_from_before_the_first_write_starts_only_the_writer() -> None:
+    reads: list = []
+    graph = _graph(reads)
+    graph.invoke("go", CONFIG)
+    first = next(
+        s.config for s in graph.get_state_history(CONFIG) if s.metadata["step"] == -1
+    )
+
+    fork = graph.update_state(first, {"a": "go"}, as_node="__input__")
+
+    assert graph.get_state(fork).next == ("writer",)
+    graph.invoke(None, fork)
+    assert reads == [[1], [1]]
+
+
+def test_a_replay_from_before_the_first_write_forks_with_only_the_writer_next() -> None:
+    reads: list = []
+    graph = _graph(reads)
+    graph.invoke("go", CONFIG)
+    first = next(
+        s.config for s in graph.get_state_history(CONFIG) if s.metadata["step"] == -1
+    )
+
+    graph.invoke(None, first, durability="sync")
+
+    fork = next(
+        s for s in graph.get_state_history(CONFIG) if s.metadata["source"] == "fork"
+    )
+    assert fork.next == ("writer",)
+    graph.invoke(None, fork.config)
+    assert reads == [[1], [1], [1]]
+
+
+async def test_an_ainput_update_from_before_the_first_write_starts_only_the_writer() -> (
+    None
+):
+    reads: list = []
+    graph = _graph(reads)
+    await graph.ainvoke("go", CONFIG)
+    first = [
+        s.config
+        async for s in graph.aget_state_history(CONFIG)
+        if s.metadata["step"] == -1
+    ][0]
+
+    fork = await graph.aupdate_state(first, {"a": "go"}, as_node="__input__")
+
+    assert (await graph.aget_state(fork)).next == ("writer",)
+    await graph.ainvoke(None, fork)
+    assert reads == [[1], [1]]

libs/langgraph/tests/test_delta_channel_to_plain.py

@@ -0,0 +1,148 @@
+import operator
+from typing import Annotated, Any
+from unittest.mock import patch
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from typing_extensions import TypedDict
+
+from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer, add_messages
+from langgraph.pregel._checkpoint import channels_from_checkpoint, empty_checkpoint
+from langgraph.types import Overwrite
+
+pytestmark = pytest.mark.anyio
+
+
+class DeltaState(TypedDict):
+    messages: Annotated[
+        list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3)
+    ]
+
+
+class PlainState(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+def _app(schema: type, checkpointer: BaseCheckpointSaver) -> Any:
+    builder = StateGraph(schema)
+    builder.add_node(
+        "chat",
+        lambda state: {"messages": [AIMessage(f"reply {len(state['messages'])}")]},
+    )
+    builder.add_edge(START, "chat")
+    builder.add_edge("chat", END)
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _delta_thread(checkpointer: BaseCheckpointSaver) -> RunnableConfig:
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    delta = _app(DeltaState, checkpointer)
+    for i in range(4):
+        delta.invoke({"messages": [HumanMessage(f"q{i}")]}, config)
+    assert any(
+        isinstance(t.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+        for t in checkpointer.list(config)
+    ), "the thread should have a snapshot to read through"
+    return config
+
+
+def test_plain_reducer_reads_every_delta_written_checkpoint(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _delta_thread(sync_checkpointer)
+    delta = _app(DeltaState, sync_checkpointer)
+    plain = _app(PlainState, sync_checkpointer)
+
+    assert [s.values for s in plain.get_state_history(config)] == [
+        s.values for s in delta.get_state_history(config)
+    ]
+
+
+def test_plain_reducer_continues_delta_written_thread(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _delta_thread(sync_checkpointer)
+    delta = _app(DeltaState, sync_checkpointer)
+    before = delta.get_state(config).values["messages"]
+
+    after = _app(PlainState, sync_checkpointer).invoke(
+        {"messages": [HumanMessage("q4")]}, config
+    )["messages"]
+
+    assert after[:-2] == before
+    assert [m.content for m in after[-2:]] == ["q4", f"reply {len(before) + 1}"]
+    head = sync_checkpointer.get_tuple(config)
+    assert head is not None
+    assert head.checkpoint["channel_values"]["messages"] == after
+    assert delta.get_state(config).values["messages"] == after
+
+
+async def test_plain_reducer_reads_and_continues_delta_written_thread_async(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    delta = _app(DeltaState, async_checkpointer)
+    for i in range(4):
+        await delta.ainvoke({"messages": [HumanMessage(f"q{i}")]}, config)
+    plain = _app(PlainState, async_checkpointer)
+
+    expected = [s.values async for s in delta.aget_state_history(config)]
+    assert [s.values async for s in plain.aget_state_history(config)] == expected
+
+    after = (await plain.ainvoke({"messages": [HumanMessage("q4")]}, config))[
+        "messages"
+    ]
+    assert after[:-2] == expected[0]["messages"]
+
+
+@pytest.mark.parametrize(
+    ("seed", "writes", "expected"),
+    [
+        (_DeltaSnapshot(["a"]), [["b"], ["c"]], ["a", "b", "c"]),
+        (["a"], [["b"], Overwrite(["c"]), ["d"]], ["c", "d"]),
+        (None, [["b"], Overwrite(["c"]), Overwrite(["d"]), ["e"]], ["d", "e"]),
+    ],
+)
+def test_plain_reducer_replays_delta_history_one_write_at_a_time(
+    seed: Any, writes: list[Any], expected: list[str]
+) -> None:
+    history: dict[str, Any] = {"writes": [("t", "items", w) for w in writes]}
+    if seed is not None:
+        history["seed"] = seed
+    checkpoint = empty_checkpoint()
+    checkpoint["channel_versions"]["items"] = 1
+    saver = InMemorySaver()
+
+    with patch.object(
+        saver, "get_delta_channel_history", return_value={"items": history}
+    ):
+        channels, _ = channels_from_checkpoint(
+            {"items": BinaryOperatorAggregate(list, operator.add)},
+            checkpoint,
+            saver=saver,
+            config={"configurable": {"thread_id": "1"}},
+        )
+
+    assert channels["items"].get() == expected
+
+
+def test_plain_thread_reads_never_walk_history() -> None:
+    saver = InMemorySaver()
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    plain = _app(PlainState, saver)
+
+    with patch.object(saver, "get_delta_channel_history") as walk:
+        plain.invoke({"messages": [HumanMessage("q0")]}, config)
+        plain.update_state(config, {"messages": [HumanMessage("note")]})
+        plain.invoke({"messages": [HumanMessage("q1")]}, config)
+        plain.get_state(config)
+        list(plain.get_state_history(config))
+
+    walk.assert_not_called()

libs/langgraph/tests/test_delta_channel_update_overwrite.py

@@ -0,0 +1,110 @@
+"""An `Overwrite` through `update_state` snapshots its DeltaChannel on the
+checkpoint the update saves, as a node's `Overwrite` does on the loop's."""
+
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import HumanMessage
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from typing_extensions import TypedDict
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.last_value import LastValue
+from langgraph.graph import START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.types import Overwrite
+
+pytestmark = pytest.mark.anyio
+
+
+class _State(TypedDict):
+    messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+
+
+def _messages_graph(saver: InMemorySaver) -> Any:
+    builder = StateGraph(_State)
+    builder.add_node("model", lambda state: {})
+    builder.add_edge(START, "model")
+    return builder.compile(checkpointer=saver)
+
+
+def _extend(current: list, writes: list) -> list:
+    return [*current, *(item for write in writes for item in write)]
+
+
+def _delta_input_graph(saver: InMemorySaver) -> Pregel:
+    node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log")
+    return Pregel(
+        nodes={"n": node},
+        channels={"log": DeltaChannel(_extend), "go": LastValue(int)},
+        input_channels=["log", "go"],
+        output_channels=["log"],
+        checkpointer=saver,
+    )
+
+
+def test_update_state_with_an_overwrite_snapshots_the_channel() -> None:
+    saver = InMemorySaver()
+    graph = _messages_graph(saver)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage(content="a", id="1")]}, config)
+
+    graph.update_state(
+        config,
+        {"messages": Overwrite([HumanMessage(content="b", id="2")])},
+        as_node="model",
+    )
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+    assert [m.content for m in graph.get_state(config).values["messages"]] == ["b"]
+
+
+async def test_aupdate_state_with_an_overwrite_snapshots_the_channel() -> None:
+    saver = InMemorySaver()
+    graph = _messages_graph(saver)
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"messages": [HumanMessage(content="a", id="1")]}, config)
+
+    await graph.aupdate_state(
+        config,
+        {"messages": Overwrite([HumanMessage(content="b", id="2")])},
+        as_node="model",
+    )
+
+    head = await saver.aget_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+    values = (await graph.aget_state(config)).values
+    assert [m.content for m in values["messages"]] == ["b"]
+
+
+def test_update_state_as_input_with_an_overwrite_snapshots_the_channel() -> None:
+    saver = InMemorySaver()
+    graph = _delta_input_graph(saver)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"log": [0], "go": 1}, config)
+
+    graph.update_state(config, {"log": Overwrite([1])}, as_node="__input__")
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("log"), _DeltaSnapshot)
+    assert graph.get_state(config).values["log"] == [1]
+
+
+async def test_aupdate_state_as_input_with_an_overwrite_snapshots_the_channel() -> None:
+    saver = InMemorySaver()
+    graph = _delta_input_graph(saver)
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"log": [0], "go": 1}, config)
+
+    await graph.aupdate_state(config, {"log": Overwrite([1])}, as_node="__input__")
+
+    head = await saver.aget_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("log"), _DeltaSnapshot)
+    assert (await graph.aget_state(config)).values["log"] == [1]

libs/langgraph/tests/test_delta_channel_update_state.py

@@ -25,10 +25,13 @@ from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import TypedDict
 
+from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.last_value import LastValue
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
-from langgraph.types import StateSnapshot, StateUpdate
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.types import Overwrite, StateSnapshot, StateUpdate
 
 pytestmark = pytest.mark.anyio
 
@@ -538,3 +541,207 @@ def test_update_state_that_snapshots_keeps_a_deferred_node_pending() -> None:
 
     assert [m.content for m in final["messages"]] == ["s", "a", "u", "b"]
     assert graph.get_state(config).next == ()
+
+
+def _sorted_extend(current: list, writes: list) -> list:
+    return sorted([*current, *(item for write in writes for item in write)])
+
+
+def _delta_input_graph(snapshot_frequency: int = 1000) -> Any:
+    node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log", "plain")
+    return Pregel(
+        nodes={"n": node},
+        channels={
+            "log": DeltaChannel(_sorted_extend, snapshot_frequency=snapshot_frequency),
+            "plain": BinaryOperatorAggregate(list, lambda a, b: sorted(a + b)),
+            "go": LastValue(int),
+        },
+        input_channels=["log", "plain", "go"],
+        output_channels=["log", "plain"],
+        checkpointer=InMemorySaver(),
+    )
+
+
+@pytest.mark.parametrize("snapshot_frequency", [1, 2])
+def test_update_as_input_reads_back_on_its_checkpoint_and_after_the_next_run(
+    snapshot_frequency: int,
+) -> None:
+    graph = _delta_input_graph(snapshot_frequency)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"log": [0], "plain": [0], "go": 1}, config)
+
+    graph.update_state(config, {"log": [1], "plain": [1], "go": 1}, as_node="__input__")
+    after_update = graph.get_state(config).values
+    graph.invoke(None, config)
+    after_run = graph.get_state(config).values
+
+    assert after_update["log"] == after_update["plain"]
+    assert after_run["log"] == after_run["plain"]
+
+
+@pytest.mark.parametrize("snapshot_frequency", [1, 2])
+async def test_aupdate_as_input_reads_back_on_its_checkpoint_and_after_the_next_run(
+    snapshot_frequency: int,
+) -> None:
+    graph = _delta_input_graph(snapshot_frequency)
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"log": [0], "plain": [0], "go": 1}, config)
+
+    await graph.aupdate_state(
+        config, {"log": [1], "plain": [1], "go": 1}, as_node="__input__"
+    )
+    after_update = (await graph.aget_state(config)).values
+    await graph.ainvoke(None, config)
+    after_run = (await graph.aget_state(config)).values
+
+    assert after_update["log"] == after_update["plain"]
+    assert after_run["log"] == after_run["plain"]
+
+
+def test_update_as_input_to_an_older_checkpoint_stays_out_of_its_other_branch() -> None:
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"go": 1}, config)
+    older = graph.get_state(config).config
+    graph.invoke({"go": 1}, config)
+    other_branch = graph.get_state(config)
+
+    edited = graph.update_state(
+        older, {"log": [1], "plain": [1], "go": 1}, as_node="__input__"
+    )
+
+    values = graph.get_state(edited).values
+    assert values["log"] == values["plain"]
+    assert graph.get_state(other_branch.config).values == other_branch.values
+
+
+async def test_aupdate_as_input_to_an_older_checkpoint_stays_out_of_its_other_branch() -> (
+    None
+):
+    graph = _delta_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"go": 1}, config)
+    older = (await graph.aget_state(config)).config
+    await graph.ainvoke({"go": 1}, config)
+    other_branch = await graph.aget_state(config)
+
+    edited = await graph.aupdate_state(
+        older, {"log": [1], "plain": [1], "go": 1}, as_node="__input__"
+    )
+
+    values = (await graph.aget_state(edited)).values
+    assert values["log"] == values["plain"]
+    assert (await graph.aget_state(other_branch.config)).values == other_branch.values
+
+
+def _message_ids(graph: Any, config: dict) -> list[str | None]:
+    return [m.id for m in graph.get_state(config).values["messages"]]
+
+
+async def _amessage_ids(graph: Any, config: dict) -> list[str | None]:
+    return [m.id for m in (await graph.aget_state(config)).values["messages"]]
+
+
+def _messages_input_graph() -> Any:
+    node = (
+        NodeBuilder()
+        .subscribe_only("go")
+        .do(lambda _: [HumanMessage("n", id="n")])
+        .write_to("messages")
+    )
+    return Pregel(
+        nodes={"n": node},
+        channels={
+            "messages": DeltaChannel(_messages_delta_reducer),
+            "go": LastValue(int),
+        },
+        input_channels=["messages", "go"],
+        output_channels=["messages"],
+        checkpointer=InMemorySaver(),
+    )
+
+
+# Writes of "b" in shapes `_messages_delta_reducer` accepts.
+MESSAGE_SHAPES = [
+    pytest.param(lambda: [HumanMessage("b")], id="message"),
+    pytest.param(lambda: ["b"], id="string"),
+    pytest.param(lambda: "b", id="bare string"),
+    pytest.param(lambda: [("user", "b")], id="tuple"),
+    pytest.param(lambda: {"role": "user", "content": "b"}, id="bare dict"),
+    pytest.param(lambda: Overwrite([HumanMessage("b")]), id="Overwrite"),
+]
+
+
+@pytest.mark.parametrize("write", MESSAGE_SHAPES)
+def test_update_state_gives_a_message_of_any_shape_an_id_that_every_read_keeps(
+    write: Any,
+) -> None:
+    graph = _build_graph(InMemorySaver())
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("a", id="a")]}, config)
+
+    graph.update_state(config, {"messages": write()})
+
+    first, second = _message_ids(graph, config), _message_ids(graph, config)
+    assert first[-1] is not None
+    assert first == second
+
+
+@pytest.mark.parametrize("write", MESSAGE_SHAPES)
+async def test_aupdate_state_gives_a_message_of_any_shape_an_id_that_every_read_keeps(
+    write: Any,
+) -> None:
+    graph = _build_graph(InMemorySaver())
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"messages": [HumanMessage("a", id="a")]}, config)
+
+    await graph.aupdate_state(config, {"messages": write()})
+
+    first, second = (
+        await _amessage_ids(graph, config),
+        await _amessage_ids(graph, config),
+    )
+    assert first[-1] is not None
+    assert first == second
+
+
+@pytest.mark.parametrize("write", MESSAGE_SHAPES)
+def test_update_as_input_gives_a_message_of_any_shape_an_id(write: Any) -> None:
+    graph = _messages_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("a", id="a")], "go": 1}, config)
+
+    graph.update_state(config, {"messages": write()}, as_node="__input__")
+
+    first, second = _message_ids(graph, config), _message_ids(graph, config)
+    assert first[-1] is not None
+    assert first == second
+
+
+def test_update_state_on_an_older_checkpoint_gives_a_message_an_id() -> None:
+    graph = _build_graph(InMemorySaver())
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("a", id="a")]}, config)
+    older = graph.get_state(config).config
+    graph.invoke({"messages": [HumanMessage("c", id="c")]}, config)
+
+    branch = graph.update_state(older, {"messages": [HumanMessage("b")]})
+
+    assert _message_ids(graph, branch)[-1] is not None
+
+
+async def test_aupdate_as_input_gives_a_message_an_id_that_every_read_keeps() -> None:
+    graph = _messages_input_graph()
+    config = {"configurable": {"thread_id": "t"}}
+    await graph.ainvoke({"messages": [HumanMessage("a", id="a")], "go": 1}, config)
+
+    await graph.aupdate_state(
+        config, {"messages": [HumanMessage("b")]}, as_node="__input__"
+    )
+
+    first, second = (
+        await _amessage_ids(graph, config),
+        await _amessage_ids(graph, config),
+    )
+    assert first[-1] is not None
+    assert first == second

libs/langgraph/tests/test_delta_channel_write_failures.py

@@ -0,0 +1,168 @@
+"""A checkpoint must never be saved without the `DeltaChannel` writes it reads."""
+
+import operator
+import threading
+from typing import Annotated, Any
+
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+from typing_extensions import TypedDict
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import START, StateGraph
+from langgraph.types import Durability
+
+pytestmark = pytest.mark.anyio
+
+INPUT = {"log": [], "plain": []}
+FINAL = {"log": ["a", "b", "c"], "plain": ["a", "b", "c"]}
+
+# Exit mode saves nothing before the failed write, so its retry starts over.
+RETRIES = [
+    pytest.param("sync", None, id="sync"),
+    pytest.param("async", None, id="async"),
+    pytest.param("exit", INPUT, id="exit"),
+]
+
+
+def _append(current: list, writes: list) -> list:
+    return [*current, *(item for write in writes for item in write)]
+
+
+class _State(TypedDict):
+    log: Annotated[list, DeltaChannel(_append)]
+    plain: Annotated[list, operator.add]
+
+
+class _FailsTheWriteOfBOnce(InMemorySaver):
+    failed = False
+
+    def _fail_once(self, writes: Any) -> None:
+        if not self.failed and ("log", ["b"]) in writes:
+            self.failed = True
+            raise ConnectionError("b's write was not saved")
+
+    def put_writes(
+        self, config: Any, writes: Any, task_id: str, task_path: str = ""
+    ) -> None:
+        self._fail_once(writes)
+        super().put_writes(config, writes, task_id, task_path)
+
+    async def aput_writes(
+        self, config: Any, writes: Any, task_id: str, task_path: str = ""
+    ) -> None:
+        self._fail_once(writes)
+        await super().aput_writes(config, writes, task_id, task_path)
+
+
+class _FailsTheSaveAfterAOnce(InMemorySaver):
+    failed = False
+
+    def _fail_once(self, checkpoint: Any) -> None:
+        if not self.failed and checkpoint["channel_values"].get("plain") == ["a"]:
+            self.failed = True
+            raise ConnectionError("the checkpoint after a was not saved")
+
+    def put(
+        self, config: Any, checkpoint: Any, metadata: Any, new_versions: Any
+    ) -> Any:
+        self._fail_once(checkpoint)
+        return super().put(config, checkpoint, metadata, new_versions)
+
+    async def aput(
+        self, config: Any, checkpoint: Any, metadata: Any, new_versions: Any
+    ) -> Any:
+        self._fail_once(checkpoint)
+        return await super().aput(config, checkpoint, metadata, new_versions)
+
+
+def _a_then_b_then_c(saver: InMemorySaver) -> Any:
+    builder = StateGraph(_State)
+    for name in "abc":
+        builder.add_node(
+            name, lambda state, name=name: {"log": [name], "plain": [name]}
+        )
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("b", "c")
+    return builder.compile(checkpointer=saver)
+
+
+@pytest.mark.parametrize(("durability", "retry_input"), RETRIES)
+def test_a_failed_delta_write_is_rerun_not_lost(
+    durability: Durability, retry_input: dict | None
+) -> None:
+    graph = _a_then_b_then_c(_FailsTheWriteOfBOnce())
+    config = {"configurable": {"thread_id": "t"}}
+
+    with pytest.raises(ConnectionError):
+        graph.invoke(INPUT, config, durability=durability)
+    for state in graph.get_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+    graph.invoke(retry_input, config, durability=durability)
+    assert graph.get_state(config).values == FINAL
+
+
+@pytest.mark.parametrize(("durability", "retry_input"), RETRIES)
+async def test_a_failed_delta_write_is_rerun_not_lost_async(
+    durability: Durability, retry_input: dict | None
+) -> None:
+    graph = _a_then_b_then_c(_FailsTheWriteOfBOnce())
+    config = {"configurable": {"thread_id": "t"}}
+
+    with pytest.raises(ConnectionError):
+        await graph.ainvoke(INPUT, config, durability=durability)
+    async for state in graph.aget_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+    await graph.ainvoke(retry_input, config, durability=durability)
+    assert (await graph.aget_state(config)).values == FINAL
+
+
+@pytest.mark.parametrize("durability", ["sync", "async"])
+def test_a_failed_checkpoint_save_is_rerun_not_built_on(
+    durability: Durability,
+) -> None:
+    graph = _a_then_b_then_c(_FailsTheSaveAfterAOnce())
+    config = {"configurable": {"thread_id": "t"}}
+
+    with pytest.raises(ConnectionError):
+        graph.invoke(INPUT, config, durability=durability)
+    for state in graph.get_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+    graph.invoke(None, config, durability=durability)
+    assert graph.get_state(config).values == FINAL
+
+
+@pytest.mark.parametrize("durability", ["sync", "async"])
+async def test_a_failed_checkpoint_save_is_rerun_not_built_on_async(
+    durability: Durability,
+) -> None:
+    graph = _a_then_b_then_c(_FailsTheSaveAfterAOnce())
+    config = {"configurable": {"thread_id": "t"}}
+
+    with pytest.raises(ConnectionError):
+        await graph.ainvoke(INPUT, config, durability=durability)
+    async for state in graph.aget_state_history(config):
+        assert state.values.get("log", []) == state.values.get("plain", [])
+
+    await graph.ainvoke(None, config, durability=durability)
+    assert (await graph.aget_state(config)).values == FINAL
+
+
+def test_a_delta_graph_finishes_on_a_single_background_thread() -> None:
+    graph = _a_then_b_then_c(InMemorySaver())
+    config = {"configurable": {"thread_id": "t"}, "max_concurrency": 1}
+    result: dict = {}
+    run = threading.Thread(
+        target=lambda: result.update(graph.invoke(INPUT, config, durability="async")),
+        daemon=True,
+    )
+
+    run.start()
+    run.join(timeout=10)
+
+    assert not run.is_alive(), "invoke hung"
+    assert result == FINAL

libs/langgraph/tests/test_interrupt_state.py

@@ -152,6 +152,83 @@ def test_finished_task_does_not_report_answered_interrupt(
         assert asked == [["A1", "B"]]
 
 
+def test_finished_task_output_applies_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    builder = StateGraph(State)
+    builder.add_node("a", lambda s: {"log": [f"a:{interrupt('A')}"]})
+    builder.add_node("b", lambda s: {"log": [f"b:{interrupt('B')}"]})
+    builder.add_node("c", lambda s: {"log": [f"c:{interrupt('C')}"]})
+    builder.add_edge(START, "a")
+    builder.add_edge(START, "b")
+    builder.add_edge(START, "c")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    while pending := result.get("__interrupt__"):
+        result = graph.invoke(
+            Command(resume={pending[0].id: "yes"}), config, durability=durability
+        )
+
+    assert sorted(result["log"]) == ["a:yes", "b:yes", "c:yes"]
+
+
+def test_resume_with_update_applies_the_update_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    def ask_three_times(state: State) -> State:
+        answers = [interrupt(str(question)) for question in range(1, 4)]
+        return {"log": ["a:" + "".join(answers)]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("a", ask_three_times)
+        .add_edge(START, "a")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    for i in range(1, 4):
+        result = graph.invoke(
+            Command(resume=str(i), update={"log": [f"u{i}"]}),
+            config,
+            durability=durability,
+        )
+
+    assert sorted(result["log"]) == ["a:123", "u1", "u2", "u3"]
+    assert sorted(graph.get_state(config).values["log"]) == sorted(result["log"])
+
+
+@pytest.mark.parametrize("checkpoint_id", ["none", "head"])
+def test_resume_at_the_latest_checkpoint_does_not_rerun_finished_task(
+    sync_checkpointer: BaseCheckpointSaver, checkpoint_id: str
+) -> None:
+    calls: Counter[str] = Counter()
+    graph = _build_parallel(sync_checkpointer, calls)
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    def resume(question: str, answer: str) -> Any:
+        snapshot = graph.get_state(config)
+        target = (
+            snapshot.config
+            if checkpoint_id == "head"
+            else {"configurable": {**config["configurable"], "checkpoint_id": None}}
+        )
+        return graph.invoke(
+            Command(resume={_interrupt_by_value(snapshot, question).id: answer}),
+            target,
+        )
+
+    resume("A1", "yes")
+    result = resume("B", "ok")
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
 @NEEDS_CONTEXTVARS
 async def test_finished_task_does_not_report_answered_interrupt_async(
     async_checkpointer: BaseCheckpointSaver,
@@ -474,6 +551,64 @@ def test_functional_task_finished_with_none_is_not_rerun(
     assert calls == {"a": 2, "b": 3}
 
 
+def test_subgraph_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    calls: Counter[str] = Counter()
+    builder = StateGraph(State)
+    builder.add_node("child", _build_parallel(None, calls))
+    builder.add_edge(START, "child")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    graph.invoke({"log": []}, config, durability=durability)
+    snapshot = graph.get_state(config)
+    graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}),
+        config,
+        durability=durability,
+    )
+    snapshot = graph.get_state(config)
+    result = graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}),
+        config,
+        durability=durability,
+    )
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
+def test_nested_entrypoint_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    calls: Counter[str] = Counter()
+
+    @task
+    def step_1(query: str) -> str:
+        calls["step_1"] += 1
+        return f"{query} bar"
+
+    @task
+    def human_feedback(query: str) -> str:
+        calls["human_feedback"] += 1
+        return f"{query} {interrupt('feedback?')}"
+
+    @entrypoint()
+    def child(query: str) -> str:
+        return human_feedback(step_1(query).result()).result()
+
+    @entrypoint(checkpointer=sync_checkpointer)
+    def parent(query: str) -> str:
+        return child.invoke(query)
+
+    config = _config()
+    parent.invoke("foo", config)
+
+    assert parent.invoke(Command(resume="baz"), config) == "foo bar baz"
+    assert calls == {"step_1": 1, "human_feedback": 2}
+
+
 # --- Reading task status from recorded writes ---
 
 

libs/langgraph/tests/test_pregel.py

@@ -5932,9 +5932,9 @@ def test_no_redundant_put_writes_for_cached_task(
     put_writes_task_ids: list[str] = []
     orig = PregelLoop.put_writes
 
-    def spy(self, task_id, writes):
+    def spy(self, task_id, writes, **kwargs):
         put_writes_task_ids.append(task_id)
-        return orig(self, task_id, writes)
+        return orig(self, task_id, writes, **kwargs)
 
     with patch.object(PregelLoop, "put_writes", spy):
         result = workflow.invoke(Command(resume="ans"), config=config)
@@ -7737,6 +7737,8 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None:
 def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> None:
     # --- CHILD GRAPH ---
 
+    answered: Counter[str] = Counter()
+
     class ChildState(BaseModel):
         prompt: str = Field(..., description="What is going to be asked to the user?")
         human_input: str | None = Field(None, description="What the human said")
@@ -7746,6 +7748,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     def get_human_input(state: ChildState):
         human_input = interrupt(state.prompt)
+        answered["human"] += 1
 
         return dict(
             human_inputs=[human_input],  # update parent state
@@ -7753,6 +7756,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     def get_dolphin_input(state: ChildState):
         human_input = interrupt(state.prompt)
+        answered["dolphin"] += 1
 
         return dict(
             human_inputs=[human_input],  # update parent state
@@ -7846,6 +7850,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     assert invokes == 5
     assert len(events) == 5
+    assert answered == {"human": 2, "dolphin": 2}
 
 
 def test_pregel_loop_refcount():

libs/langgraph/tests/test_pregel_async.py

@@ -8157,9 +8157,9 @@ async def test_no_redundant_put_writes_for_cached_task(
     put_writes_task_ids: list[str] = []
     orig = PregelLoop.put_writes
 
-    def spy(self, task_id, writes):
+    def spy(self, task_id, writes, **kwargs):
         put_writes_task_ids.append(task_id)
-        return orig(self, task_id, writes)
+        return orig(self, task_id, writes, **kwargs)
 
     with patch.object(PregelLoop, "put_writes", spy):
         result = await workflow.ainvoke(Command(resume="ans"), config=config)

libs/langgraph/tests/test_prune_postgres.py

@@ -0,0 +1,467 @@
+import operator
+import sys
+from collections.abc import Callable
+from contextlib import contextmanager
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.postgres import PostgresSaver
+from langgraph.checkpoint.postgres.base import PRUNE_DELTA_THREADS_SQL, PRUNE_HEADS_SQL
+from typing_extensions import TypedDict
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer, add_messages
+from langgraph.types import Command, StateSnapshot, interrupt
+from tests.conftest import NO_DOCKER
+from tests.conftest_checkpointer import (
+    DEFAULT_POSTGRES_URI,
+    _checkpointer_postgres,
+    _checkpointer_postgres_aio,
+    _checkpointer_postgres_aio_pipe,
+    _checkpointer_postgres_aio_pool,
+    _checkpointer_postgres_pipe,
+    _checkpointer_postgres_pool,
+)
+
+pytestmark = [
+    pytest.mark.anyio,
+    pytest.mark.skipif(NO_DOCKER, reason="needs Postgres"),
+]
+
+NEEDS_CONTEXTVARS = pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+
+SYNC = [
+    _checkpointer_postgres,
+    _checkpointer_postgres_pipe,
+    _checkpointer_postgres_pool,
+]
+ASYNC = [
+    _checkpointer_postgres_aio,
+    _checkpointer_postgres_aio_pipe,
+    _checkpointer_postgres_aio_pool,
+]
+
+
+class State(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+    notes: Annotated[list[str], operator.add]
+
+
+class DeltaState(TypedDict):
+    messages: Annotated[list[AnyMessage], DeltaChannel(_messages_delta_reducer)]
+
+
+def _reply(ask: bool):
+    def reply(state: dict) -> dict:
+        n = len(state["messages"])
+        answer = interrupt("approve?") if ask else "ok"
+        return {"messages": [AIMessage(f"{answer} {n}", id=f"reply-{n}")]}
+
+    return reply
+
+
+def _graph(
+    checkpointer: Any,
+    *,
+    ask: bool = False,
+    fail: set[str] | None = None,
+    calls: dict[str, int] | None = None,
+) -> Any:
+    def note(name: str):
+        def node(state: State) -> dict:
+            if calls is not None:
+                calls[name] = calls.get(name, 0) + 1
+            if fail is not None and name in fail:
+                fail.discard(name)
+                raise ValueError(f"{name} failed")
+            return {"notes": [name]}
+
+        return node
+
+    builder = StateGraph(State)
+    builder.add_node("left", note("left"))
+    builder.add_node("right", note("right"))
+    builder.add_node("reply", _reply(ask))
+    builder.add_edge(START, "left")
+    builder.add_edge(START, "right")
+    builder.add_edge(["left", "right"], "reply")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _counting_subgraph(name: str, *, checkpointer: bool | None) -> Any:
+    class Calls(TypedDict):
+        messages: Annotated[list[AnyMessage], add_messages]
+        calls: Annotated[list[str], operator.add]
+
+    def count(state: Calls) -> dict:
+        n = len(state["messages"])
+        reply = AIMessage(f"{name} call {len(state['calls']) + 1}", id=f"{name}-{n}")
+        return {"calls": [name], "messages": [reply]}
+
+    builder = StateGraph(Calls)
+    builder.add_node("count", count)
+    builder.add_edge(START, "count")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _delta_graph(checkpointer: Any) -> Any:
+    builder = StateGraph(DeltaState)
+    builder.add_node("reply", _reply(ask=False))
+    builder.add_edge(START, "reply")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _config(thread_id: str) -> RunnableConfig:
+    return {"configurable": {"thread_id": thread_id}}
+
+
+def _say(i: int) -> dict:
+    return {"messages": [HumanMessage(f"hi {i}", id=f"hi-{i}")]}
+
+
+def _checkpoint_ids(checkpointer: Any, thread_id: str) -> list[str]:
+    return [c.checkpoint["id"] for c in checkpointer.list(_config(thread_id))]
+
+
+def _namespaces(checkpointer: Any, thread_id: str) -> list[str]:
+    return sorted(
+        c.config["configurable"]["checkpoint_ns"]
+        for c in checkpointer.list(_config(thread_id))
+    )
+
+
+async def _acheckpoint_ids(checkpointer: Any, thread_id: str) -> list[str]:
+    return [c.checkpoint["id"] async for c in checkpointer.alist(_config(thread_id))]
+
+
+def _parentless(snapshot: StateSnapshot) -> StateSnapshot:
+    return snapshot._replace(
+        parent_config=None,
+        tasks=tuple(
+            t._replace(state=_parentless(t.state))
+            if isinstance(t.state, StateSnapshot)
+            else t
+            for t in snapshot.tasks
+        ),
+    )
+
+
+def _run_once_after(
+    saver: PostgresSaver, sql: str, action: Callable[[], Any], *, nth: int = 1
+) -> None:
+    seen = 0
+    cursor = saver._cursor
+
+    class Hooked:
+        def __init__(self, cur: Any) -> None:
+            self.cur = cur
+
+        def __getattr__(self, name: str) -> Any:
+            return getattr(self.cur, name)
+
+        def execute(self, query: Any, params: Any = None, **kwargs: Any) -> None:
+            nonlocal seen
+            self.cur.execute(query, params, **kwargs)
+            if query is sql:
+                seen += 1
+                if seen == nth:
+                    action()
+
+    @contextmanager
+    def hooked(**kwargs: Any):
+        with cursor(**kwargs) as cur:
+            yield Hooked(cur)
+
+    saver._cursor = hooked
+
+
+@pytest.mark.parametrize("durability", ["sync", "async", "exit"])
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_keeps_the_state_and_the_thread_runs_on_like_its_twin(
+    make_checkpointer: Any, durability: str
+) -> None:
+    with make_checkpointer() as checkpointer:
+        graph = _graph(checkpointer)
+        for i in range(3):
+            for thread_id in ("pruned", "twin"):
+                graph.invoke(_say(i), _config(thread_id), durability=durability)
+        before = graph.get_state(_config("pruned"))
+
+        checkpointer.prune(["pruned"])
+
+        assert _checkpoint_ids(checkpointer, "pruned") == [
+            before.config["configurable"]["checkpoint_id"]
+        ]
+        assert graph.get_state(_config("pruned")) == _parentless(before)
+        for thread_id in ("pruned", "twin"):
+            graph.invoke(_say(3), _config(thread_id), durability=durability)
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_keeps_a_pending_interrupt_resumable(
+    make_checkpointer: Any,
+) -> None:
+    with make_checkpointer() as checkpointer:
+        graph = _graph(checkpointer, ask=True)
+        for thread_id in ("pruned", "twin"):
+            for i in range(2):
+                graph.invoke(_say(i), _config(thread_id))
+                graph.invoke(Command(resume=f"yes {i}"), _config(thread_id))
+            graph.invoke(_say(2), _config(thread_id))
+        before = graph.get_state(_config("pruned"))
+        assert before.interrupts
+
+        checkpointer.prune(["pruned"])
+
+        assert graph.get_state(_config("pruned")) == _parentless(before)
+        for thread_id in ("pruned", "twin"):
+            graph.invoke(Command(resume="yes 2"), _config(thread_id))
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_keeps_writes_of_tasks_that_finished_before_a_failure(
+    make_checkpointer: Any,
+) -> None:
+    with make_checkpointer() as checkpointer:
+        calls: dict[str, int] = {}
+        fail: set[str] = set()
+        graph = _graph(checkpointer, fail=fail, calls=calls)
+        for thread_id in ("pruned", "twin"):
+            for i in range(2):
+                graph.invoke(_say(i), _config(thread_id))
+            fail.add("right")
+            with pytest.raises(ValueError, match="right failed"):
+                graph.invoke(_say(2), _config(thread_id))
+        before = graph.get_state(_config("pruned"))
+
+        checkpointer.prune(["pruned"])
+
+        assert graph.get_state(_config("pruned")) == _parentless(before)
+        left_runs = calls["left"]
+        graph.invoke(None, _config("pruned"))
+        assert calls["left"] == left_runs, "a task that had finished ran again"
+        graph.invoke(None, _config("twin"))
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_keeps_an_interrupted_subgraph_resumable(
+    make_checkpointer: Any,
+) -> None:
+    with make_checkpointer() as checkpointer:
+        sub = StateGraph(State)
+        sub.add_node("reply", _reply(ask=True))
+        sub.add_edge(START, "reply")
+        parent = StateGraph(State)
+        parent.add_node("sub", sub.compile())
+        parent.add_edge(START, "sub")
+        graph = parent.compile(checkpointer=checkpointer)
+        for thread_id in ("pruned", "twin"):
+            for i in range(2):
+                graph.invoke(_say(i), _config(thread_id))
+                graph.invoke(Command(resume=f"yes {i}"), _config(thread_id))
+            graph.invoke(_say(2), _config(thread_id))
+        before = graph.get_state(_config("pruned"), subgraphs=True)
+        interrupted = before.tasks[0].state.config["configurable"]["checkpoint_ns"]
+
+        checkpointer.prune(["pruned"])
+
+        assert _namespaces(checkpointer, "pruned") == ["", interrupted]
+        assert graph.get_state(_config("pruned"), subgraphs=True) == _parentless(before)
+        for thread_id in ("pruned", "twin"):
+            graph.invoke(Command(resume="yes 2"), _config(thread_id))
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+@pytest.mark.parametrize("durability", ["sync", "async", "exit"])
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_keeps_a_checkpointer_true_subgraph_but_not_finished_calls(
+    make_checkpointer: Any, durability: str
+) -> None:
+    with make_checkpointer() as checkpointer:
+        parent = StateGraph(State)
+        parent.add_node("memory", _counting_subgraph("memory", checkpointer=True))
+        parent.add_node("call", _counting_subgraph("call", checkpointer=None))
+        parent.add_edge(START, "memory")
+        parent.add_edge("memory", "call")
+        graph = parent.compile(checkpointer=checkpointer)
+        for i in range(3):
+            for thread_id in ("pruned", "twin"):
+                graph.invoke(_say(i), _config(thread_id), durability=durability)
+
+        checkpointer.prune(["pruned"])
+
+        assert _namespaces(checkpointer, "pruned") == ["", "memory"]
+        for thread_id in ("pruned", "twin"):
+            graph.invoke(_say(3), _config(thread_id), durability=durability)
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+def test_keep_latest_keeps_a_turn_saved_after_it_read_the_heads() -> None:
+    with (
+        _checkpointer_postgres() as checkpointer,
+        PostgresSaver.from_conn_string(
+            DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname
+        ) as pruner,
+    ):
+        graph = _graph(checkpointer)
+        for i in range(3):
+            for thread_id in ("pruned", "twin"):
+                graph.invoke(_say(i), _config(thread_id))
+        _run_once_after(
+            pruner, PRUNE_HEADS_SQL, lambda: graph.invoke(_say(3), _config("pruned"))
+        )
+
+        pruner.prune(["pruned"])
+
+        graph.invoke(_say(3), _config("twin"))
+        assert (
+            graph.get_state(_config("pruned")).values
+            == graph.get_state(_config("twin")).values
+        )
+
+
+def test_keep_latest_refuses_a_thread_that_switches_to_a_delta_channel_mid_batch() -> (
+    None
+):
+    with (
+        _checkpointer_postgres() as checkpointer,
+        PostgresSaver.from_conn_string(
+            DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname
+        ) as pruner,
+    ):
+        graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer)
+        for i in range(3):
+            for thread_id in ("switched", "twin"):
+                graph.invoke(_say(i), _config(thread_id))
+        _run_once_after(
+            pruner,
+            PRUNE_DELTA_THREADS_SQL,
+            lambda: delta_graph.invoke(_say(3), _config("switched")),
+        )
+
+        with pytest.raises(NotImplementedError, match="switched"):
+            pruner.prune(["switched"])
+
+        delta_graph.invoke(_say(3), _config("twin"))
+        assert (
+            delta_graph.get_state(_config("switched")).values
+            == delta_graph.get_state(_config("twin")).values
+        )
+
+
+def test_keep_latest_keeps_a_delta_channel_turn_saved_after_the_thread_check() -> None:
+    with (
+        _checkpointer_postgres() as checkpointer,
+        PostgresSaver.from_conn_string(
+            DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname
+        ) as pruner,
+    ):
+        graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer)
+        for i in range(3):
+            for thread_id in ("switched", "twin"):
+                graph.invoke(_say(i), _config(thread_id))
+        head = _checkpoint_ids(checkpointer, "switched")[0]
+        _run_once_after(
+            pruner,
+            PRUNE_DELTA_THREADS_SQL,
+            lambda: delta_graph.invoke(_say(3), _config("switched")),
+            nth=2,
+        )
+
+        pruner.prune(["switched"])
+
+        assert _checkpoint_ids(checkpointer, "switched")[-1] == head
+        delta_graph.invoke(_say(3), _config("twin"))
+        assert (
+            delta_graph.get_state(_config("switched")).values
+            == delta_graph.get_state(_config("twin")).values
+        )
+
+
+@pytest.mark.parametrize("make_checkpointer", SYNC)
+def test_keep_latest_refuses_a_batch_with_a_delta_channel_thread(
+    make_checkpointer: Any,
+) -> None:
+    with make_checkpointer() as checkpointer:
+        graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer)
+        for i in range(3):
+            graph.invoke(_say(i), _config("ordinary"))
+            delta_graph.invoke(_say(i), _config("delta"))
+        before = {t: _checkpoint_ids(checkpointer, t) for t in ("ordinary", "delta")}
+        delta_before = delta_graph.get_state(_config("delta"))
+
+        with pytest.raises(NotImplementedError, match="delta"):
+            checkpointer.prune(["ordinary", "delta"])
+
+        assert {t: _checkpoint_ids(checkpointer, t) for t in before} == before
+        assert delta_graph.get_state(_config("delta")) == delta_before
+
+
+@NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("make_checkpointer", ASYNC)
+async def test_akeep_latest_keeps_a_pending_interrupt_resumable(
+    make_checkpointer: Any,
+) -> None:
+    async with make_checkpointer() as checkpointer:
+        graph = _graph(checkpointer, ask=True)
+        for thread_id in ("pruned", "twin"):
+            for i in range(2):
+                await graph.ainvoke(_say(i), _config(thread_id))
+                await graph.ainvoke(Command(resume=f"yes {i}"), _config(thread_id))
+            await graph.ainvoke(_say(2), _config(thread_id))
+        before = await graph.aget_state(_config("pruned"))
+
+        await checkpointer.aprune(["pruned"])
+
+        assert await _acheckpoint_ids(checkpointer, "pruned") == [
+            before.config["configurable"]["checkpoint_id"]
+        ]
+        assert await graph.aget_state(_config("pruned")) == _parentless(before)
+        for thread_id in ("pruned", "twin"):
+            await graph.ainvoke(Command(resume="yes 2"), _config(thread_id))
+        assert (await graph.aget_state(_config("pruned"))).values == (
+            await graph.aget_state(_config("twin"))
+        ).values
+
+
+@pytest.mark.parametrize("make_checkpointer", ASYNC)
+async def test_akeep_latest_refuses_a_batch_with_a_delta_channel_thread(
+    make_checkpointer: Any,
+) -> None:
+    async with make_checkpointer() as checkpointer:
+        graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer)
+        for i in range(3):
+            await graph.ainvoke(_say(i), _config("ordinary"))
+            await delta_graph.ainvoke(_say(i), _config("delta"))
+        before = {
+            t: await _acheckpoint_ids(checkpointer, t) for t in ("ordinary", "delta")
+        }
+
+        with pytest.raises(NotImplementedError, match="delta"):
+            await checkpointer.aprune(["ordinary", "delta"])
+
+        assert {t: await _acheckpoint_ids(checkpointer, t) for t in before} == before

libs/langgraph/tests/test_remote_graph.py

@@ -16,13 +16,11 @@ from langgraph_sdk.schema import StreamPart
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
-from langgraph.channels.last_value import LastValue
 from langgraph.errors import GraphInterrupt
-from langgraph.func import task
 from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
-from langgraph.pregel import NodeBuilder, Pregel
-from langgraph.pregel.remote import RemoteException, RemoteGraph
-from langgraph.types import Command, Interrupt, RetryPolicy, StateSnapshot
+from langgraph.pregel import Pregel
+from langgraph.pregel.remote import RemoteGraph
+from langgraph.types import Interrupt, StateSnapshot
 from tests.any_str import AnyStr
 from tests.conftest import NO_DOCKER
 from tests.example_app.example_graph import app
@@ -886,441 +884,6 @@ def test_stream_sanitizes_thread_id():
     assert not passed_config["configurable"]
 
 
-class _ApprovalState(TypedDict, total=False):
-    request: str
-    approved: bool
-    result: str
-
-
-_REMOTE_INTERRUPT_ID = "0123456789abcdef0123456789abcdef"
-
-_REMOTE_INTERRUPT_RUN = [
-    StreamPart(
-        event="updates",
-        data={
-            "__interrupt__": [
-                {"value": {"question": "Approve X?"}, "id": _REMOTE_INTERRUPT_ID}
-            ]
-        },
-    )
-]
-
-_REMOTE_DONE_RUN = [
-    StreamPart(
-        event="values", data={"request": "X", "approved": True, "result": "done"}
-    )
-]
-
-
-_INTERRUPTED_THREAD = {
-    "status": "interrupted",
-    "interrupts": {
-        "task": [{"value": {"question": "Approve X?"}, "id": _REMOTE_INTERRUPT_ID}]
-    },
-}
-
-_IDLE_THREAD = {"status": "idle", "interrupts": {}}
-
-_RETRY = RetryPolicy(initial_interval=0, jitter=False, retry_on=ValueError)
-
-
-def _remote_interrupt_parent(remote_client: MagicMock, next_client: MagicMock):
-    return (
-        StateGraph(_ApprovalState)
-        .add_node(
-            "agent_b",
-            RemoteGraph("agent_b", sync_client=remote_client, client=remote_client),
-            retry_policy=_RETRY,
-        )
-        .add_node(
-            "agent_c",
-            RemoteGraph("agent_c", sync_client=next_client, client=next_client),
-            retry_policy=_RETRY,
-        )
-        .add_edge(START, "agent_b")
-        .add_edge("agent_b", "agent_c")
-        .compile(checkpointer=InMemorySaver())
-    )
-
-
-def _sync_clients(remote_runs: list, next_runs: list) -> tuple[MagicMock, MagicMock]:
-    remote_client = MagicMock()
-    remote_client.runs.stream.side_effect = remote_runs
-    remote_client.threads.get.return_value = _INTERRUPTED_THREAD
-    next_client = MagicMock()
-    next_client.runs.stream.side_effect = next_runs
-    next_client.threads.get.return_value = _IDLE_THREAD
-    return remote_client, next_client
-
-
-@pytest.mark.parametrize(
-    "resume",
-    [{"approved": True}, {_REMOTE_INTERRUPT_ID: {"approved": True}}],
-    ids=["value", "interrupt_id"],
-)
-def test_subgraph_resume_is_forwarded_to_remote_thread(resume):
-    remote_client, next_client = _sync_clients(
-        [_REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN], [_REMOTE_DONE_RUN]
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-
-    out = parent.invoke({"request": "X"}, config)
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-
-    out = parent.invoke(Command(resume=resume), config)
-    assert "__interrupt__" not in out
-    assert out["result"] == "done"
-
-    first, second = remote_client.runs.stream.call_args_list
-    assert first.kwargs["input"] == {"request": "X"}
-    assert first.kwargs["command"] is None
-    assert second.kwargs["input"] is None
-    assert second.kwargs["command"] == {"resume": resume}
-    assert second.kwargs["thread_id"] == first.kwargs["thread_id"]
-
-    next_kwargs = next_client.runs.stream.call_args.kwargs
-    assert next_kwargs["command"] is None
-    assert next_kwargs["input"] == {"request": "X", "approved": True, "result": "done"}
-
-
-def test_subgraph_resume_for_other_interrupt_reraises_remote_interrupt():
-    remote_client, next_client = _sync_clients([_REMOTE_INTERRUPT_RUN], [])
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    other_id = "fedcba9876543210fedcba9876543210"
-    out = parent.invoke(Command(resume={other_id: {"approved": True}}), config)
-
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-    assert remote_client.runs.stream.call_count == 1
-
-
-def test_subgraph_retry_after_resume_sends_input_to_idle_remote_thread():
-    remote_client, next_client = _sync_clients(
-        [_REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN],
-        [ValueError("transient"), _REMOTE_DONE_RUN],
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    out = parent.invoke(
-        Command(resume={_REMOTE_INTERRUPT_ID: {"approved": True}}), config
-    )
-
-    assert out["result"] == "done"
-    retry = next_client.runs.stream.call_args_list[1]
-    assert retry.kwargs["command"] is None
-    assert retry.kwargs["input"] == {"request": "X", "approved": True, "result": "done"}
-
-
-def test_subgraph_retry_after_consumed_resume_reraises_remote_interrupt():
-    remote_client, next_client = _sync_clients(
-        [_REMOTE_INTERRUPT_RUN, ValueError("transient"), _REMOTE_DONE_RUN],
-        [_REMOTE_DONE_RUN],
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    out = parent.invoke(Command(resume={"approved": True}), config)
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-    assert remote_client.runs.stream.call_count == 2
-
-    out = parent.invoke(Command(resume={"approved": True}), config)
-    assert out["result"] == "done"
-    for call in remote_client.runs.stream.call_args_list[1:]:
-        assert call.kwargs["input"] is None
-        assert call.kwargs["command"] == {"resume": {"approved": True}}
-
-
-def test_subgraph_resume_map_is_scoped_to_each_remote():
-    id_a, id_b = _REMOTE_INTERRUPT_ID, "fedcba9876543210fedcba9876543210"
-
-    def client(interrupt_id: str, done: dict) -> MagicMock:
-        pending = [{"value": "q", "id": interrupt_id}]
-        c = MagicMock()
-        c.runs.stream.side_effect = [
-            [StreamPart(event="updates", data={"__interrupt__": pending})],
-            [StreamPart(event="values", data=done)],
-        ]
-        c.threads.get.return_value = {
-            "status": "interrupted",
-            "interrupts": {"task": pending},
-        }
-        return c
-
-    client_a = client(id_a, {"approved": True})
-    client_b = client(id_b, {"result": "done"})
-    parent = (
-        StateGraph(_ApprovalState)
-        .add_node("a", RemoteGraph("a", sync_client=client_a, client=client_a))
-        .add_node("b", RemoteGraph("b", sync_client=client_b, client=client_b))
-        .add_edge(START, "a")
-        .add_edge(START, "b")
-        .compile(checkpointer=InMemorySaver())
-    )
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    out = parent.invoke(Command(resume={id_a: "answer a", id_b: "answer b"}), config)
-
-    assert out["approved"] is True
-    assert out["result"] == "done"
-    assert client_a.runs.stream.call_args.kwargs["command"] == {
-        "resume": {id_a: "answer a"}
-    }
-    assert client_b.runs.stream.call_args.kwargs["command"] == {
-        "resume": {id_b: "answer b"}
-    }
-
-
-def _resumed_then_dropped():
-    # A remote node's next interrupt() reuses the id of the one just resumed.
-    yield from _REMOTE_INTERRUPT_RUN
-    raise ValueError("transient")
-
-
-def test_subgraph_retry_does_not_resend_resume_map():
-    remote_client, next_client = _sync_clients(
-        [_REMOTE_INTERRUPT_RUN, _resumed_then_dropped()], []
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    out = parent.invoke(
-        Command(resume={_REMOTE_INTERRUPT_ID: {"approved": True}}), config
-    )
-
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-    assert remote_client.runs.stream.call_count == 2
-
-
-def _ancestor_retry_parent(remote_client: MagicMock, via: str = "node"):
-    # A low-level Pregel subgraph has no START step, so an ancestor retry's
-    # RESUMING flag reaches the RemoteGraph task with a fresh node_attempt.
-    remote = RemoteGraph("agent_b", sync_client=remote_client, client=remote_client)
-    inner = Pregel(
-        nodes={
-            "agent_b": NodeBuilder()
-            .subscribe_only("request")
-            .do(remote)
-            .write_to("result")
-        },
-        channels={"request": LastValue(str), "result": LastValue(object)},
-        input_channels=["request"],
-        output_channels=["result"],
-    )
-    if via == "node":
-        p = inner
-    elif via == "task":
-
-        @task
-        def child(request: str):
-            return inner.invoke({"request": request})["result"]
-
-        def p(state: _ApprovalState):
-            return {"result": child(state["request"]).result()}
-
-    else:
-
-        @task
-        async def child(request: str):
-            return (await inner.ainvoke({"request": request}))["result"]
-
-        async def p(state: _ApprovalState):
-            return {"result": await child(state["request"])}
-
-    return (
-        StateGraph(_ApprovalState)
-        .add_node("p", p, retry_policy=_RETRY)
-        .add_edge(START, "p")
-        .compile(checkpointer=InMemorySaver())
-    )
-
-
-@pytest.mark.parametrize("via", ["node", "task"])
-def test_subgraph_ancestor_retry_does_not_resend_resume_map(via):
-    remote_client, _ = _sync_clients(
-        [_REMOTE_INTERRUPT_RUN, _resumed_then_dropped(), _REMOTE_DONE_RUN], []
-    )
-    parent = _ancestor_retry_parent(remote_client, via)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    parent.invoke({"request": "X"}, config)
-
-    out = parent.invoke(Command(resume={_REMOTE_INTERRUPT_ID: "first"}), config)
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-
-    parent.invoke(Command(resume={_REMOTE_INTERRUPT_ID: "second"}), config)
-    assert [c.kwargs["command"] for c in remote_client.runs.stream.call_args_list] == [
-        None,
-        {"resume": {_REMOTE_INTERRUPT_ID: "first"}},
-        {"resume": {_REMOTE_INTERRUPT_ID: "second"}},
-    ]
-
-
-def test_subgraph_empty_remote_interrupt_pauses_parent():
-    remote_client, next_client = _sync_clients(
-        [[StreamPart(event="updates", data={"__interrupt__": []})]], []
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-
-    parent.invoke({"request": "X"}, config)
-
-    assert parent.get_state(config).next == ("agent_b",)
-    next_client.runs.stream.assert_not_called()
-
-
-def test_subgraph_remote_error_after_interrupt_is_raised():
-    remote_client, next_client = _sync_clients(
-        [
-            [
-                *_REMOTE_INTERRUPT_RUN,
-                StreamPart(event="error", data={"error": "boom", "message": "x"}),
-            ]
-        ],
-        [],
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-
-    with pytest.raises(RemoteException):
-        parent.invoke({"request": "X"}, {"configurable": {"thread_id": "t"}})
-
-
-def test_subgraph_surfaces_all_remote_interrupts():
-    second_id = "fedcba9876543210fedcba9876543210"
-    remote_client, next_client = _sync_clients(
-        [
-            [
-                *_REMOTE_INTERRUPT_RUN,
-                StreamPart(event="values", data={"request": "X"}),
-                StreamPart(
-                    event="updates",
-                    data={"__interrupt__": [{"value": "second", "id": second_id}]},
-                ),
-            ]
-        ],
-        [],
-    )
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-
-    out = parent.invoke({"request": "X"}, config)
-
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID),
-        Interrupt(value="second", id=second_id),
-    ]
-
-
-async def test_subgraph_resume_is_forwarded_to_remote_thread_async():
-    def async_runs(*runs: list) -> list[MagicMock]:
-        iters = []
-        for parts in runs:
-            async_iter = MagicMock()
-            async_iter.__aiter__.return_value = parts
-            iters.append(async_iter)
-        return iters
-
-    remote_client = MagicMock()
-    remote_client.runs.stream.side_effect = async_runs(
-        _REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN
-    )
-    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
-    next_client = MagicMock()
-    next_client.runs.stream.side_effect = async_runs(_REMOTE_DONE_RUN)
-    next_client.threads.get = AsyncMock(return_value=_IDLE_THREAD)
-    parent = _remote_interrupt_parent(remote_client, next_client)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-
-    out = await parent.ainvoke({"request": "X"}, config)
-    assert "__interrupt__" in out
-
-    out = await parent.ainvoke(Command(resume={"approved": True}), config)
-    assert "__interrupt__" not in out
-    assert out["result"] == "done"
-
-    second = remote_client.runs.stream.call_args_list[1]
-    assert second.kwargs["input"] is None
-    assert second.kwargs["command"] == {"resume": {"approved": True}}
-    assert next_client.runs.stream.call_args.kwargs["command"] is None
-
-
-async def test_subgraph_retry_does_not_resend_resume_map_async():
-    async def run(*parts: StreamPart, error: Exception | None = None):
-        for part in parts:
-            yield part
-        if error:
-            raise error
-
-    remote_client = MagicMock()
-    remote_client.runs.stream.side_effect = [
-        run(*_REMOTE_INTERRUPT_RUN),
-        run(*_REMOTE_INTERRUPT_RUN, error=ValueError("transient")),
-    ]
-    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
-    parent = _remote_interrupt_parent(remote_client, MagicMock())
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    await parent.ainvoke({"request": "X"}, config)
-
-    resume = {_REMOTE_INTERRUPT_ID: {"approved": True}}
-    out = await parent.ainvoke(Command(resume=resume), config)
-
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-    assert remote_client.runs.stream.call_count == 2
-    assert remote_client.runs.stream.call_args.kwargs["command"] == {"resume": resume}
-
-
-@pytest.mark.parametrize(
-    "via", ["node", pytest.param("atask", marks=NEEDS_CONTEXTVARS)]
-)
-async def test_subgraph_ancestor_retry_does_not_resend_resume_map_async(via):
-    async def run(*parts: StreamPart, error: Exception | None = None):
-        for part in parts:
-            yield part
-        if error:
-            raise error
-
-    remote_client = MagicMock()
-    remote_client.runs.stream.side_effect = [
-        run(*_REMOTE_INTERRUPT_RUN),
-        run(*_REMOTE_INTERRUPT_RUN, error=ValueError("transient")),
-        run(*_REMOTE_DONE_RUN),
-    ]
-    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
-    parent = _ancestor_retry_parent(remote_client, via)
-    config = {"configurable": {"thread_id": "parent_thread"}}
-    await parent.ainvoke({"request": "X"}, config)
-
-    out = await parent.ainvoke(Command(resume={_REMOTE_INTERRUPT_ID: "first"}), config)
-    assert out["__interrupt__"] == [
-        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
-    ]
-
-    await parent.ainvoke(Command(resume={_REMOTE_INTERRUPT_ID: "second"}), config)
-    assert [c.kwargs["command"] for c in remote_client.runs.stream.call_args_list] == [
-        None,
-        {"resume": {_REMOTE_INTERRUPT_ID: "first"}},
-        {"resume": {_REMOTE_INTERRUPT_ID: "second"}},
-    ]
-
-
 @pytest.mark.anyio
 async def test_ainvoke():
     # set up test

libs/langgraph/tests/test_runtime.py

@@ -248,6 +248,95 @@ def test_drain_from_subgraph_can_resume_parent() -> None:
     }
 
 
+class _BreakpointState(TypedDict, total=False):
+    first: str
+    second: str
+
+
+def _drain_before_breakpoint_builder(
+    second_runs: list[str],
+) -> StateGraph[_BreakpointState]:
+    def first(state: _BreakpointState, runtime: Runtime) -> _BreakpointState:
+        runtime.control.request_drain("rollout")
+        return {"first": "done"}
+
+    def second(state: _BreakpointState) -> _BreakpointState:
+        second_runs.append("second")
+        return {"second": "done"}
+
+    builder = StateGraph(_BreakpointState)
+    builder.add_node("first", first)
+    builder.add_node("second", second)
+    builder.add_edge(START, "first")
+    builder.add_edge("first", "second")
+    return builder
+
+
+@pytest.mark.parametrize("at_compile", [True, False])
+def test_drain_before_interrupt_before_stops_at_breakpoint(at_compile: bool) -> None:
+    # A resume passes the next step's breakpoints, so a drain right before a
+    # breakpoint has to stop there instead, or the resumed run skips it.
+    second_runs: list[str] = []
+    breakpoints = {"interrupt_before": ["second"]}
+    compiled = _drain_before_breakpoint_builder(second_runs).compile(
+        checkpointer=MemorySaver(), **(breakpoints if at_compile else {})
+    )
+    run_kwargs = {} if at_compile else breakpoints
+    config = {"configurable": {"thread_id": "drain-breakpoint"}}
+
+    assert compiled.invoke({}, config, control=RunControl(), **run_kwargs) == {
+        "first": "done"
+    }
+    assert compiled.get_state(config).next == ("second",)
+    assert second_runs == []
+
+    assert compiled.invoke(None, config, **run_kwargs) == {
+        "first": "done",
+        "second": "done",
+    }
+    assert second_runs == ["second"]
+
+
+def test_drain_before_subgraph_interrupt_before_stops_at_breakpoint() -> None:
+    second_runs: list[str] = []
+    child = _drain_before_breakpoint_builder(second_runs).compile(
+        interrupt_before=["second"]
+    )
+    parent = StateGraph(_BreakpointState)
+    parent.add_node("child", child)
+    parent.add_edge(START, "child")
+    compiled = parent.compile(checkpointer=MemorySaver())
+    config = {"configurable": {"thread_id": "drain-subgraph-breakpoint"}}
+
+    compiled.invoke({}, config, control=RunControl())
+    state = compiled.get_state(config, subgraphs=True)
+    assert state.next == ("child",)
+    assert state.tasks[0].state.next == ("second",)
+    assert second_runs == []
+
+    assert compiled.invoke(None, config) == {"first": "done", "second": "done"}
+    assert second_runs == ["second"]
+
+
+@pytest.mark.anyio
+async def test_drain_before_interrupt_before_stops_at_breakpoint_async() -> None:
+    second_runs: list[str] = []
+    compiled = _drain_before_breakpoint_builder(second_runs).compile(
+        checkpointer=MemorySaver(), interrupt_before=["second"]
+    )
+    config = {"configurable": {"thread_id": "drain-breakpoint-async"}}
+
+    assert await compiled.ainvoke({}, config, control=RunControl()) == {"first": "done"}
+    assert (await compiled.aget_state(config)).next == ("second",)
+    assert second_runs == []
+
+    assert await compiled.ainvoke(None, config) == {
+        "first": "done",
+        "second": "done",
+    }
+    assert second_runs == ["second"]
+
+
 @pytest.mark.anyio
 async def test_drain_requested_in_terminal_step_finishes_normally_async() -> None:
     class State(TypedDict, total=False):

libs/langgraph/tests/test_time_travel.py

@@ -1177,6 +1177,43 @@ def test_subgraph_interrupt_resume_with_explicit_head_checkpoint_id(
     assert result["value"] == ["sub_a", "human:answer", "sub_b"]
 
 
+def test_subgraph_replay_reasks_answered_parallel_interrupt(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Replaying a subgraph checkpoint asks its step's questions again, including
+    one already answered. An update checkpoint replays in place instead of
+    forking, so its tasks keep their ids and its saved answers could be reused."""
+
+    subgraph = (
+        StateGraph(State)
+        .add_node("a", lambda s: {"value": [f"a:{interrupt('a')}"]})
+        .add_node("b", lambda s: {"value": [f"b:{interrupt('b')}"]})
+        .add_edge(START, "a")
+        .add_edge(START, "b")
+        .compile(checkpointer=True)
+    )
+    graph = (
+        StateGraph(State)
+        .add_node("subgraph_node", subgraph)
+        .add_edge(START, "subgraph_node")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+
+    def subgraph_head() -> RunnableConfig:
+        return graph.get_state(config, subgraphs=True).tasks[0].state.config
+
+    graph.invoke({"value": []}, config)
+    graph.update_state(subgraph_head(), {"value": ["u"]}, as_node="__start__")
+    result = graph.invoke(None, config)
+    question_a = next(i for i in result["__interrupt__"] if i.value == "a")
+    graph.invoke(Command(resume={question_a.id: "yes"}), config)
+
+    result = graph.invoke(None, subgraph_head())
+
+    assert sorted(i.value for i in result["__interrupt__"]) == ["a", "b"]
+
+
 def test_subgraph_replay_loads_accumulated_state_then_resume(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/uv.lock

@@ -1451,7 +1451,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.14"
+version = "1.2.15"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1685,7 +1685,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.2"
+version = "3.2.0"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -1734,7 +1734,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.1"
+version = "3.1.2"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
     { name = "aiosqlite" },
@@ -1835,7 +1835,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.14"
+version = "1.2.15"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -418,7 +418,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.2"
+version = "3.2.0"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -467,7 +467,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.1"
+version = "3.1.2"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
     { name = "aiosqlite" },
@@ -512,7 +512,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/__init__.py

@@ -3,7 +3,7 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 
-__version__ = "0.4.6"
+__version__ = "0.4.7"
 
 __all__ = [
     "Auth",

libs/sdk-py/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.14"
+version = "1.2.15"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -433,7 +433,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },