docs(langgraph): say each interrupt() call has its own id in UnknownInterruptError

Bumps langgraph-cli from 0.4.40 to 0.4.41.

Includes:

- Remove the outdated private-beta notice for `--agent-id` and
`--agent-environment` from `langgraph deploy` (#9506)

.github/workflows/_integration_test.yml

@@ -115,21 +115,16 @@ jobs:
           echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env
           timeout --kill-after=30 300 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-h
           echo "Finished starting up langgraph-test-h"
-          LANGGRAPH_VERSION=$(docker run --rm --entrypoint "" langgraph-test-h python -c "import sys; from importlib.metadata import version; v = version('langgraph'); print(v);")
-          if [ "$LANGGRAPH_VERSION" != "1.2.6" ]; then
-            echo "LANGGRAPH_VERSION != 1.2.6; $LANGGRAPH_VERSION"
-            exit 1
-          fi
-          LANGCHAIN_OPENAI_VERSION=$(docker run --rm --entrypoint "" langgraph-test-h python -c "import sys; from importlib.metadata import version; v = version('langchain-openai'); print(v);")
-          if [ "$LANGCHAIN_OPENAI_VERSION" != "1.1.14" ]; then
-            echo "LANGCHAIN_OPENAI_VERSION != 1.1.14; $LANGCHAIN_OPENAI_VERSION"
-            exit 1
-          fi
-          LANGCHAIN_ANTHROPIC_VERSION=$(docker run --rm --entrypoint "" langgraph-test-h python -c "import sys; from importlib.metadata import version; v = version('langchain-anthropic'); print(v);")
-          if [ "$LANGCHAIN_ANTHROPIC_VERSION" != "1.4.6" ]; then
-            echo "LANGCHAIN_ANTHROPIC_VERSION != 1.4.6; $LANGCHAIN_ANTHROPIC_VERSION"
-            exit 1
-          fi
+          pins=$(grep -oE '"[a-z0-9-]+==[^"]+"' pyproject.toml | tr -d '"')
+          [ -n "$pins" ] || { echo "no == pins found in pyproject.toml"; exit 1; }
+          for pin in $pins; do
+            pkg=${pin%%==*}
+            installed=$(docker run --rm --entrypoint "" langgraph-test-h python -c "from importlib.metadata import version; print(version('$pkg'))")
+            if [ "$installed" != "${pin#*==}" ]; then
+              echo "$pkg $installed != pinned ${pin#*==}"
+              exit 1
+            fi
+          done
 
       - name: Build and test prerelease reqs fail service
         if: ${{ (steps.changed-files.outputs.all || github.event_name == 'workflow_dispatch') &&matrix.example.name == 'A' }}

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

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import copy
-import json
 import logging
 from collections.abc import AsyncIterator, Collection, Iterator, Mapping, Sequence
 from typing import (
@@ -27,8 +26,6 @@ from langgraph.checkpoint.serde.types import (
     RESUME,
     SCHEDULED,
     ChannelProtocol,
-    LogAppend,
-    LogStep,
 )
 
 V = TypeVar("V", int, float, str)
@@ -185,93 +182,6 @@ class DeltaChannelHistory(TypedDict):
     seed: NotRequired[Any]
 
 
-class LogPage(NamedTuple):
-    """A page of a log channel, as returned by `BaseCheckpointSaver.read_log`.
-
-    !!! warning "Prototype"
-
-        Part of the transcript log prototype. Shape and semantics may change.
-    """
-
-    entries: list[tuple[int, Any]]
-    """`(position, entry)` pairs in path order. Entries are JSON values."""
-    length: int
-    """Number of committed entries on the path to the checkpoint's leaf."""
-    pending: list[Any]
-    """Entries written by finished tasks of the step in progress at the head,
-    not committed yet. Empty unless the checkpoint read is the thread's latest."""
-
-
-def log_range(
-    length: int, after: int | None, before: int | None, limit: int | None
-) -> tuple[int, int]:
-    """The position range `[lo, hi)` a `read_log` call asks for.
-
-    `after=p` starts after position `p`; `before=p` stops before it. With a
-    `limit`, a read with only `after` takes the first `limit` entries of the
-    range; any other read takes the last `limit` (chat UIs page from the end).
-    """
-    lo = 0 if after is None else max(after + 1, 0)
-    hi = length if before is None else min(before, length)
-    if limit is not None:
-        if after is not None and before is None:
-            hi = min(hi, lo + limit)
-        else:
-            lo = max(lo, hi - limit)
-    return lo, max(lo, hi)
-
-
-def _without_checkpoint_id(config: RunnableConfig) -> RunnableConfig:
-    conf = {k: v for k, v in config["configurable"].items() if k != "checkpoint_id"}
-    return {**config, "configurable": conf}
-
-
-def _pending_log_entries(tup: CheckpointTuple, channel: str) -> list[Any]:
-    return [
-        json.loads(text)
-        for _, ch, value in tup.pending_writes or ()
-        if ch == channel and isinstance(value, LogAppend)
-        for text in value.entries
-    ]
-
-
-def _step_entries(
-    step: LogStep, parent: CheckpointTuple | None, channel: str
-) -> list[str]:
-    """The JSON texts of one step: inline, or the `LogAppend` writes stored on
-    the checkpoint the step started from."""
-    if step.entries is not None:
-        texts = list(step.entries)
-    else:
-        texts = [
-            text
-            for _, ch, value in (parent.pending_writes or () if parent else ())
-            if ch == channel
-            and isinstance(value, LogAppend)
-            and value.start == step.start
-            for text in value.entries
-        ]
-    if len(texts) != step.count:
-        raise ValueError(
-            f"log channel {channel!r}: the step at position {step.start} "
-            f"committed {step.count} entries, found {len(texts)}"
-        )
-    return texts
-
-
-def _page_entries(
-    steps: Sequence[tuple[int, Sequence[str]]], lo: int, hi: int
-) -> list[tuple[int, Any]]:
-    """Positions `[lo, hi)` from steps given as `(start, texts)`, any order."""
-    out: list[tuple[int, Any]] = []
-    for start, texts in sorted(steps, key=lambda s: s[0]):
-        for i, text in enumerate(texts):
-            pos = start + i
-            if lo <= pos < hi:
-                out.append((pos, json.loads(text)))
-    return out
-
-
 def writes_sort_key(
     task_path: str, task_id: str = "", idx: int = 0
 ) -> tuple[str, str, int]:
@@ -321,14 +231,6 @@ class BaseCheckpointSaver(Generic[V]):
 
     serde: SerializerProtocol = JsonPlusSerializer()
 
-    supports_log: bool = False
-    """Whether this saver stores log channels natively (prototype).
-
-    A saver without native support still works: it stores `LogAppend` writes
-    and `LogStep` values like any other, and the default `read_log` walks the
-    parent chain to read them back.
-    """
-
     def __init__(
         self,
         *,
@@ -702,109 +604,6 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    def read_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        """Read a log channel as of the checkpoint `config` names (or the latest).
-
-        !!! warning "Prototype"
-
-            Part of the transcript log prototype. Shape and semantics may change.
-
-        The log as of a checkpoint is the path from the root to the leaf its
-        channel version points at. See `log_range` for `after`, `before` and
-        `limit`.
-
-        The default implementation walks the parent chain through `get_tuple`:
-        a step's entries are inline in its `LogStep`, or are the `LogAppend`
-        writes stored on the checkpoint the step started from. It is correct
-        and costs one `get_tuple` per checkpoint walked. Savers with direct
-        storage access override it with range reads.
-        """
-        tup = self.get_tuple(config)
-        if tup is None:
-            return LogPage([], 0, [])
-        latest = self.get_tuple(_without_checkpoint_id(tup.config))
-        is_head = latest is not None and latest.checkpoint["id"] == tup.checkpoint["id"]
-        pending = _pending_log_entries(tup, channel) if is_head else []
-        marker = tup.checkpoint["channel_values"].get(channel)
-        if not isinstance(marker, LogStep):
-            return LogPage([], 0, pending)
-        lo, hi = log_range(marker.end, after, before, limit)
-        steps: list[tuple[int, list[str]]] = []
-        cur: CheckpointTuple | None = tup
-        while cur is not None:
-            version = cur.checkpoint["channel_versions"].get(channel)
-            step = cur.checkpoint["channel_values"].get(channel)
-            if version is None or not isinstance(step, LogStep) or step.end <= lo:
-                break
-            # climb to the checkpoint that committed this version: later ones
-            # carry its marker without inline entries
-            parent = self.get_tuple(cur.parent_config) if cur.parent_config else None
-            while (
-                parent is not None
-                and parent.checkpoint["channel_versions"].get(channel) == version
-            ):
-                step = parent.checkpoint["channel_values"].get(channel, step)
-                parent = (
-                    self.get_tuple(parent.parent_config)
-                    if parent.parent_config
-                    else None
-                )
-            steps.append((step.start, _step_entries(step, parent, channel)))
-            cur = parent
-        return LogPage(_page_entries(steps, lo, hi), marker.end, pending)
-
-    async def aread_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        """Async version of `read_log`."""
-        tup = await self.aget_tuple(config)
-        if tup is None:
-            return LogPage([], 0, [])
-        latest = await self.aget_tuple(_without_checkpoint_id(tup.config))
-        is_head = latest is not None and latest.checkpoint["id"] == tup.checkpoint["id"]
-        pending = _pending_log_entries(tup, channel) if is_head else []
-        marker = tup.checkpoint["channel_values"].get(channel)
-        if not isinstance(marker, LogStep):
-            return LogPage([], 0, pending)
-        lo, hi = log_range(marker.end, after, before, limit)
-        steps: list[tuple[int, list[str]]] = []
-        cur: CheckpointTuple | None = tup
-        while cur is not None:
-            version = cur.checkpoint["channel_versions"].get(channel)
-            step = cur.checkpoint["channel_values"].get(channel)
-            if version is None or not isinstance(step, LogStep) or step.end <= lo:
-                break
-            parent = (
-                await self.aget_tuple(cur.parent_config) if cur.parent_config else None
-            )
-            while (
-                parent is not None
-                and parent.checkpoint["channel_versions"].get(channel) == version
-            ):
-                step = parent.checkpoint["channel_values"].get(channel, step)
-                parent = (
-                    await self.aget_tuple(parent.parent_config)
-                    if parent.parent_config
-                    else None
-                )
-            steps.append((step.start, _step_entries(step, parent, channel)))
-            cur = parent
-        return LogPage(_page_entries(steps, lo, hi), marker.end, pending)
-
     def get_delta_channel_history(
         self, *, config: RunnableConfig, channels: Sequence[str]
     ) -> Mapping[str, DeltaChannelHistory]:

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

@@ -21,15 +21,12 @@ from langgraph.checkpoint.base import (
     CheckpointMetadata,
     CheckpointTuple,
     DeltaChannelHistory,
-    LogPage,
     PendingWrite,
     SerializerProtocol,
     get_checkpoint_id,
     get_checkpoint_metadata,
     writes_sort_key,
 )
-from langgraph.checkpoint.memory._log import InMemoryLog
-from langgraph.checkpoint.serde.types import LogAppend, LogStep
 
 logger = logging.getLogger(__name__)
 
@@ -69,8 +66,6 @@ class InMemorySaver(
         ```
     """
 
-    supports_log = True
-
     # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
     storage: defaultdict[
         str,
@@ -98,7 +93,6 @@ class InMemorySaver(
         self.storage = factory(lambda: defaultdict(dict))
         self.writes = factory(dict)
         self.blobs = factory()
-        self.log = InMemoryLog()
         self.stack = ExitStack()
         if factory is not defaultdict:
             self.stack.enter_context(self.storage)  # type: ignore[arg-type]
@@ -137,11 +131,6 @@ class InMemorySaver(
     ) -> dict[str, Any]:
         result: dict[str, Any] = {}
         for k, ver in versions.items():
-            if (
-                marker := self.log.marker(thread_id, checkpoint_ns, k, ver)
-            ) is not None:
-                result[k] = marker
-                continue
             kk = (thread_id, checkpoint_ns, k, ver)
             if kk not in self.blobs:
                 continue
@@ -160,87 +149,6 @@ class InMemorySaver(
             for k in sorted(stored, key=lambda k: writes_sort_key(stored[k][3], *k))
         ]
 
-    def _pending_writes(
-        self, thread_id: str, checkpoint_ns: str, checkpoint_id: str
-    ) -> list[PendingWrite]:
-        """A checkpoint's writes, including its uncommitted log writes, in
-        `writes_sort_key` order."""
-        stored = self.writes.get((thread_id, checkpoint_ns, checkpoint_id), {})
-        rows: list[tuple[tuple[str, str, int], PendingWrite]] = [
-            (
-                writes_sort_key(v[3], *k),
-                (v[0], v[1], self.serde.loads_typed(v[2])),
-            )
-            for k, v in stored.items()
-        ]
-        for task_path, task_id, channel, idx, value in self.log.pending_writes(
-            thread_id, checkpoint_ns, checkpoint_id
-        ):
-            rows.append(
-                (writes_sort_key(task_path, task_id, idx), (task_id, channel, value))
-            )
-        rows.sort(key=lambda r: r[0])
-        return [w for _, w in rows]
-
-    def _version(
-        self,
-        thread_id: str,
-        checkpoint_ns: str,
-        checkpoint_id: str | None,
-        channel: str,
-    ) -> Any:
-        if checkpoint_id is None:
-            return None
-        saved = (
-            self.storage.get(thread_id, {}).get(checkpoint_ns, {}).get(checkpoint_id)
-        )
-        if saved is None:
-            return None
-        version = self.serde.loads_typed(saved[0])["channel_versions"].get(channel)
-        if (thread_id, checkpoint_ns, channel, version) in self.blobs:
-            # stored before the field became a log channel: its log is empty
-            return None
-        return version
-
-    def read_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        """Read a log channel with range reads over its segment chain."""
-        thread_id = config["configurable"]["thread_id"]
-        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-        ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
-        latest = max(ns_storage, default=None)
-        checkpoint_id = get_checkpoint_id(config) or latest
-        if checkpoint_id is None or checkpoint_id not in ns_storage:
-            return LogPage([], 0, [])
-        return self.log.read(
-            thread_id,
-            checkpoint_ns,
-            channel,
-            self._version(thread_id, checkpoint_ns, checkpoint_id, channel),
-            checkpoint_id if checkpoint_id == latest else None,
-            after=after,
-            before=before,
-            limit=limit,
-        )
-
-    async def aread_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        return self.read_log(config, channel, after=after, before=before, limit=limit)
-
     def get_delta_channel_history(
         self, *, config: RunnableConfig, channels: Sequence[str]
     ) -> Mapping[str, DeltaChannelHistory]:
@@ -347,6 +255,7 @@ class InMemorySaver(
         if checkpoint_id := get_checkpoint_id(config):
             if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                 checkpoint, metadata, parent_checkpoint_id = saved
+                writes = self._ordered_writes(thread_id, checkpoint_ns, checkpoint_id)
                 checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
                 return CheckpointTuple(
                     config=config,
@@ -357,9 +266,9 @@ class InMemorySaver(
                         ),
                     },
                     metadata=self.serde.loads_typed(metadata),
-                    pending_writes=self._pending_writes(
-                        thread_id, checkpoint_ns, checkpoint_id
-                    ),
+                    pending_writes=[
+                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
+                    ],
                     parent_config=(
                         {
                             "configurable": {
@@ -376,6 +285,7 @@ class InMemorySaver(
             if checkpoints := self.storage[thread_id][checkpoint_ns]:
                 checkpoint_id = max(checkpoints.keys())
                 checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
+                writes = self._ordered_writes(thread_id, checkpoint_ns, checkpoint_id)
                 checkpoint_ = self.serde.loads_typed(checkpoint)
                 return CheckpointTuple(
                     config={
@@ -392,9 +302,9 @@ class InMemorySaver(
                         ),
                     },
                     metadata=self.serde.loads_typed(metadata),
-                    pending_writes=self._pending_writes(
-                        thread_id, checkpoint_ns, checkpoint_id
-                    ),
+                    pending_writes=[
+                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
+                    ],
                     parent_config=(
                         {
                             "configurable": {
@@ -478,6 +388,10 @@ class InMemorySaver(
                     elif limit is not None:
                         limit -= 1
 
+                    writes = self._ordered_writes(
+                        thread_id, checkpoint_ns, checkpoint_id
+                    )
+
                     checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
 
                     yield CheckpointTuple(
@@ -508,9 +422,9 @@ class InMemorySaver(
                             if parent_checkpoint_id
                             else None
                         ),
-                        pending_writes=self._pending_writes(
-                            thread_id, checkpoint_ns, checkpoint_id
-                        ),
+                        pending_writes=[
+                            (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
+                        ],
                     )
 
     def put(
@@ -537,22 +451,8 @@ class InMemorySaver(
         c = checkpoint.copy()
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"]["checkpoint_ns"]
-        parent_id = config["configurable"].get("checkpoint_id")
         values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
         for k, v in new_versions.items():
-            if isinstance(values.get(k), LogStep):
-                # a log channel: the step row is its "blob" at this version
-                self.log.put_step(
-                    thread_id,
-                    checkpoint_ns,
-                    k,
-                    self._version(thread_id, checkpoint_ns, parent_id, k),
-                    parent_id,
-                    checkpoint["id"],
-                    v,
-                    values[k],
-                )
-                continue
             self.blobs[(thread_id, checkpoint_ns, k, v)] = (
                 self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
             )
@@ -600,19 +500,6 @@ class InMemorySaver(
         outer_key = (thread_id, checkpoint_ns, checkpoint_id)
         outer_writes_ = self.writes.get(outer_key)
         for idx, (c, v) in enumerate(writes):
-            if isinstance(v, LogAppend):
-                self.log.put_append(
-                    thread_id,
-                    checkpoint_ns,
-                    c,
-                    self._version(thread_id, checkpoint_ns, checkpoint_id, c),
-                    checkpoint_id,
-                    task_id,
-                    task_path,
-                    idx,
-                    v,
-                )
-                continue
             inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
             if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_:
                 continue
@@ -641,72 +528,6 @@ class InMemorySaver(
         for k in list(self.blobs.keys()):
             if k[0] == thread_id:
                 del self.blobs[k]
-        self.log.delete_thread(thread_id)
-
-    def copy_thread(self, source_thread_id: str, target_thread_id: str) -> None:
-        """Copy all checkpoints, writes, blobs and log rows to another thread."""
-        for ns, checkpoints in self.storage.get(source_thread_id, {}).items():
-            self.storage[target_thread_id][ns] = dict(checkpoints)
-        for (t, ns, cid), rows in list(self.writes.items()):
-            if t == source_thread_id:
-                self.writes[(target_thread_id, ns, cid)] = dict(rows)
-        for (t, ns, ch, ver), blob in list(self.blobs.items()):
-            if t == source_thread_id:
-                self.blobs[(target_thread_id, ns, ch, ver)] = blob
-        self.log.copy_thread(source_thread_id, target_thread_id)
-
-    def prune(
-        self, thread_ids: Sequence[str], *, strategy: str = "keep_latest"
-    ) -> None:
-        """Prune checkpoints; the log keeps what remaining checkpoints reach.
-
-        `"keep_latest"` keeps the latest checkpoint per namespace, `"delete"`
-        deletes every checkpoint. Writes and blobs no remaining checkpoint
-        uses are deleted with them.
-        """
-        if strategy not in ("keep_latest", "delete"):
-            raise ValueError(f"Unknown prune strategy: {strategy}")
-        for thread_id in thread_ids:
-            keep = {
-                ns: ({max(cps)} if strategy == "keep_latest" and cps else set())
-                for ns, cps in self.storage.get(thread_id, {}).items()
-            }
-            self._retain(thread_id, keep)
-
-    def delete_for_runs(self, run_ids: Sequence[str]) -> None:
-        """Delete the checkpoints the given runs created; the log keeps what
-        the remaining checkpoints reach."""
-        runs = set(run_ids)
-        for thread_id, namespaces in list(self.storage.items()):
-            keep = {
-                ns: {
-                    cid
-                    for cid, saved in cps.items()
-                    if self.serde.loads_typed(saved[1]).get("run_id") not in runs
-                }
-                for ns, cps in namespaces.items()
-            }
-            self._retain(thread_id, keep)
-
-    def _retain(self, thread_id: str, keep: dict[str, set[str]]) -> None:
-        namespaces = self.storage.get(thread_id, {})
-        for ns, cps in list(namespaces.items()):
-            for cid in [c for c in cps if c not in keep.get(ns, set())]:
-                del cps[cid]
-                self.writes.pop((thread_id, ns, cid), None)
-        used = {
-            (thread_id, ns, ch, ver)
-            for ns, cps in namespaces.items()
-            for saved in cps.values()
-            for ch, ver in self.serde.loads_typed(saved[0])["channel_versions"].items()
-        }
-        for k in [k for k in self.blobs if k[0] == thread_id and k not in used]:
-            del self.blobs[k]
-        self.log.collect(
-            thread_id,
-            {ns: dict.fromkeys(cps) for ns, cps in namespaces.items()},
-            lambda ns, cid, ch: self._version(thread_id, ns, cid, ch),
-        )
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Asynchronous version of `get_tuple`.
@@ -798,17 +619,6 @@ class InMemorySaver(
         """
         return self.delete_thread(thread_id)
 
-    async def acopy_thread(self, source_thread_id: str, target_thread_id: str) -> None:
-        return self.copy_thread(source_thread_id, target_thread_id)
-
-    async def aprune(
-        self, thread_ids: Sequence[str], *, strategy: str = "keep_latest"
-    ) -> None:
-        return self.prune(thread_ids, strategy=strategy)
-
-    async def adelete_for_runs(self, run_ids: Sequence[str]) -> None:
-        return self.delete_for_runs(run_ids)
-
     def get_next_version(self, current: str | None, channel: None) -> str:
         if current is None:
             current_v = 0

libs/checkpoint/langgraph/checkpoint/memory/_log.py

@@ -1,401 +0,0 @@
-"""Log channel storage for `InMemorySaver` (transcript log prototype).
-
-The reference implementation of the log storage design: entries stored once,
-keyed in path order; one step row per committing channel version; segments
-for linear runs, with parent pointers between them. Reads are range reads
-over a segment chain, and lifecycle is reachability from the checkpoints that
-remain.
-
-Rows, per `(thread_id, checkpoint_ns, channel)`:
-
-- segment: `seg -> (parent_seg | None, fork_pos)`
-- step: `version -> StepRow(seg, start, count, batch)`, unique on `(seg, start)`
-- entry: `seg -> {(start, batch, task_path, task_id, idx, j): json_text}`
-
-`batch` names the execution that wrote a step's entries: the checkpoint the
-step started from for entries that came through `put_writes`, the committing
-checkpoint for entries a `LogStep` carried inline.
-"""
-
-from __future__ import annotations
-
-import json
-import threading
-import uuid
-from collections import defaultdict
-from collections.abc import Callable, Iterable
-from dataclasses import dataclass, field
-from typing import Any
-
-from langgraph.checkpoint.base import LogPage, log_range, writes_sort_key
-from langgraph.checkpoint.serde.types import LogAppend, LogStep
-
-_SEGMENT_NS = uuid.UUID("6f6c6f67-0000-4000-8000-6c6f67736567")
-
-EntryKey = tuple[int, str, str, str, int, int]
-
-
-@dataclass
-class StepRow:
-    seg: str
-    start: int
-    count: int
-    batch: str
-    exec: str
-
-    @property
-    def end(self) -> int:
-        return self.start + self.count
-
-
-@dataclass
-class ChannelLog:
-    segments: dict[str, tuple[str | None, int]] = field(default_factory=dict)
-    steps: dict[str, StepRow] = field(default_factory=dict)
-    step_at: dict[tuple[str, int], str] = field(default_factory=dict)
-    entries: defaultdict[str, dict[EntryKey, str]] = field(
-        default_factory=lambda: defaultdict(dict)
-    )
-    batch_segments: defaultdict[str, list[str]] = field(
-        default_factory=lambda: defaultdict(list)
-    )
-
-    def copy(self) -> ChannelLog:
-        new = ChannelLog()
-        new.segments = dict(self.segments)
-        new.steps = {
-            v: StepRow(r.seg, r.start, r.count, r.batch, r.exec)
-            for v, r in self.steps.items()
-        }
-        new.step_at = dict(self.step_at)
-        for seg, rows in self.entries.items():
-            new.entries[seg] = dict(rows)
-        for batch, segs in self.batch_segments.items():
-            new.batch_segments[batch] = list(segs)
-        return new
-
-    # -- structure ---------------------------------------------------------
-
-    def committed(self, seg: str) -> bool:
-        return any(s == seg for s, _ in self.step_at)
-
-    def tip(self, seg: str) -> int:
-        ends = [self.steps[v].end for (s, _), v in self.step_at.items() if s == seg]
-        return max(ends) if ends else self.segments[seg][1]
-
-    def chain(self, seg: str, upto: int) -> list[tuple[str, int]]:
-        """`(segment, upto)` pairs from the leaf's segment to the root."""
-        out: list[tuple[str, int]] = []
-        cur: str | None = seg
-        while cur is not None:
-            out.append((cur, upto))
-            cur, upto = self.segments[cur]
-        return out
-
-    def segment_for(self, leaf: tuple[str | None, int], batch: str, exec_: str) -> str:
-        """The fork rule: continue the leaf's segment if the leaf is its tip;
-        otherwise use a new segment forking at the leaf, reusing one this
-        batch already started and nothing committed to."""
-        seg, end = leaf
-        if seg is not None and self.tip(seg) == end:
-            return seg
-        for cand in self.batch_segments.get(batch, ()):
-            if self.segments[cand] == leaf and not self.committed(cand):
-                return cand
-        new = str(
-            uuid.uuid5(
-                _SEGMENT_NS, f"{batch}:{exec_}:{len(self.batch_segments[batch])}"
-            )
-        )
-        self.segments[new] = leaf
-        self.batch_segments[batch].append(new)
-        return new
-
-    def step_texts(self, seg: str, start: int, batch: str) -> list[str]:
-        rows = self.entries.get(seg, {})
-        keys = sorted(
-            (k for k in rows if k[0] == start and k[1] == batch),
-            key=lambda k: (writes_sort_key(k[2], k[3], k[4]), k[5]),
-        )
-        return [rows[k] for k in keys]
-
-    def replace_write(
-        self,
-        seg: str,
-        start: int,
-        batch: str,
-        task_path: str,
-        task_id: str,
-        idx: int,
-        texts: Iterable[str],
-    ) -> None:
-        rows = self.entries[seg]
-        for k in [k for k in rows if k[:5] == (start, batch, task_path, task_id, idx)]:
-            del rows[k]
-        for j, text in enumerate(texts):
-            rows[(start, batch, task_path, task_id, idx, j)] = text
-
-    def is_committed_step(self, seg: str, start: int, batch: str) -> bool:
-        v = self.step_at.get((seg, start))
-        return v is not None and self.steps[v].batch == batch
-
-    def pending(self, batch: str) -> list[tuple[str, str, int, LogAppend, str]]:
-        """Uncommitted writes stored under `batch`, as
-        `(task_path, task_id, idx, LogAppend, exec)`, in write order."""
-        groups: dict[tuple[str, int, str, str, int], list[tuple[int, str]]] = (
-            defaultdict(list)
-        )
-        for seg, rows in self.entries.items():
-            for (start, b, task_path, task_id, idx, j), text in rows.items():
-                if b != batch or self.is_committed_step(seg, start, b):
-                    continue
-                groups[(seg, start, task_path, task_id, idx)].append((j, text))
-        out = []
-        for (_, start, task_path, task_id, idx), items in groups.items():
-            texts = tuple(t for _, t in sorted(items))
-            out.append((task_path, task_id, idx, LogAppend(start, texts), ""))
-        out.sort(key=lambda w: writes_sort_key(w[0], w[1], w[2]))
-        return out
-
-
-class InMemoryLog:
-    """Log storage for one `InMemorySaver`."""
-
-    def __init__(self) -> None:
-        self.logs: dict[tuple[str, str, str], ChannelLog] = {}
-        self.lock = threading.RLock()
-
-    def _log(self, thread_id: str, ns: str, channel: str) -> ChannelLog:
-        key = (thread_id, ns, channel)
-        if key not in self.logs:
-            self.logs[key] = ChannelLog()
-        return self.logs[key]
-
-    def is_log_version(
-        self, thread_id: str, ns: str, channel: str, version: Any
-    ) -> bool:
-        log = self.logs.get((thread_id, ns, channel))
-        return log is not None and version in log.steps
-
-    def marker(
-        self, thread_id: str, ns: str, channel: str, version: Any
-    ) -> LogStep | None:
-        log = self.logs.get((thread_id, ns, channel))
-        if log is None or version not in log.steps:
-            return None
-        row = log.steps[version]
-        return LogStep(row.start, row.count, row.exec)
-
-    def leaf(
-        self, thread_id: str, ns: str, channel: str, version: Any
-    ) -> tuple[str | None, int]:
-        if version is None:
-            return (None, 0)
-        log = self.logs.get((thread_id, ns, channel))
-        if log is None or version not in log.steps:
-            raise ValueError(
-                f"log channel {channel!r}: version {version} has no step row"
-            )
-        row = log.steps[version]
-        return (row.seg, row.end)
-
-    # -- writes ------------------------------------------------------------
-
-    def put_append(
-        self,
-        thread_id: str,
-        ns: str,
-        channel: str,
-        parent_version: Any,
-        batch: str,
-        task_id: str,
-        task_path: str,
-        idx: int,
-        value: LogAppend,
-    ) -> None:
-        with self.lock:
-            log = self._log(thread_id, ns, channel)
-            leaf = self.leaf(thread_id, ns, channel, parent_version)
-            if value.start != leaf[1]:
-                raise ValueError(
-                    f"log channel {channel!r}: append at {value.start}, but the "
-                    f"checkpoint's log has {leaf[1]} entries"
-                )
-            seg = log.segment_for(leaf, batch, value.exec)
-            if log.is_committed_step(seg, value.start, batch):
-                return  # already committed: a reused write, nothing to do
-            log.replace_write(
-                seg, value.start, batch, task_path, task_id, idx, value.entries
-            )
-
-    def put_step(
-        self,
-        thread_id: str,
-        ns: str,
-        channel: str,
-        parent_version: Any,
-        parent_id: str | None,
-        checkpoint_id: str,
-        version: Any,
-        step: LogStep,
-    ) -> None:
-        with self.lock:
-            log = self._log(thread_id, ns, channel)
-            leaf = self.leaf(thread_id, ns, channel, parent_version)
-            if step.start != leaf[1]:
-                raise ValueError(
-                    f"log channel {channel!r}: step starts at {step.start}, but "
-                    f"the parent checkpoint's log has {leaf[1]} entries"
-                )
-            inline = step.entries is not None
-            batch = checkpoint_id if inline or parent_id is None else parent_id
-            seg = log.segment_for(leaf, batch, step.exec)
-            if inline:
-                log.replace_write(seg, step.start, batch, "", "", 0, step.entries)
-            texts = log.step_texts(seg, step.start, batch)
-            if len(texts) != step.count:
-                raise ValueError(
-                    f"log channel {channel!r}: the step at {step.start} commits "
-                    f"{step.count} entries, but {len(texts)} were written"
-                )
-            existing = log.step_at.get((seg, step.start))
-            if existing is not None and existing != version:
-                raise ValueError(
-                    f"log channel {channel!r}: position {step.start} of segment "
-                    f"{seg} was already committed by another step (concurrent "
-                    "writers on one checkpoint)"
-                )
-            log.steps[version] = StepRow(seg, step.start, step.count, batch, step.exec)
-            log.step_at[(seg, step.start)] = version
-
-    # -- reads -------------------------------------------------------------
-
-    def read(
-        self,
-        thread_id: str,
-        ns: str,
-        channel: str,
-        version: Any,
-        batch: str | None,
-        *,
-        after: int | None,
-        before: int | None,
-        limit: int | None,
-    ) -> LogPage:
-        with self.lock:
-            log = self.logs.get((thread_id, ns, channel))
-            pending: list[Any] = []
-            if log is not None and batch is not None:
-                pending = [
-                    json.loads(text)
-                    for *_, value, _ in log.pending(batch)
-                    for text in value.entries
-                ]
-            if log is None or version is None or version not in log.steps:
-                return LogPage([], 0, pending)
-            row = log.steps[version]
-            lo, hi = log_range(row.end, after, before, limit)
-            out: list[tuple[int, Any]] = []
-            for seg, upto in log.chain(row.seg, row.end):
-                if upto <= lo:
-                    break
-                starts = sorted(s for (sg, s) in log.step_at if sg == seg and s < upto)
-                for start in starts:
-                    srow = log.steps[log.step_at[(seg, start)]]
-                    if srow.end <= lo or srow.start >= hi:
-                        continue
-                    texts = log.step_texts(seg, start, srow.batch)
-                    if len(texts) != srow.count:
-                        raise ValueError(
-                            f"log channel {channel!r}: step at {start} has "
-                            f"{len(texts)} entries, expected {srow.count}"
-                        )
-                    for i, text in enumerate(texts):
-                        if lo <= start + i < hi:
-                            out.append((start + i, json.loads(text)))
-            out.sort(key=lambda e: e[0])
-            return LogPage(out, row.end, pending)
-
-    def pending_writes(
-        self, thread_id: str, ns: str, checkpoint_id: str
-    ) -> list[tuple[str, str, str, int, LogAppend]]:
-        """Uncommitted log writes stored on a checkpoint, as
-        `(task_path, task_id, channel, idx, LogAppend)`."""
-        with self.lock:
-            out = []
-            for (t, n, channel), log in self.logs.items():
-                if t != thread_id or n != ns:
-                    continue
-                for task_path, task_id, idx, value, _ in log.pending(checkpoint_id):
-                    out.append((task_path, task_id, channel, idx, value))
-            return out
-
-    # -- lifecycle ---------------------------------------------------------
-
-    def delete_thread(self, thread_id: str) -> None:
-        with self.lock:
-            for key in [k for k in self.logs if k[0] == thread_id]:
-                del self.logs[key]
-
-    def copy_thread(self, source: str, target: str) -> None:
-        with self.lock:
-            for (t, ns, channel), log in list(self.logs.items()):
-                if t == source:
-                    self.logs[(target, ns, channel)] = log.copy()
-
-    def collect(
-        self,
-        thread_id: str,
-        remaining: dict[str, dict[str, Any]],
-        version_of: Callable[[str, str, str], Any],
-    ) -> None:
-        """Keep exactly what remaining checkpoints reach.
-
-        `remaining` maps each namespace to its remaining checkpoint ids;
-        `version_of(ns, checkpoint_id, channel)` returns a checkpoint's
-        version of a channel.
-        """
-        with self.lock:
-            for (t, ns, channel), log in list(self.logs.items()):
-                if t != thread_id:
-                    continue
-                cids = remaining.get(ns, {})
-                reach: dict[str, int] = {}
-                for cid in cids:
-                    version = version_of(ns, cid, channel)
-                    if version is None or version not in log.steps:
-                        continue
-                    row = log.steps[version]
-                    for seg, upto in log.chain(row.seg, row.end):
-                        reach[seg] = max(reach.get(seg, -1), upto)
-                kept_steps = {
-                    v: r
-                    for v, r in log.steps.items()
-                    if r.seg in reach and r.start < reach[r.seg]
-                }
-                kept_keys = {(r.seg, r.start, r.batch) for r in kept_steps.values()}
-                for seg in list(log.entries):
-                    rows = log.entries[seg]
-                    for k in list(rows):
-                        start, batch = k[0], k[1]
-                        if (seg, start, batch) in kept_keys:
-                            continue
-                        if batch in cids and not log.is_committed_step(
-                            seg, start, batch
-                        ):
-                            continue  # pending at a remaining checkpoint
-                        del rows[k]
-                    if not rows:
-                        del log.entries[seg]
-                log.steps = kept_steps
-                log.step_at = {(r.seg, r.start): v for v, r in kept_steps.items()}
-                live = set(reach) | set(log.entries)
-                log.segments = {s: p for s, p in log.segments.items() if s in live}
-                for batch in list(log.batch_segments):
-                    segs = [s for s in log.batch_segments[batch] if s in live]
-                    if segs:
-                        log.batch_segments[batch] = segs
-                    else:
-                        del log.batch_segments[batch]
-                if not log.steps and not log.entries:
-                    del self.logs[(t, ns, channel)]

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -34,8 +34,6 @@ from langgraph.checkpoint.serde import _msgpack as _lg_msgpack
 from langgraph.checkpoint.serde.base import SerializerProtocol
 from langgraph.checkpoint.serde.event_hooks import emit_serde_event
 from langgraph.checkpoint.serde.types import (
-    LogAppend,
-    LogStep,
     SendProtocol,
     _DeltaSnapshot,
 )
@@ -324,29 +322,11 @@ EXT_PYDANTIC_V1 = 4
 EXT_PYDANTIC_V2 = 5
 EXT_NUMPY_ARRAY = 6
 EXT_DELTA_SNAPSHOT = 7
-EXT_LOG_APPEND = 8
-EXT_LOG_STEP = 9
 
 
 def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
     if isinstance(obj, _DeltaSnapshot):
         return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
-    elif isinstance(obj, LogAppend):
-        return ormsgpack.Ext(
-            EXT_LOG_APPEND, _msgpack_enc([obj.start, list(obj.entries), obj.exec])
-        )
-    elif isinstance(obj, LogStep):
-        return ormsgpack.Ext(
-            EXT_LOG_STEP,
-            _msgpack_enc(
-                [
-                    obj.start,
-                    obj.count,
-                    obj.exec,
-                    None if obj.entries is None else list(obj.entries),
-                ]
-            ),
-        )
     elif hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,
@@ -679,14 +659,6 @@ def _create_msgpack_ext_hook(
                     data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS
                 )
             )
-        elif code == EXT_LOG_APPEND:
-            start, entries, exec_ = ormsgpack.unpackb(data)
-            return LogAppend(start, tuple(entries), exec_)
-        elif code == EXT_LOG_STEP:
-            start, count, exec_, entries = ormsgpack.unpackb(data)
-            return LogStep(
-                start, count, exec_, None if entries is None else tuple(entries)
-            )
         elif code == EXT_CONSTRUCTOR_SINGLE_ARG:
             tup = None
             try:

libs/checkpoint/langgraph/checkpoint/serde/types.py

@@ -31,52 +31,6 @@ class _DeltaSnapshot(NamedTuple):
     value: Any
 
 
-class LogAppend(NamedTuple):
-    """One task write to a log channel, as handed to `put_writes`.
-
-    !!! warning "Prototype"
-
-        Part of the transcript log prototype. Shape and semantics may change.
-
-    `start` is the log's length at the checkpoint the step started from (the
-    position the step's entries begin at). `entries` are the appended entries
-    as JSON text, exactly as written. `exec` identifies the execution that
-    wrote them, so a log-aware saver can keep two executions from one
-    checkpoint apart.
-    """
-
-    start: int
-    entries: tuple[str, ...]
-    exec: str = ""
-
-
-class LogStep(NamedTuple):
-    """A log channel's value in a checkpoint: the step that produced its version.
-
-    !!! warning "Prototype"
-
-        Part of the transcript log prototype. Shape and semantics may change.
-
-    The step appended `count` entries starting at position `start`, so the
-    log as of the checkpoint has `start + count` entries. `entries` carries
-    the step's entries inline when they did not reach the saver through
-    `put_writes` (graph input, `update_state`, `durability="exit"`); it is
-    `None` when they did.
-    """
-
-    start: int
-    count: int
-    exec: str = ""
-    entries: tuple[str, ...] | None = None
-
-    @property
-    def end(self) -> int:
-        return self.start + self.count
-
-    def stripped(self) -> "LogStep":
-        return LogStep(self.start, self.count, self.exec, None)
-
-
 Value = TypeVar("Value", covariant=True)
 Update = TypeVar("Update", contravariant=True)
 C = TypeVar("C")

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.40"
+__version__ = "0.4.41"

libs/langgraph/langgraph/_internal/_constants.py

@@ -74,6 +74,8 @@ 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_STRICT_RESUME = sys.intern("__pregel_strict_resume")
+# when True, the root graph refuses a resume that can't be matched to its pending interrupts
 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.
@@ -134,6 +136,7 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     CONFIG_KEY_RESUME_MAP,
+    CONFIG_KEY_STRICT_RESUME,
     CONFIG_KEY_STREAM_MESSAGES_V2,
     # other constants
     PUSH,

libs/langgraph/langgraph/channels/__init__.py

@@ -4,7 +4,6 @@ from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue, LastValueAfterFinish
-from langgraph.channels.log import Log, LogRef, MessagesLog
 from langgraph.channels.named_barrier_value import (
     NamedBarrierValue,
     NamedBarrierValueAfterFinish,
@@ -23,9 +22,6 @@ __all__ = (
     "EphemeralValue",
     "BinaryOperatorAggregate",
     "DeltaChannel",
-    "Log",
-    "LogRef",
-    "MessagesLog",
     "NamedBarrierValue",
     "NamedBarrierValueAfterFinish",
     # topics

libs/langgraph/langgraph/channels/binop.py

@@ -1,5 +1,4 @@
 import collections.abc
-import warnings
 from collections.abc import Callable, Sequence
 from typing import Any, Generic
 
@@ -30,26 +29,6 @@ def _strip_extras(t):  # type: ignore[no-untyped-def]
     return t
 
 
-def _is_add_messages(operator: Callable) -> bool:
-    """Whether `operator` is `add_messages`, alone or with `format=`."""
-    return getattr(operator, "_is_add_messages", False) or getattr(
-        getattr(operator, "func", None), "_is_add_messages", False
-    )
-
-
-def _warn_overwrite_messages() -> None:
-    from langgraph.warnings import LangGraphDeprecatedSinceV13
-
-    warnings.warn(
-        "Replacing all messages with `Overwrite` in an `add_messages` channel "
-        "is deprecated: stored history will be append-only. Build the messages "
-        "the model sees when calling it instead, and use the `MessagesLog` "
-        "channel, which stores messages once instead of in every checkpoint.",
-        category=LangGraphDeprecatedSinceV13,
-        stacklevel=4,
-    )
-
-
 def _get_overwrite(value: Any) -> tuple[bool, Any]:
     """Inspects the given value and returns (is_overwrite, overwrite_value).
 
@@ -161,8 +140,6 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
                         error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
                     )
                     raise InvalidUpdateError(msg)
-                if _is_add_messages(self.operator):
-                    _warn_overwrite_messages()
                 self.value = overwrite_value
                 seen_overwrite = True
                 continue

libs/langgraph/langgraph/channels/log.py

@@ -1,328 +0,0 @@
-"""Append-only log channels (transcript log prototype).
-
-!!! warning "Prototype"
-
-    The transcript log is a prototype. Its API and storage may change.
-
-A log channel is append-only and has no reducer: the only write is "append
-these entries". Its entries are not stored in checkpoints. A checkpoint's
-value for the channel is a marker (`LogStep`) for the step that produced its
-version, and the entries live in the saver's log storage, read with
-`BaseCheckpointSaver.read_log`.
-
-Two channel types share that storage:
-
-- `Log` holds JSON values. Anything that isn't JSON is rejected.
-- `MessagesLog` holds LangChain messages, stored as canonical JSON. Edit ops
-  (`RemoveMessage`, `Overwrite`, re-adding an existing id) are rejected; a
-  graph that needs a different model context builds it as a projection.
-
-A thread whose field held a plain list before it became a log channel (for
-example `Annotated[list, add_messages]` changed to `MessagesLog()`) keeps
-working: the stored list becomes the log's first entries, the "seed". The
-seed is committed with the first checkpoint saved after the change, and
-checkpoints from before the change still read as the list they stored.
-"""
-
-from __future__ import annotations
-
-import json
-import math
-import uuid
-from collections.abc import Sequence
-from typing import Any, NamedTuple
-
-from langgraph.checkpoint.serde.types import LogAppend, LogStep
-from typing_extensions import Self
-
-from langgraph._internal._typing import MISSING
-from langgraph.channels.base import BaseChannel
-from langgraph.channels.binop import _get_overwrite
-from langgraph.errors import EmptyChannelError, InvalidUpdateError
-
-__all__ = ("Log", "LogRef", "MessagesLog")
-
-
-class LogRef(NamedTuple):
-    """What state reads return for a log channel: the log's length at the
-    checkpoint. Read the entries with `read_log`."""
-
-    length: int
-
-
-def _check_json(value: Any, path: str = "entry") -> None:
-    if value is None or isinstance(value, (str, bool, int)):
-        return
-    if isinstance(value, float):
-        if not math.isfinite(value):
-            raise InvalidUpdateError(f"Log {path} is not JSON: {value!r}")
-        return
-    if isinstance(value, list):
-        for i, item in enumerate(value):
-            _check_json(item, f"{path}[{i}]")
-        return
-    if isinstance(value, dict):
-        for k, item in value.items():
-            if not isinstance(k, str):
-                raise InvalidUpdateError(f"Log {path} has a non-string key: {k!r}")
-            _check_json(item, f"{path}.{k}")
-        return
-    raise InvalidUpdateError(
-        f"Log {path} is not JSON: {type(value).__name__} {value!r}"
-    )
-
-
-def _dumps(value: Any) -> str:
-    return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(",", ":"))
-
-
-class Log(BaseChannel[list, list, LogStep]):
-    """An append-only log of JSON entries.
-
-    Write a list of entries to append them. Nodes read the whole log as a
-    list. State reads return a `LogRef` instead of the entries.
-    """
-
-    __slots__ = ("entries", "base", "marker", "loaded", "extra", "seed", "seeding")
-
-    def __init__(self, typ: Any = list) -> None:
-        super().__init__(typ)
-        self.entries: list[Any] = []
-        # length covered by the last committed step (the checkpoint's leaf)
-        self.base = 0
-        self.marker: LogStep | None = None
-        # False when built from a checkpoint without reading the log
-        self.loaded = True
-        # entries appended to a channel that was not loaded
-        self.extra = 0
-        # True while the entries hold a plain list stored before the field
-        # became a log channel, not committed to the log yet
-        self.seed = False
-        # True from the commit that includes the seed until the loop has
-        # stamped that checkpoint's step
-        self.seeding = False
-
-    def __eq__(self, other: object) -> bool:
-        return type(self) is type(other)
-
-    @property
-    def ValueType(self) -> Any:
-        return list
-
-    @property
-    def UpdateType(self) -> Any:
-        return list
-
-    # -- encoding ----------------------------------------------------------
-
-    def encode(self, entry: Any) -> Any:
-        """An entry as the JSON value stored for it."""
-        _check_json(entry)
-        return entry
-
-    def decode(self, data: Any) -> Any:
-        """A stored JSON value as the entry nodes see."""
-        return data
-
-    def coerce(self, value: Any) -> list[Any]:
-        """A write as the list of entries it appends."""
-        if not isinstance(value, list):
-            raise InvalidUpdateError(
-                f"Log channel {self.key!r} takes a list of entries, got "
-                f"{type(value).__name__}"
-            )
-        return value
-
-    def append_value(self, value: Any, exec_: str) -> LogAppend:
-        """A write as the `LogAppend` handed to the saver."""
-        self._reject_overwrite(value)
-        texts = tuple(_dumps(self.encode(e)) for e in self.coerce(value))
-        return LogAppend(self.base, texts, exec_)
-
-    # -- channel protocol --------------------------------------------------
-
-    def _new(self) -> Self:
-        new = self.__class__.__new__(self.__class__)
-        BaseChannel.__init__(new, self.typ, self.key)
-        new.entries = []
-        new.base = 0
-        new.marker = None
-        new.loaded = True
-        new.extra = 0
-        new.seed = False
-        new.seeding = False
-        return new
-
-    def copy(self) -> Self:
-        new = self._new()
-        new.entries = list(self.entries)
-        new.base = self.base
-        new.marker = self.marker
-        new.loaded = self.loaded
-        new.extra = self.extra
-        new.seed = self.seed
-        new.seeding = self.seeding
-        return new
-
-    def from_checkpoint(self, checkpoint: Any) -> Self:
-        """A channel that knows its marker but not its entries.
-
-        State reads build channels this way; `get` then returns a `LogRef`.
-        The loop builds log channels with `from_log` instead.
-
-        A plain list is what the field stored before it became a log channel.
-        The channel then holds it as its seed: uncommitted entries that the
-        next commit writes to the log, ahead of anything appended after.
-        """
-        new = self._new()
-        if isinstance(checkpoint, LogStep):
-            new.marker = checkpoint.stripped()
-            new.base = checkpoint.end
-            new.loaded = False
-        elif isinstance(checkpoint, list):
-            new.entries = self._seed_entries(checkpoint)
-            new.seed = bool(new.entries)
-        elif checkpoint is not MISSING and checkpoint is not None:
-            raise InvalidUpdateError(
-                f"Log channel {self.key!r} can't start from {type(checkpoint).__name__}"
-            )
-        return new
-
-    def _seed_entries(self, stored: list[Any]) -> list[Any]:
-        """The entries a stored plain list becomes."""
-        return list(stored)
-
-    def from_log(self, marker: Any, entries: Sequence[Any]) -> Self:
-        """A loaded channel: the marker, and the entries read from the log."""
-        new = self._new()
-        if isinstance(marker, LogStep):
-            if len(entries) != marker.end:
-                raise ValueError(
-                    f"Log channel {self.key!r}: read {len(entries)} entries, "
-                    f"the checkpoint has {marker.end}"
-                )
-            new.marker = marker.stripped()
-            new.base = marker.end
-        new.entries = [self.decode(e) for e in entries]
-        return new
-
-    def get(self) -> Any:
-        if not self.loaded:
-            return LogRef(self.base + self.extra)
-        if not self.entries:
-            raise EmptyChannelError()
-        return self.entries
-
-    def is_available(self) -> bool:
-        return (not self.loaded and self.base + self.extra > 0) or bool(self.entries)
-
-    def update(self, values: Sequence[Any]) -> bool:
-        if not values:
-            return False
-        appended: list[Any] = []
-        for value in values:
-            if isinstance(value, LogAppend):
-                # a write read back from the saver (pending writes on resume)
-                appended.extend(self.decode(json.loads(t)) for t in value.entries)
-                continue
-            self._reject_overwrite(value)
-            for entry in self.coerce(value):
-                self.encode(entry)  # validate before anything is applied
-                appended.append(entry)
-        if not appended:
-            return False
-        self._check_appends(appended)
-        if not self.loaded:
-            self.extra += len(appended)
-        else:
-            self.entries.extend(appended)
-        return True
-
-    def _check_appends(self, appended: list[Any]) -> None:
-        pass
-
-    def _reject_overwrite(self, value: Any) -> None:
-        if _get_overwrite(value)[0]:
-            raise InvalidUpdateError(
-                f"Log channel {self.key!r} is append-only: Overwrite is not supported"
-            )
-
-    def checkpoint(self) -> Any:
-        return MISSING if self.marker is None else self.marker
-
-    def commit(self) -> LogStep | None:
-        """The marker for a checkpoint about to be saved.
-
-        Entries appended since the last commit become a new `LogStep`
-        (carrying them inline; the loop strips them when they already reached
-        the saver through `put_writes`), and the channel's base moves to its
-        end.
-        """
-        if self.loaded and len(self.entries) > self.base:
-            texts = tuple(_dumps(self.encode(e)) for e in self.entries[self.base :])
-            step = LogStep(self.base, len(self.entries) - self.base, "", texts)
-            self.base = len(self.entries)
-            self.marker = step.stripped()
-            self.seeding, self.seed = self.seed, False
-            return step
-        return self.marker
-
-
-class MessagesLog(Log):
-    """An append-only log of LangChain messages.
-
-    Messages are stored as canonical JSON (`message_to_dict`). Every message
-    has an id; writing an id that is already in the log is an error, as are
-    `RemoveMessage` and `Overwrite`.
-    """
-
-    __slots__ = ()
-
-    def encode(self, entry: Any) -> Any:
-        from langchain_core.messages import BaseMessage, RemoveMessage, message_to_dict
-
-        if isinstance(entry, RemoveMessage):
-            raise InvalidUpdateError(
-                f"MessagesLog {self.key!r} is append-only: RemoveMessage is not "
-                "supported. To give the model a different set of messages, build "
-                "them when calling the model instead of changing stored history."
-            )
-        if not isinstance(entry, BaseMessage):
-            raise InvalidUpdateError(
-                f"MessagesLog {self.key!r} takes messages, got {type(entry).__name__}"
-            )
-        data = message_to_dict(entry)
-        _check_json(data, "message")
-        return data
-
-    def decode(self, data: Any) -> Any:
-        from langchain_core.messages import messages_from_dict
-
-        return messages_from_dict([data])[0]
-
-    def _seed_entries(self, stored: list[Any]) -> list[Any]:
-        # the list `add_messages` stored: already merged, so ids are unique
-        # and there are no `RemoveMessage`s
-        return self.coerce(stored)
-
-    def coerce(self, value: Any) -> list[Any]:
-        from langchain_core.messages import convert_to_messages
-
-        items = value if isinstance(value, list) else [value]
-        messages = convert_to_messages(items)
-        for m in messages:
-            if not m.id:
-                m.id = str(uuid.uuid4())
-        return messages
-
-    def _check_appends(self, appended: list[Any]) -> None:
-        seen = {m.id for m in self.entries} if self.loaded else set()
-        for m in appended:
-            if m.id in seen:
-                raise InvalidUpdateError(
-                    f"MessagesLog {self.key!r} already has a message with id "
-                    f"{m.id!r}: the log is append-only, so a message can't be "
-                    "replaced. Return only new messages from a node, not the "
-                    "whole list."
-                )
-            seen.add(m.id)

libs/langgraph/langgraph/errors.py

@@ -28,6 +28,7 @@ __all__ = (
     "ParentCommand",
     "EmptyInputError",
     "TaskNotFound",
+    "UnknownInterruptError",
     "is_invalid_resume",
 )
 
@@ -146,6 +147,46 @@ class TaskNotFound(Exception):
     pass
 
 
+class UnknownInterruptError(Exception):
+    """Raised in strict resume mode when a resume doesn't match the pending interrupts.
+
+    `reason` says how it doesn't match:
+
+    - `"not_pending"`: the resume names interrupts that aren't pending, listed in
+      `interrupt_ids`.
+    - `"empty_map"`: the resume is an empty map.
+    - `"ambiguous"`: the resume is a single value while several interrupts are
+      pending.
+
+    `pending_ids` lists the interrupts still waiting for an answer. The error is
+    raised before anything is written, so they can still be answered.
+
+    A single value with no interrupt pending is accepted, because it resumes a
+    graph paused at a static breakpoint (`interrupt_before` or `interrupt_after`).
+
+    Each `interrupt()` call in a task has its own id, so an answer retried after
+    the task asked its next question is refused instead of answering that
+    question.
+    """
+
+    reason: Literal["not_pending", "empty_map", "ambiguous"]
+    interrupt_ids: tuple[str, ...]
+    pending_ids: tuple[str, ...]
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        reason: Literal["not_pending", "empty_map", "ambiguous"],
+        interrupt_ids: Sequence[str] = (),
+        pending_ids: Sequence[str] = (),
+    ) -> None:
+        super().__init__(message)
+        self.reason = reason
+        self.interrupt_ids = tuple(interrupt_ids)
+        self.pending_ids = tuple(pending_ids)
+
+
 @dataclass(frozen=True, slots=True)
 class NodeError:
     """Failure context passed to a node-level error handler.

libs/langgraph/langgraph/graph/message.py

@@ -27,7 +27,7 @@ from langgraph.graph.state import StateGraph
 from langgraph.pregel._messages import (
     _messages_delta_reducer as _messages_delta_reducer,
 )
-from langgraph.warnings import LangGraphDeprecatedSinceV10, LangGraphDeprecatedSinceV13
+from langgraph.warnings import LangGraphDeprecatedSinceV10
 
 __all__ = (
     "add_messages",
@@ -41,27 +41,7 @@ Messages = list[MessageLikeRepresentation] | MessageLikeRepresentation
 REMOVE_ALL_MESSAGES = "__remove_all__"
 
 
-_EDIT_DEPRECATION = (
-    "{what} through `add_messages` is deprecated: stored history will be "
-    "append-only. Build the messages the model sees when calling it instead "
-    "(for example from a summary kept in state), and use the `MessagesLog` "
-    "channel, which stores messages once instead of in every checkpoint."
-)
-
-
-def _warn_edit(what: str) -> None:
-    warnings.warn(
-        _EDIT_DEPRECATION.format(what=what),
-        category=LangGraphDeprecatedSinceV13,
-        stacklevel=3,
-    )
-
-
 def _add_messages_wrapper(func: Callable) -> Callable[[Messages, Messages], Messages]:
-    # lets `BinaryOperatorAggregate` recognize the reducer, alone or with
-    # `format=`, without importing this module
-    func._is_add_messages = True  # type: ignore[attr-defined]
-
     def _add_messages(
         left: Messages | None = None, right: Messages | None = None, **kwargs: Any
     ) -> Messages | Callable[[Messages, Messages], Messages]:
@@ -77,7 +57,6 @@ def _add_messages_wrapper(func: Callable) -> Callable[[Messages, Messages], Mess
             return partial(func, **kwargs)
 
     _add_messages.__doc__ = func.__doc__
-    _add_messages._is_add_messages = True  # type: ignore[attr-defined]
     return cast(Callable[[Messages, Messages], Messages], _add_messages)
 
 
@@ -234,11 +213,9 @@ def add_messages(
             remove_all_idx = idx
 
     if remove_all_idx is not None:
-        _warn_edit("Removing all messages with `REMOVE_ALL_MESSAGES`")
         return right[remove_all_idx + 1 :]
 
     # merge
-    left_ids = {m.id for m in left}
     merged = left.copy()
     merged_by_id = {m.id: i for i, m in enumerate(merged)}
     ids_to_remove = set()
@@ -258,10 +235,6 @@ def add_messages(
             merged_by_id[m.id] = len(merged)
             merged.append(m)
     merged = [m for m in merged if m.id not in ids_to_remove]
-    if any(isinstance(m, RemoveMessage) for m in right):
-        _warn_edit("Removing messages with `RemoveMessage`")
-    elif any(m.id in left_ids for m in right):
-        _warn_edit("Replacing a message by id")
 
     if format == "langchain-openai":
         merged = _format_messages(merged)

libs/langgraph/langgraph/pregel/_algo.py

@@ -85,6 +85,7 @@ from langgraph.types import (
     RetryPolicy,
     Send,
     TimeoutPolicy,
+    _interrupt_id,
 )
 
 GetNextVersion = Callable[[V | None, None], V]
@@ -626,7 +627,7 @@ def prepare_single_task(
                 config[CONF].get(CONFIG_KEY_SCRATCHPAD),
                 pending_writes,
                 task_id,
-                xxh3_128_hexdigest(task_checkpoint_ns.encode()),
+                task_checkpoint_ns,
                 config[CONF].get(CONFIG_KEY_RESUME_MAP),
                 step,
                 stop,
@@ -873,7 +874,7 @@ def prepare_push_task_functional(
             configurable.get(CONFIG_KEY_SCRATCHPAD),
             pending_writes,
             task_id,
-            xxh3_128_hexdigest(task_checkpoint_ns.encode()),
+            task_checkpoint_ns,
             configurable.get(CONFIG_KEY_RESUME_MAP),
             step,
             stop,
@@ -1035,7 +1036,7 @@ def prepare_push_task_send(
             config[CONF].get(CONFIG_KEY_SCRATCHPAD),
             pending_writes,
             task_id,
-            xxh3_128_hexdigest(task_checkpoint_ns.encode()),
+            task_checkpoint_ns,
             config[CONF].get(CONFIG_KEY_RESUME_MAP),
             step,
             stop,
@@ -1184,7 +1185,7 @@ def prepare_node_error_handler_task(
         config[CONF].get(CONFIG_KEY_SCRATCHPAD),
         pending_writes,
         task_id,
-        xxh3_128_hexdigest(task_checkpoint_ns.encode()),
+        task_checkpoint_ns,
         config[CONF].get(CONFIG_KEY_RESUME_MAP),
         step,
         stop,
@@ -1280,7 +1281,7 @@ def _scratchpad(
     parent_scratchpad: PregelScratchpad | None,
     pending_writes: list[PendingWrite],
     task_id: str,
-    namespace_hash: str,
+    task_checkpoint_ns: str,
     resume_map: dict[str, Any] | None,
     step: int,
     stop: int,
@@ -1307,10 +1308,15 @@ def _scratchpad(
             task_resume_write = []
         del w
 
-        # find namespace and task-specific resume value
-        if resume_map and namespace_hash in resume_map:
-            mapped_resume_write = resume_map[namespace_hash]
-            task_resume_write.append(mapped_resume_write)
+        # find the resume value for the task's next interrupt
+        if resume_map:
+            next_id = _interrupt_id(task_checkpoint_ns, len(task_resume_write))
+            first_id = _interrupt_id(task_checkpoint_ns, 0)
+            if next_id in resume_map:
+                task_resume_write.append(resume_map[next_id])
+            # Older versions gave all of a task's interrupts the first id.
+            elif first_id in resume_map:
+                task_resume_write.append(resume_map[first_id])
 
     else:
         null_resume_write = None

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -4,7 +4,7 @@ import uuid
 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 (
@@ -16,7 +16,7 @@ from langgraph.checkpoint.base import (
     PendingWrite,
 )
 from langgraph.checkpoint.base.id import uuid6
-from langgraph.checkpoint.serde.types import LogStep, _DeltaSnapshot
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 from langgraph._internal._config import (
     DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT,
@@ -35,7 +35,6 @@ 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.channels.log import Log
 from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
 
 LATEST_VERSION = 4
@@ -326,10 +325,6 @@ def create_checkpoint(
                     bumped[k] = (old, channel_versions[k])
                 values[k] = _DeltaSnapshot(ch.get())
             else:
-                if isinstance(ch, Log):
-                    if (log_step := ch.commit()) is not None:
-                        values[k] = log_step
-                    continue
                 v = ch.checkpoint()
                 if v is not MISSING:
                     values[k] = v
@@ -460,39 +455,12 @@ def _require_saver_for_history(
         )
 
 
-def _log_specs(
-    specs: Mapping[str, BaseChannel], checkpoint: Checkpoint, load_logs: bool
-) -> list[str]:
-    """Log channels whose entries must be read from the saver."""
-    if not load_logs:
-        return []
-    return [
-        k
-        for k, spec in specs.items()
-        if isinstance(spec, Log)
-        and isinstance(checkpoint["channel_values"].get(k), LogStep)
-    ]
-
-
-def _require_saver_for_log(
-    log_channels: list[str],
-    saver: BaseCheckpointSaver | None,
-    config: RunnableConfig | None,
-) -> None:
-    if log_channels and (saver is None or config is None):
-        raise ValueError(
-            f"Log channels {log_channels} have entries to read but no "
-            "checkpointer or config was passed to read them"
-        )
-
-
 def channels_from_checkpoint(
     specs: Mapping[str, BaseChannel | ManagedValueSpec],
     checkpoint: Checkpoint,
     *,
     saver: BaseCheckpointSaver | None = None,
     config: RunnableConfig | None = None,
-    load_logs: bool = True,
 ) -> tuple[Mapping[str, BaseChannel], ManagedValueMapping]:
     """Hydrate channels from a checkpoint.
 
@@ -519,20 +487,12 @@ def channels_from_checkpoint(
         histories = saver.get_delta_channel_history(
             config=config, channels=delta_channels
         )
-    log_channels = _log_specs(channel_specs, checkpoint, load_logs)
-    _require_saver_for_log(log_channels, saver, config)
-    logs: dict[str, list[Any]] = {}
-    if log_channels and saver is not None and config is not None:
-        for k in log_channels:
-            logs[k] = [e for _, e in saver.read_log(config, k).entries]
 
     channels: dict[str, BaseChannel] = {}
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
             ch = _replay(spec, histories[k])
-        elif k in logs:
-            ch = cast(Log, spec).from_log(checkpoint["channel_values"][k], logs[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch
@@ -545,7 +505,6 @@ async def achannels_from_checkpoint(
     *,
     saver: BaseCheckpointSaver | None = None,
     config: RunnableConfig | None = None,
-    load_logs: bool = True,
 ) -> tuple[Mapping[str, BaseChannel], ManagedValueMapping]:
     """Async version of `channels_from_checkpoint`. See docstring there."""
     channel_specs: dict[str, BaseChannel] = {}
@@ -563,20 +522,12 @@ async def achannels_from_checkpoint(
         histories = await saver.aget_delta_channel_history(
             config=config, channels=delta_channels
         )
-    log_channels = _log_specs(channel_specs, checkpoint, load_logs)
-    _require_saver_for_log(log_channels, saver, config)
-    logs: dict[str, list[Any]] = {}
-    if log_channels and saver is not None and config is not None:
-        for k in log_channels:
-            logs[k] = [e for _, e in (await saver.aread_log(config, k)).entries]
 
     channels: dict[str, BaseChannel] = {}
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
             ch = _replay(spec, histories[k])
-        elif k in logs:
-            ch = cast(Log, spec).from_log(checkpoint["channel_values"][k], logs[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch

libs/langgraph/langgraph/pregel/_loop.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import asyncio
 import binascii
 import concurrent.futures
-import json
 from collections import defaultdict, deque
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import (
@@ -13,7 +12,6 @@ from contextlib import (
     ExitStack,
 )
 from datetime import datetime, timezone
-from functools import partial
 from types import TracebackType
 from typing import (
     Any,
@@ -21,7 +19,7 @@ from typing import (
     TypeVar,
     cast,
 )
-from uuid import UUID, uuid4, uuid5
+from uuid import UUID, uuid5
 
 from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
@@ -35,7 +33,6 @@ from langgraph.checkpoint.base import (
     CheckpointTuple,
     PendingWrite,
 )
-from langgraph.checkpoint.serde.types import LogAppend, LogStep
 from langgraph.store.base import BaseStore
 from typing_extensions import ParamSpec, Self
 
@@ -51,6 +48,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_RUNTIME,
     CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_STREAM,
+    CONFIG_KEY_STRICT_RESUME,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_THREAD_ID,
     ERROR,
@@ -75,12 +73,12 @@ from langgraph.callbacks import (
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.binop import _get_overwrite
 from langgraph.channels.delta import DeltaChannel
-from langgraph.channels.log import Log
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import (
     EmptyInputError,
     GraphInterrupt,
+    UnknownInterruptError,
 )
 from langgraph.managed.base import (
     ManagedValueMapping,
@@ -360,9 +358,6 @@ class PregelLoop:
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
         self.durability = durability
-        # identifies this execution's log writes (transcript log prototype)
-        self._log_exec = str(uuid4())
-        self._log_inline_first_step = False
         self._has_graph_lifecycle_callbacks = has_graph_lifecycle_callbacks
         self._graph_lifecycle_events = deque()
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
@@ -520,7 +515,6 @@ class PregelLoop:
         # save writes
         self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
         if self.durability != "exit" and self.checkpointer_put_writes is not None:
-            writes_to_save = self._log_writes_for_saver(writes_to_save)
             config = patch_configurable(
                 self.checkpoint_config,
                 {
@@ -530,21 +524,13 @@ class PregelLoop:
                     CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
                 },
             )
-            # A log write is placed relative to the log of the checkpoint it is
-            # stored on, so that checkpoint must be saved first.
-            put_writes_fn = self.checkpointer_put_writes
-            if any(isinstance(self.specs.get(c), Log) for c, _ in writes_to_save):
-                put_writes_fn = partial(
-                    self._put_writes_after_checkpoint,
-                    getattr(self, "_put_checkpoint_fut", None),
-                )
             if self.checkpointer_put_writes_accepts_task_path:
                 if hasattr(self, "tasks"):
                     task = self.tasks.get(task_id)
                 else:
                     task = None
                 fut = self.submit(
-                    put_writes_fn,
+                    self.checkpointer_put_writes,
                     config,
                     writes_to_save,
                     task_id,
@@ -552,16 +538,13 @@ class PregelLoop:
                 )
             else:
                 fut = self.submit(
-                    put_writes_fn,
+                    self.checkpointer_put_writes,
                     config,
                     writes_to_save,
                     task_id,
                 )
-            # the checkpoint that commits these writes must not be saved before
-            # them: a delta replay or a log step reads them
             if self._delta_write_futs is not None and any(
-                isinstance(self.specs.get(c), (DeltaChannel, Log))
-                for c, _ in writes_to_save
+                isinstance(self.specs.get(c), DeltaChannel) for c, _ in writes_to_save
             ):
                 self._delta_write_futs.append(fut)
             # ERROR_SOURCE_NODE is only appended by commit() when the task
@@ -575,24 +558,6 @@ class PregelLoop:
         if hasattr(self, "tasks"):
             self.output_writes(task_id, writes, cached=cached)
 
-    def _log_write_from_saver(self, channel: str, value: Any) -> Any:
-        """A saved `LogAppend` as the write it was: a list of entries."""
-        spec = self.specs.get(channel)
-        if isinstance(value, LogAppend) and isinstance(spec, Log):
-            return [spec.decode(json.loads(t)) for t in value.entries]
-        return value
-
-    def _log_writes_for_saver(self, writes: WritesT) -> WritesT:
-        """Writes to log channels as the `LogAppend`s handed to the saver."""
-        if not any(isinstance(self.channels.get(c), Log) for c, _ in writes):
-            return writes
-        return [
-            (c, ch.append_value(v, self._log_exec))
-            if isinstance(ch := self.channels.get(c), Log)
-            else (c, v)
-            for c, v in writes
-        ]
-
     def _put_pending_writes(self) -> None:
         if self.checkpointer_put_writes is None:
             return
@@ -616,7 +581,6 @@ class PregelLoop:
                 by_task[task_id].append((channel, value))
         # submit writes to checkpointer
         for task_id, writes in by_task.items():
-            writes = self._log_writes_for_saver(writes)
             if self.checkpointer_put_writes_accepts_task_path and hasattr(
                 self, "tasks"
             ):
@@ -1062,17 +1026,45 @@ class PregelLoop:
                         "Cannot use Command(resume=...) without checkpointer"
                     )
 
+                strict = (
+                    self.config[CONF].get(CONFIG_KEY_STRICT_RESUME, False)
+                    # Nested calls inherit the parent's configurable.
+                    and not self.is_nested
+                    # A finished task can rerun and interrupt again on replay, while
+                    # its earlier output still makes its interrupt look answered.
+                    and not self._addressed_checkpoint_superseded
+                )
                 if resume_is_map := (
                     isinstance(resume, dict)
                     and all(is_xxh3_128_hexdigest(k) for k in resume)
                 ):
+                    if strict:
+                        pending = self._pending_interrupts()
+                        if not resume:
+                            raise UnknownInterruptError(
+                                "The resume map names no interrupt",
+                                reason="empty_map",
+                                pending_ids=sorted(pending),
+                            )
+                        if unknown := sorted(resume.keys() - pending):
+                            raise UnknownInterruptError(
+                                f"Unknown or already-consumed interrupt: {unknown[0]}",
+                                reason="not_pending",
+                                interrupt_ids=unknown,
+                                pending_ids=sorted(pending),
+                            )
                     self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
                 else:
-                    if len(self._pending_interrupts()) > 1:
-                        raise RuntimeError(
+                    if len(pending := self._pending_interrupts()) > 1:
+                        message = (
                             "When there are multiple pending interrupts, you must specify the interrupt id when resuming. "
-                            "Docs: https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation."
+                            "Docs: https://docs.langchain.com/oss/python/langgraph/interrupts#handling-multiple-interrupts."
                         )
+                        if strict:
+                            raise UnknownInterruptError(
+                                message, reason="ambiguous", pending_ids=sorted(pending)
+                            )
+                        raise RuntimeError(message)
 
             writes: defaultdict[str, list[tuple[str, Any]]] = defaultdict(list)
             # group writes by task ID
@@ -1349,8 +1341,6 @@ class PregelLoop:
             stored_versions=self.checkpoint_previous_versions,
             trigger_to_nodes=self.trigger_to_nodes,
         )
-        if do_checkpoint:
-            self._finish_log_steps(metadata)
         for k in channels_to_snapshot:
             new_counters[k] = (0, 0)
         if do_checkpoint:
@@ -1423,51 +1413,6 @@ class PregelLoop:
             # increment step
             self.step += 1
 
-    def _finish_log_steps(self, metadata: CheckpointMetadata) -> None:
-        """Stamp the log steps of the checkpoint about to be saved.
-
-        A step's entries reached the saver through `put_writes` when the
-        checkpoint comes from the loop and writes are saved as they happen;
-        then the step carries no entries. Otherwise (graph input, a fork,
-        `durability="exit"`) it carries them inline.
-
-        A resume at a checkpoint the thread moved past re-runs finished tasks
-        on that same checkpoint, and a saver without native log support keeps
-        the first run's writes under the same keys. So the first step of such
-        a resume carries its entries inline too.
-
-        So does a step that commits a seed (the list a field stored before it
-        became a log channel), which never went through `put_writes`. The
-        channel's version is bumped if the step didn't change it, so that the
-        saver stores the step and the checkpoint points at the log.
-        """
-        inline = (
-            metadata.get("source") != "loop"
-            or self.durability == "exit"
-            or self._log_inline_first_step
-        )
-        if metadata.get("source") == "loop":
-            self._log_inline_first_step = False
-        values = self.checkpoint["channel_values"]
-        for k, ch in self.channels.items():
-            if not isinstance(ch, Log):
-                continue
-            step = values.get(k)
-            seeding, ch.seeding = ch.seeding, False
-            if seeding:
-                versions = self.checkpoint["channel_versions"]
-                if versions.get(k) == self.checkpoint_previous_versions.get(k):
-                    versions[k] = self.checkpointer_get_next_version(
-                        versions.get(k), None
-                    )
-            if isinstance(step, LogStep) and step.entries is not None:
-                values[k] = LogStep(
-                    step.start,
-                    step.count,
-                    self._log_exec,
-                    step.entries if inline or seeding else None,
-                )
-
     def _put_exit_delta_writes(self) -> None:
         """Stage stub + accumulated delta writes so final_checkpoint's put
         waits on them (visibility invariant: both must be durable before
@@ -1794,13 +1739,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             self.checkpointer_put_writes = None
             self.checkpointer_put_writes_accepts_task_path = False
 
-    def _put_writes_after_checkpoint(
-        self, prev: concurrent.futures.Future | None, *args: Any
-    ) -> None:
-        if prev is not None:
-            prev.result()
-        cast(Callable[..., None], self.checkpointer_put_writes)(*args)
-
     def _checkpointer_put_after_previous(
         self,
         prev: concurrent.futures.Future | None,
@@ -1913,10 +1851,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
                 self._addressed_checkpoint_superseded = checkpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )
-                self._log_inline_first_step = (
-                    self._addressed_checkpoint_superseded
-                    and not self.checkpointer.supports_log
-                )
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1961,10 +1895,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (
-            [
-                (str(tid), k, self._log_write_from_saver(k, v))
-                for tid, k, v in saved.pending_writes
-            ]
+            [(str(tid), k, v) for tid, k, v in saved.pending_writes]
             if saved.pending_writes is not None
             else []
         )
@@ -2064,13 +1995,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             self.checkpointer_put_writes = None
             self.checkpointer_put_writes_accepts_task_path = False
 
-    async def _put_writes_after_checkpoint(
-        self, prev: asyncio.Task | None, *args: Any
-    ) -> None:
-        if prev is not None:
-            await prev
-        await cast(Callable[..., Any], self.checkpointer_put_writes)(*args)
-
     async def _checkpointer_put_after_previous(
         self,
         prev: asyncio.Task | None,
@@ -2183,10 +2107,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
                 self._addressed_checkpoint_superseded = await acheckpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )
-                self._log_inline_first_step = (
-                    self._addressed_checkpoint_superseded
-                    and not self.checkpointer.supports_log
-                )
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint
@@ -2231,10 +2151,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (
-            [
-                (str(tid), k, self._log_write_from_saver(k, v))
-                for tid, k, v in saved.pending_writes
-            ]
+            [(str(tid), k, v) for tid, k, v in saved.pending_writes]
             if saved.pending_writes is not None
             else []
         )

libs/langgraph/langgraph/pregel/_messages.py

@@ -26,7 +26,6 @@ from langgraph._internal._config import filter_to_user_tags
 from langgraph._internal._constants import NS_SEP, PUSH
 from langgraph.channels.binop import _get_overwrite
 from langgraph.channels.delta import DeltaChannel
-from langgraph.channels.log import MessagesLog
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
 from langgraph.pregel.protocol import StreamChunk
 from langgraph.types import Command, Overwrite
@@ -538,13 +537,6 @@ def ensure_message_ids(
     """
     for i, (ch, value) in enumerate(writes):
         spec = channels.get(ch)
-        if isinstance(spec, MessagesLog):
-            # ids must be set before the write is saved, so the saved
-            # entries and the run's in-memory messages agree
-            new_value = _with_message_ids(value)
-            if new_value is not value:
-                cast(MutableSequence[tuple[str, Any]], writes)[i] = (ch, new_value)
-            continue
         if not isinstance(spec, DeltaChannel):
             continue
         if spec.reducer is _messages_delta_reducer:

libs/langgraph/langgraph/pregel/_retry.py

@@ -28,6 +28,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_RUNTIME,
+    CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_SEND,
     CONFIG_KEY_STREAM,
     CONFIG_KEY_TASK_ID,
@@ -44,6 +45,7 @@ from langgraph.errors import (
     NodeTimeoutError,
     ParentCommand,
 )
+from langgraph.pregel._algo import LazyAtomicCounter
 from langgraph.pregel.protocol import StreamProtocol
 from langgraph.runtime import ExecutionInfo, Runtime
 from langgraph.types import Command, PregelExecutableTask, RetryPolicy, TimeoutPolicy
@@ -571,6 +573,23 @@ def _checkpoint_ns_for_parent_command(ns: str) -> str:
     return NS_SEP.join(parts)
 
 
+def _reset_interrupt_counter(config: RunnableConfig) -> RunnableConfig:
+    # A retry reruns the node from its first line, so its interrupt() calls count
+    # from the first again to reuse the answers its task already has, rather than
+    # asking again under an id the next resume can't match.
+    scratchpad = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
+    if scratchpad is None:
+        return config
+    return patch_configurable(
+        config,
+        {
+            CONFIG_KEY_SCRATCHPAD: replace(
+                scratchpad, interrupt_counter=LazyAtomicCounter()
+            )
+        },
+    )
+
+
 def run_with_retry(
     task: PregelExecutableTask,
     retry_policy: Sequence[RetryPolicy] | None,
@@ -681,6 +700,7 @@ def run_with_retry(
             )
             # signal subgraphs to resume (if available)
             config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})
+            config = _reset_interrupt_counter(config)
 
 
 async def arun_with_retry(
@@ -829,6 +849,7 @@ async def arun_with_retry(
             )
             # signal subgraphs to resume (if available)
             config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})
+            config = _reset_interrupt_counter(config)
 
 
 def _return_value(task: PregelExecutableTask) -> Any:

libs/langgraph/langgraph/pregel/main.py

@@ -49,8 +49,6 @@ from langgraph.checkpoint.base import (
     ChannelVersions,
     Checkpoint,
     CheckpointTuple,
-    LogPage,
-    log_range,
 )
 from langgraph.store.base import BaseStore
 from pydantic import BaseModel, TypeAdapter
@@ -112,7 +110,6 @@ from langgraph.callbacks import (
 )
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.delta import DeltaChannel
-from langgraph.channels.log import Log
 from langgraph.channels.topic import Topic
 from langgraph.config import get_config
 from langgraph.constants import END
@@ -1248,7 +1245,6 @@ class Pregel(
             saved.checkpoint,
             saver=saver,
             config=saved.config,
-            load_logs=False,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
@@ -1374,7 +1370,6 @@ class Pregel(
             saved.checkpoint,
             saver=saver,
             config=saved.config,
-            load_logs=False,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
@@ -1487,97 +1482,6 @@ class Pregel(
             live=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
         )
 
-    def read_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        """Read a log channel as of a checkpoint (prototype).
-
-        The same operation as `BaseCheckpointSaver.read_log`, with entries
-        decoded by the channel (messages for a `MessagesLog`).
-        """
-        checkpointer = self._state_checkpointer(config)
-        if subgraph := self._subgraph_for_namespace(config, checkpointer):
-            pregel, subgraph_config = subgraph
-            return pregel.read_log(  # type: ignore[attr-defined]
-                subgraph_config, channel, after=after, before=before, limit=limit
-            )
-        config = merge_configs(self.config, config) if self.config else config
-        config = self._own_checkpoint_config(config)
-        page = checkpointer.read_log(
-            config, channel, after=after, before=before, limit=limit
-        )
-        if page.length == 0 and (tup := checkpointer.get_tuple(config)) is not None:
-            if (
-                pre := self._pre_log_page(channel, tup, after, before, limit)
-            ) is not None:
-                return pre
-        return self._decode_log_page(channel, page)
-
-    async def aread_log(
-        self,
-        config: RunnableConfig,
-        channel: str,
-        *,
-        after: int | None = None,
-        before: int | None = None,
-        limit: int | None = None,
-    ) -> LogPage:
-        """Async version of `read_log`."""
-        checkpointer = self._state_checkpointer(config)
-        if subgraph := self._subgraph_for_namespace(config, checkpointer):
-            pregel, subgraph_config = subgraph
-            return await pregel.aread_log(  # type: ignore[attr-defined]
-                subgraph_config, channel, after=after, before=before, limit=limit
-            )
-        config = merge_configs(self.config, config) if self.config else config
-        config = self._own_checkpoint_config(config)
-        page = await checkpointer.aread_log(
-            config, channel, after=after, before=before, limit=limit
-        )
-        if (
-            page.length == 0
-            and (tup := await checkpointer.aget_tuple(config)) is not None
-        ):
-            if (
-                pre := self._pre_log_page(channel, tup, after, before, limit)
-            ) is not None:
-                return pre
-        return self._decode_log_page(channel, page)
-
-    def _pre_log_page(
-        self,
-        channel: str,
-        tup: CheckpointTuple,
-        after: int | None,
-        before: int | None,
-        limit: int | None,
-    ) -> LogPage | None:
-        """A page of the list a checkpoint stored before the field became a
-        log channel, or `None` if it stored none."""
-        spec = self.channels.get(channel)
-        stored = tup.checkpoint["channel_values"].get(channel)
-        if not isinstance(spec, Log) or not isinstance(stored, list):
-            return None
-        entries = spec.from_checkpoint(stored).entries
-        lo, hi = log_range(len(entries), after, before, limit)
-        return LogPage([(i, entries[i]) for i in range(lo, hi)], len(entries), [])
-
-    def _decode_log_page(self, channel: str, page: LogPage) -> LogPage:
-        spec = self.channels.get(channel)
-        if not isinstance(spec, Log):
-            raise ValueError(f"Channel {channel!r} is not a log channel")
-        return LogPage(
-            [(pos, spec.decode(e)) for pos, e in page.entries],
-            page.length,
-            [spec.decode(e) for e in page.pending],
-        )
-
     async def aget_state(
         self, config: RunnableConfig, *, subgraphs: bool = False
     ) -> StateSnapshot:

libs/langgraph/langgraph/pregel/remote.py

@@ -50,6 +50,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_STREAM,
+    CONFIG_KEY_STRICT_RESUME,
     CONFIG_KEY_TASK_ID,
     INTERRUPT,
     NS_SEP,
@@ -81,6 +82,7 @@ _CONF_DROPLIST = frozenset(
         CONFIG_KEY_CHECKPOINT_ID,
         CONFIG_KEY_CHECKPOINT_NS,
         CONFIG_KEY_TASK_ID,
+        CONFIG_KEY_STRICT_RESUME,
     ),
 )
 

libs/langgraph/langgraph/types.py

@@ -33,6 +33,7 @@ from xxhash import xxh3_128_hexdigest
 
 from langgraph._internal._cache import default_cache_key
 from langgraph._internal._constants import INTERRUPT as _INTERRUPT_KEY
+from langgraph._internal._constants import NS_SEP
 from langgraph._internal._fields import get_cached_annotated_keys, get_update_as_tuples
 from langgraph._internal._retry import default_retry_on
 from langgraph._internal._typing import MISSING, DeprecatedKwargs
@@ -579,6 +580,15 @@ def omit_payload(_value: Any) -> dict[str, Any]:
 _DEFAULT_INTERRUPT_ID = "placeholder-id"
 
 
+def _interrupt_id(ns: str, answered: int) -> str:
+    # Keyed by how many answers the task already has, the slot this interrupt's
+    # answer fills; call order isn't stable across parallel calls in one task.
+    # With none it's the bare namespace hash older versions gave every
+    # interrupt, so saved ids still match. The suffix can't form another task's
+    # namespace, whose segments all contain ":".
+    return xxh3_128_hexdigest((f"{ns}{NS_SEP}{answered}" if answered else ns).encode())
+
+
 @final
 @dataclass(init=False, slots=True)
 class Interrupt(Generic[ResponseT]):
@@ -641,7 +651,7 @@ class Interrupt(Generic[ResponseT]):
     ) -> Interrupt[ResponseT]:
         return cls(
             value=value,
-            id=xxh3_128_hexdigest(ns.encode()),
+            id=_interrupt_id(ns, 0),
             response_schema=response_schema,
         )
 
@@ -1037,9 +1047,11 @@ def interrupt(
     # no resume value found
     raise GraphInterrupt(
         (
-            Interrupt.from_ns(
+            Interrupt(
                 value=value,
-                ns=conf[CONFIG_KEY_CHECKPOINT_NS],
+                id=_interrupt_id(
+                    conf[CONFIG_KEY_CHECKPOINT_NS], len(scratchpad.resume)
+                ),
                 response_schema=adapter.json_schema() if adapter else response_schema,
             ),
         )

libs/langgraph/langgraph/warnings.py

@@ -7,7 +7,6 @@ __all__ = (
     "LangGraphDeprecatedSinceV05",
     "LangGraphDeprecatedSinceV10",
     "LangGraphDeprecatedSinceV11",
-    "LangGraphDeprecatedSinceV13",
 )
 
 
@@ -68,10 +67,3 @@ class LangGraphDeprecatedSinceV11(LangGraphDeprecationWarning):
 
     def __init__(self, message: str, *args: object) -> None:
         super().__init__(message, *args, since=(1, 1), expected_removal=(3, 0))
-
-
-class LangGraphDeprecatedSinceV13(LangGraphDeprecationWarning):
-    """A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v1.3.0"""
-
-    def __init__(self, message: str, *args: object) -> None:
-        super().__init__(message, *args, since=(1, 3), expected_removal=(2, 0))

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_log_channel.py

@@ -1,718 +0,0 @@
-"""Transcript log prototype: append-only `Log` / `MessagesLog` channels.
-
-Every scenario runs against `InMemorySaver` (native log storage: segments,
-step rows, range reads) and `SqliteSaver` (no native support: the default
-`read_log` walks the parent chain), and both must read the same log.
-"""
-
-import operator
-import sqlite3
-import warnings
-from collections.abc import Iterator
-from typing import Annotated, Any
-
-import pytest
-from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
-from langchain_core.tools import tool
-from langgraph.checkpoint.base import BaseCheckpointSaver, LogPage
-from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.sqlite import SqliteSaver
-from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-from langgraph.prebuilt import ToolNode, tools_condition
-from typing_extensions import TypedDict
-
-from langgraph.channels import Log, LogRef, MessagesLog
-from langgraph.errors import InvalidUpdateError
-from langgraph.graph import END, START, StateGraph
-from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages
-from langgraph.types import Command, Overwrite, Send, interrupt
-from langgraph.warnings import LangGraphDeprecatedSinceV13
-
-pytestmark = pytest.mark.anyio
-
-
-@pytest.fixture(params=["memory", "sqlite"])
-def saver(request: pytest.FixtureRequest) -> Iterator[BaseCheckpointSaver]:
-    if request.param == "memory":
-        yield InMemorySaver()
-    else:
-        conn = sqlite3.connect(":memory:", check_same_thread=False)
-        yield SqliteSaver(conn)
-        conn.close()
-
-
-def contents(page: LogPage) -> list[Any]:
-    return [e.content if hasattr(e, "content") else e for _, e in page.entries]
-
-
-class ChatState(TypedDict):
-    messages: Annotated[list, MessagesLog()]
-
-
-def chat_graph(saver: BaseCheckpointSaver, **compile_kw: Any):
-    def model(state: ChatState) -> dict:
-        return {"messages": [AIMessage(f"reply to {state['messages'][-1].content}")]}
-
-    builder = StateGraph(ChatState)
-    builder.add_node("model", model)
-    builder.add_edge(START, "model")
-    builder.add_edge("model", END)
-    return builder.compile(checkpointer=saver, **compile_kw)
-
-
-def test_appends_reads_and_state_marker(saver: BaseCheckpointSaver) -> None:
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    out = graph.invoke({"messages": [HumanMessage("b")]}, config)
-
-    # nodes and invoke() see the list
-    assert [m.content for m in out["messages"]] == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    # state reads return the marker, not the transcript
-    assert graph.get_state(config).values == {"messages": LogRef(4)}
-    # one read operation, same arguments everywhere
-    page = graph.read_log(config, "messages")
-    assert page.length == 4
-    assert contents(page) == ["a", "reply to a", "b", "reply to b"]
-    assert [p for p, _ in page.entries] == [0, 1, 2, 3]
-    assert contents(graph.read_log(config, "messages", limit=2)) == ["b", "reply to b"]
-    assert contents(graph.read_log(config, "messages", before=2, limit=1)) == [
-        "reply to a"
-    ]
-    assert contents(graph.read_log(config, "messages", after=1, limit=1)) == ["b"]
-    # the saver-level read returns JSON
-    raw = saver.read_log(config, "messages")
-    assert raw.entries[0][1]["type"] == "human"
-    # every checkpoint points at its own leaf
-    lengths = [
-        (s.metadata["source"], graph.read_log(s.config, "messages").length)
-        for s in graph.get_state_history(config)
-    ]
-    assert lengths == [
-        ("loop", 4),
-        ("loop", 3),
-        ("input", 2),
-        ("loop", 2),
-        ("loop", 1),
-        ("input", 0),
-    ]
-
-
-async def test_async_appends_and_reads() -> None:
-    async with AsyncSqliteSaver.from_conn_string(":memory:") as fallback:
-        for saver in (InMemorySaver(), fallback):
-            graph = chat_graph(saver)
-            config = {"configurable": {"thread_id": "1"}}
-            await graph.ainvoke({"messages": [HumanMessage("a")]}, config)
-            await graph.ainvoke({"messages": [HumanMessage("b")]}, config)
-            state = await graph.aget_state(config)
-            assert state.values == {"messages": LogRef(4)}
-            page = await graph.aread_log(config, "messages")
-            assert contents(page) == ["a", "reply to a", "b", "reply to b"]
-
-
-def test_log_entries_are_not_stored_in_checkpoints() -> None:
-    saver = InMemorySaver()
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    for i in range(20):
-        graph.invoke({"messages": [HumanMessage(f"m{i}")]}, config)
-    assert not [k for k in saver.blobs if k[2] == "messages"]
-    log = saver.log.logs[("1", "", "messages")]
-    stored = sum(len(rows) for rows in log.entries.values())
-    # each entry stored once: linear in the transcript
-    assert stored == 40 == graph.read_log(config, "messages").length
-    assert len(log.segments) == 1
-
-
-def test_parallel_tasks_keep_one_order(saver: BaseCheckpointSaver) -> None:
-    class State(TypedDict):
-        events: Annotated[list, Log()]
-        seen: Annotated[list, operator.add]
-
-    def fan_out(state: State) -> list[Send]:
-        return [Send("worker", {"i": i}) for i in range(5)]
-
-    def worker(arg: dict) -> dict:
-        return {"events": [{"worker": arg["i"]}, {"worker": arg["i"], "n": 2}]}
-
-    def after(state: State) -> dict:
-        return {"seen": [list(state["events"])]}
-
-    builder = StateGraph(State)
-    builder.add_node("worker", worker)
-    builder.add_node("after", after)
-    builder.add_conditional_edges(START, fan_out, ["worker"])
-    builder.add_edge("worker", "after")
-    builder.add_edge("after", END)
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "1"}}
-    out = graph.invoke({"events": [], "seen": []}, config)
-    in_run = out["seen"][-1]
-    assert len(in_run) == 10
-    # read back after the run: the same order the run's nodes saw
-    assert contents(graph.read_log(config, "events")) == in_run
-
-
-def test_interrupt_and_resume_reuse_finished_writes(
-    saver: BaseCheckpointSaver,
-) -> None:
-    calls: list[str] = []
-
-    class State(TypedDict):
-        events: Annotated[list, Log()]
-
-    def fast(state: State) -> dict:
-        calls.append("fast")
-        return {"events": [{"from": "fast"}]}
-
-    def asks(state: State) -> dict:
-        answer = interrupt("question")
-        calls.append("asks")
-        return {"events": [{"from": "asks", "answer": answer}]}
-
-    builder = StateGraph(State)
-    builder.add_node("fast", fast)
-    builder.add_node("asks", asks)
-    builder.add_edge(START, "fast")
-    builder.add_edge(START, "asks")
-    builder.add_edge(["fast", "asks"], END)
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"events": [{"from": "input"}]}, config)
-    # fast finished; its entry is pending at the head, not committed
-    page = saver.read_log(config, "events")
-    assert page.length == 1
-    assert page.pending == [{"from": "fast"}]
-
-    out = graph.invoke(Command(resume="yes"), config)
-    assert calls == ["fast", "asks"]  # fast was not re-run
-    # the reused write and the new one commit as one step, in path order,
-    # and the log reads back exactly what the run's state held
-    assert out["events"] == [
-        {"from": "input"},
-        {"from": "asks", "answer": "yes"},
-        {"from": "fast"},
-    ]
-    assert contents(graph.read_log(config, "events")) == out["events"]
-    assert graph.read_log(config, "events").pending == []
-
-
-def test_time_travel_forks_without_touching_the_original(
-    saver: BaseCheckpointSaver,
-) -> None:
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    graph.invoke({"messages": [HumanMessage("b")]}, config)
-    original = graph.get_state(config).config
-    # the checkpoint before the second reply ("b" added, model not run)
-    before_reply = next(
-        s
-        for s in graph.get_state_history(config)
-        if s.next == ("model",) and graph.read_log(s.config, "messages").length == 3
-    )
-    graph.invoke(None, before_reply.config)  # replay: a new branch
-    head = graph.get_state(config).config
-    assert (
-        head["configurable"]["checkpoint_id"]
-        != original["configurable"]["checkpoint_id"]
-    )
-    assert contents(graph.read_log(head, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    assert contents(graph.read_log(original, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    a = graph.read_log(head, "messages").entries
-    b = graph.read_log(original, "messages").entries
-    # same prefix, different last message (a new run of the model node)
-    assert [m.id for _, m in a[:3]] == [m.id for _, m in b[:3]]
-    assert a[3][1].id != b[3][1].id
-    if isinstance(saver, InMemorySaver):
-        assert len(saver.log.logs[("1", "", "messages")].segments) == 2
-
-
-def test_update_state_fork_starts_a_branch(saver: BaseCheckpointSaver) -> None:
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    graph.invoke({"messages": [HumanMessage("b")]}, config)
-    original = graph.get_state(config).config
-    first_turn = next(
-        s
-        for s in graph.get_state_history(config)
-        if graph.read_log(s.config, "messages").length == 2
-        and s.metadata["source"] == "loop"
-    )
-    # edit: fork from the end of the first turn with a different question
-    forked = graph.update_state(
-        first_turn.config, {"messages": [HumanMessage("c")]}, as_node="model"
-    )
-    assert contents(graph.read_log(forked, "messages")) == ["a", "reply to a", "c"]
-    graph.invoke(Command(goto="model"), forked)
-    head = graph.get_state(config).config
-    assert contents(graph.read_log(head, "messages")) == [
-        "a",
-        "reply to a",
-        "c",
-        "reply to c",
-    ]
-    assert contents(graph.read_log(original, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-
-
-def test_subgraph_has_its_own_log(saver: BaseCheckpointSaver) -> None:
-    class Sub(TypedDict):
-        notes: Annotated[list, Log()]
-
-    def note(state: Sub) -> dict:
-        return {"notes": [{"n": len(state.get("notes", []))}]}
-
-    sub_builder = StateGraph(Sub)
-    sub_builder.add_node("note", note)
-    sub_builder.add_edge(START, "note")
-    sub = sub_builder.compile(checkpointer=True)
-
-    class Parent(TypedDict):
-        messages: Annotated[list, MessagesLog()]
-
-    def call_sub(state: Parent) -> dict:
-        sub.invoke({"notes": [{"from": "parent"}]})
-        return {"messages": [AIMessage("done")]}
-
-    builder = StateGraph(Parent)
-    builder.add_node("call_sub", call_sub)
-    builder.add_edge(START, "call_sub")
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("go")]}, config)
-    graph.invoke({"messages": [HumanMessage("again")]}, config)
-    assert contents(graph.read_log(config, "messages")) == [
-        "go",
-        "done",
-        "again",
-        "done",
-    ]
-    sub_config = {"configurable": {"thread_id": "1", "checkpoint_ns": "call_sub"}}
-    assert contents(saver.read_log(sub_config, "notes")) == [
-        {"from": "parent"},
-        {"n": 1},
-        {"from": "parent"},
-        {"n": 3},
-    ]
-
-
-def test_exit_durability_commits_inline(saver: BaseCheckpointSaver) -> None:
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config, durability="exit")
-    graph.invoke({"messages": [HumanMessage("b")]}, config, durability="exit")
-    assert contents(graph.read_log(config, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-
-
-def test_prune_keeps_the_heads_transcript() -> None:
-    saver = InMemorySaver()
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    graph.invoke({"messages": [HumanMessage("b")]}, config)
-    original = graph.get_state(config).config
-    before_reply = next(
-        s
-        for s in graph.get_state_history(config)
-        if s.next == ("model",) and graph.read_log(s.config, "messages").length == 3
-    )
-    graph.invoke(None, before_reply.config)
-    log = saver.log.logs[("1", "", "messages")]
-    assert len(log.segments) == 2
-
-    saver.prune(["1"], strategy="keep_latest")
-    assert len(saver.storage["1"][""]) == 1
-    # the head's full transcript survives without its ancestors' checkpoints
-    assert contents(graph.read_log(config, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    # the abandoned branch is collected
-    log = saver.log.logs[("1", "", "messages")]
-    assert sum(len(r) for r in log.entries.values()) == 4
-    assert len(log.steps) == 4
-    assert graph.read_log(original, "messages").length == 0  # its checkpoint is gone
-
-    # the thread keeps working after prune
-    graph.invoke({"messages": [HumanMessage("c")]}, config)
-    assert graph.read_log(config, "messages").length == 6
-
-    saver.prune(["1"], strategy="delete")
-    assert ("1", "", "messages") not in saver.log.logs
-
-
-def test_rollback_collects_the_runs_entries() -> None:
-    saver = InMemorySaver()
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke(
-        {"messages": [HumanMessage("a")]}, {**config, "metadata": {"run_id": "r1"}}
-    )
-    graph.invoke(
-        {"messages": [HumanMessage("b")]}, {**config, "metadata": {"run_id": "r2"}}
-    )
-    saver.delete_for_runs(["r2"])
-    assert contents(graph.read_log(config, "messages")) == ["a", "reply to a"]
-    log = saver.log.logs[("1", "", "messages")]
-    assert sum(len(r) for r in log.entries.values()) == 2
-    # the tip moved back: the next run continues the same segment
-    graph.invoke({"messages": [HumanMessage("c")]}, config)
-    assert contents(graph.read_log(config, "messages")) == [
-        "a",
-        "reply to a",
-        "c",
-        "reply to c",
-    ]
-    assert len(saver.log.logs[("1", "", "messages")].segments) == 1
-
-
-def test_copy_and_delete_thread() -> None:
-    saver = InMemorySaver()
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    saver.copy_thread("1", "2")
-    copy_config = {"configurable": {"thread_id": "2"}}
-    graph.invoke({"messages": [HumanMessage("b")]}, copy_config)
-    assert contents(graph.read_log(config, "messages")) == ["a", "reply to a"]
-    assert contents(graph.read_log(copy_config, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    saver.delete_thread("1")
-    assert ("1", "", "messages") not in saver.log.logs
-    assert graph.read_log(copy_config, "messages").length == 4
-
-
-def test_log_takes_only_json(saver: BaseCheckpointSaver) -> None:
-    class State(TypedDict):
-        events: Annotated[list, Log()]
-
-    def bad(state: State) -> dict:
-        return {"events": [{"when": object()}]}
-
-    builder = StateGraph(State)
-    builder.add_node("bad", bad)
-    builder.add_edge(START, "bad")
-    graph = builder.compile(checkpointer=saver)
-    with pytest.raises(InvalidUpdateError, match="not JSON"):
-        graph.invoke({"events": [{"ok": 1}]}, {"configurable": {"thread_id": "1"}})
-    with pytest.raises(InvalidUpdateError, match="not JSON"):
-        graph.invoke({"events": [float("nan")]}, {"configurable": {"thread_id": "2"}})
-
-
-@pytest.mark.parametrize(
-    "write",
-    [
-        pytest.param(lambda s: [RemoveMessage(id=s["messages"][0].id)], id="remove"),
-        pytest.param(lambda s: Overwrite([AIMessage("x")]), id="overwrite"),
-        pytest.param(lambda s: [s["messages"][0]], id="replace-by-id"),
-        pytest.param(lambda s: s["messages"] + [AIMessage("x")], id="whole-list"),
-    ],
-)
-def test_messages_log_rejects_edits(saver: BaseCheckpointSaver, write: Any) -> None:
-    def edits(state: ChatState) -> dict:
-        return {"messages": write(state)}
-
-    builder = StateGraph(ChatState)
-    builder.add_node("edits", edits)
-    builder.add_edge(START, "edits")
-    graph = builder.compile(checkpointer=saver)
-    with pytest.raises(InvalidUpdateError):
-        graph.invoke(
-            {"messages": [HumanMessage("a")]}, {"configurable": {"thread_id": "1"}}
-        )
-
-
-def test_fallback_and_native_read_the_same_log() -> None:
-    def run(saver: BaseCheckpointSaver) -> list[list[Any]]:
-        graph = chat_graph(saver)
-        config = {"configurable": {"thread_id": "1"}}
-        for q in "abc":
-            graph.invoke({"messages": [HumanMessage(q)]}, config)
-        return [
-            contents(graph.read_log(s.config, "messages"))
-            for s in graph.get_state_history(config)
-        ]
-
-    conn = sqlite3.connect(":memory:", check_same_thread=False)
-    try:
-        assert run(InMemorySaver()) == run(SqliteSaver(conn))
-    finally:
-        conn.close()
-
-
-def test_resume_at_a_superseded_checkpoint_branches(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """A resume at a checkpoint the thread moved past re-runs finished tasks on
-    that same checkpoint, without a fork checkpoint. The log puts the re-run
-    in a new segment and leaves the original branch alone."""
-    calls: list[str] = []
-
-    class State(TypedDict):
-        events: Annotated[list, Log()]
-
-    def fast(state: State) -> dict:
-        calls.append("fast")
-        return {"events": [{"from": "fast", "run": len(calls)}]}
-
-    def asks(state: State) -> dict:
-        answer = interrupt("question")
-        return {"events": [{"from": "asks", "answer": answer}]}
-
-    builder = StateGraph(State)
-    builder.add_node("fast", fast)
-    builder.add_node("asks", asks)
-    builder.add_edge(START, "fast")
-    builder.add_edge(START, "asks")
-    builder.add_edge(["fast", "asks"], END)
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"events": [{"from": "input"}]}, config)
-    interrupted = graph.get_state(config).config
-    first = graph.invoke(Command(resume="yes"), config)["events"]
-    original_head = graph.get_state(config).config
-
-    second = graph.invoke(Command(resume="no"), interrupted)["events"]
-    assert calls == ["fast", "fast"]  # the finished task re-ran
-    new_head = graph.get_state(config).config
-    assert contents(graph.read_log(new_head, "events")) == second
-    assert contents(graph.read_log(original_head, "events")) == first
-    assert first != second
-    if isinstance(saver, InMemorySaver):
-        assert len(saver.log.logs[("1", "", "events")].segments) == 2
-
-
-def test_tool_calling_agent_runs_on_a_messages_log(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Prebuilt `ToolNode` and `tools_condition` work unchanged: they read the
-    list and append messages."""
-
-    @tool
-    def add(a: int, b: int) -> int:
-        """Add two numbers."""
-        return a + b
-
-    def model(state: ChatState) -> dict:
-        last = state["messages"][-1]
-        if isinstance(last, HumanMessage):
-            call = {
-                "name": "add",
-                "args": {"a": 2, "b": 3},
-                "id": f"call-{len(state['messages'])}",
-            }
-            return {"messages": [AIMessage("", tool_calls=[call])]}
-        return {"messages": [AIMessage(f"the answer is {last.content}")]}
-
-    builder = StateGraph(ChatState)
-    builder.add_node("model", model)
-    builder.add_node("tools", ToolNode([add]))
-    builder.add_edge(START, "model")
-    builder.add_conditional_edges("model", tools_condition)
-    builder.add_edge("tools", "model")
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "1"}}
-    out = graph.invoke({"messages": [HumanMessage("2+3?")]}, config)
-    assert [type(m).__name__ for m in out["messages"]] == [
-        "HumanMessage",
-        "AIMessage",
-        "ToolMessage",
-        "AIMessage",
-    ]
-    page = graph.read_log(config, "messages")
-    assert [m.id for _, m in page.entries] == [m.id for m in out["messages"]]
-    assert page.entries[-1][1].content == "the answer is 5"
-
-
-def test_input_after_update_state_reads_the_inline_step(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """`update_state` commits its entries inline; the input checkpoint after
-    it keeps the version with a stripped marker, and reads must still find
-    the entries on the checkpoint that committed them."""
-    graph = chat_graph(saver)
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"messages": [HumanMessage("a")]}, config)
-    graph.update_state(config, {"messages": [HumanMessage("note")]})
-    graph.invoke({"messages": [HumanMessage("b")]}, config)
-    assert contents(graph.read_log(config, "messages")) == [
-        "a",
-        "reply to a",
-        "note",
-        "b",
-        "reply to b",
-    ]
-
-
-# -- existing `add_messages` threads ------------------------------------------
-
-
-class OldChatState(TypedDict):
-    messages: Annotated[list, add_messages]
-
-
-def _reply(state: dict) -> dict:
-    return {"messages": [AIMessage(f"reply to {state['messages'][-1].content}")]}
-
-
-def _chat(schema: type, saver: BaseCheckpointSaver, **nodes: Any):
-    builder = StateGraph(schema)
-    builder.add_node("model", nodes.get("model", _reply))
-    builder.add_edge(START, "model")
-    if "review" in nodes:
-        builder.add_node("review", nodes["review"])
-        builder.add_edge("model", "review")
-        builder.add_edge("review", END)
-    else:
-        builder.add_edge("model", END)
-    return builder.compile(checkpointer=saver)
-
-
-def test_add_messages_thread_switches_to_messages_log(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """A thread written with `add_messages` keeps working on `MessagesLog`:
-    the stored list is the log's first entries."""
-    config = {"configurable": {"thread_id": "1"}}
-    old = _chat(OldChatState, saver)
-    old.invoke({"messages": [HumanMessage("a", id="h1")]}, config)
-    old.invoke({"messages": [HumanMessage("b", id="h2")]}, config)
-    before_switch = old.get_state(config).config
-
-    new = _chat(ChatState, saver)
-    out = new.invoke({"messages": [HumanMessage("c", id="h3")]}, config)
-    assert [m.content for m in out["messages"]] == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-        "c",
-        "reply to c",
-    ]
-    assert new.get_state(config).values == {"messages": LogRef(6)}
-    page = new.read_log(config, "messages")
-    assert [p for p, _ in page.entries] == list(range(6))
-    assert contents(page) == [m.content for m in out["messages"]]
-    assert [m.id for _, m in page.entries][:1] == ["h1"]
-    # checkpoints from before the switch still read as the list they stored
-    assert contents(new.read_log(before_switch, "messages")) == [
-        "a",
-        "reply to a",
-        "b",
-        "reply to b",
-    ]
-    assert len(new.get_state(before_switch).values["messages"]) == 4
-    # the next turn appends after the seed
-    new.invoke({"messages": [HumanMessage("d")]}, config)
-    assert new.read_log(config, "messages").length == 8
-    if isinstance(saver, InMemorySaver):
-        log = saver.log.logs[("1", "", "messages")]
-        assert sum(len(rows) for rows in log.entries.values()) == 8
-
-
-def test_switch_with_an_interrupted_add_messages_run(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """The first checkpoint after the switch can come from the loop (a
-    resume), not graph input; the seed is committed with it."""
-
-    def review(state: dict) -> dict:
-        answer = interrupt("approve?")
-        return {"messages": [AIMessage(f"review: {answer}")]}
-
-    config = {"configurable": {"thread_id": "1"}}
-    old = _chat(OldChatState, saver, review=review)
-    old.invoke({"messages": [HumanMessage("a")]}, config)
-    assert old.get_state(config).next == ("review",)
-
-    new = _chat(ChatState, saver, review=review)
-    out = new.invoke(Command(resume="ok"), config)
-    assert [m.content for m in out["messages"]] == ["a", "reply to a", "review: ok"]
-    assert contents(new.read_log(config, "messages")) == [
-        "a",
-        "reply to a",
-        "review: ok",
-    ]
-
-
-def test_switch_then_update_state(saver: BaseCheckpointSaver) -> None:
-    config = {"configurable": {"thread_id": "1"}}
-    old = _chat(OldChatState, saver)
-    old.invoke({"messages": [HumanMessage("a")]}, config)
-    new = _chat(ChatState, saver)
-    new.update_state(config, {"messages": [HumanMessage("note")]})
-    assert contents(new.read_log(config, "messages")) == ["a", "reply to a", "note"]
-    out = new.invoke({"messages": [HumanMessage("b")]}, config)
-    assert [m.content for m in out["messages"]][-2:] == ["b", "reply to b"]
-    assert new.read_log(config, "messages").length == 5
-
-
-# -- deprecation of message edit ops on `add_messages` -------------------------
-
-
-@pytest.mark.parametrize(
-    "edit",
-    [
-        pytest.param(lambda s: [RemoveMessage(id=s["messages"][0].id)], id="remove"),
-        pytest.param(
-            lambda s: [RemoveMessage(id=REMOVE_ALL_MESSAGES), AIMessage("x")],
-            id="remove-all",
-        ),
-        pytest.param(
-            lambda s: [HumanMessage("edited", id=s["messages"][0].id)],
-            id="replace-by-id",
-        ),
-        pytest.param(lambda s: Overwrite([AIMessage("x")]), id="overwrite"),
-    ],
-)
-def test_add_messages_edit_ops_are_deprecated(edit: Any) -> None:
-    def node(state: dict) -> dict:
-        return {"messages": edit(state)}
-
-    graph = _chat(OldChatState, InMemorySaver(), model=node)
-    with pytest.warns(LangGraphDeprecatedSinceV13, match="MessagesLog"):
-        graph.invoke(
-            {"messages": [HumanMessage("a")]}, {"configurable": {"thread_id": "1"}}
-        )
-
-
-def test_add_messages_appends_are_not_deprecated() -> None:
-    graph = _chat(OldChatState, InMemorySaver())
-    with warnings.catch_warnings():
-        warnings.simplefilter("error", LangGraphDeprecatedSinceV13)
-        graph.invoke(
-            {"messages": [HumanMessage("a")]}, {"configurable": {"thread_id": "1"}}
-        )

libs/langgraph/tests/test_remote_graph.py

@@ -16,7 +16,7 @@ from langgraph_sdk.schema import StreamPart
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
-from langgraph._internal._constants import CONFIG_KEY_STREAM
+from langgraph._internal._constants import CONFIG_KEY_STREAM, CONFIG_KEY_STRICT_RESUME
 from langgraph.errors import GraphInterrupt
 from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
 from langgraph.pregel import Pregel
@@ -1360,6 +1360,7 @@ def test_sanitize_config():
             "checkpoint_map": {"key": "value"},  # Should be dropped
             "checkpoint_id": "123",  # Should be dropped
             "checkpoint_ns": "ns",  # Should be dropped
+            CONFIG_KEY_STRICT_RESUME: True,  # Should be dropped
         }
     }
     sanitized = remote._sanitize_config(config_with_drops)

libs/langgraph/tests/test_strict_resume.py

@@ -0,0 +1,423 @@
+import operator
+import sys
+import uuid
+from typing import Annotated
+
+import pytest
+from langchain_core.runnables import RunnableConfig
+from langchain_core.runnables.config import get_executor_for_config
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from typing_extensions import TypedDict
+from xxhash import xxh3_128_hexdigest
+
+from langgraph._internal._constants import CONFIG_KEY_STRICT_RESUME, INTERRUPT
+from langgraph.errors import UnknownInterruptError
+from langgraph.graph import START, StateGraph
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.types import Command, Durability, Interrupt, RetryPolicy, interrupt
+
+pytestmark = pytest.mark.anyio
+NEEDS_CONTEXTVARS = pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+
+UNKNOWN_ID = "0" * 32
+
+
+class State(TypedDict, total=False):
+    log: Annotated[list[str], operator.add]
+
+
+def _config(*, strict: bool = True) -> RunnableConfig:
+    return {
+        "configurable": {
+            "thread_id": str(uuid.uuid4()),
+            CONFIG_KEY_STRICT_RESUME: strict,
+        }
+    }
+
+
+def _parallel(checkpointer: BaseCheckpointSaver) -> CompiledStateGraph:
+    def a(state: State):
+        return {"log": [f"a:{interrupt('a?')}"]}
+
+    def b(state: State):
+        return {"log": [f"b:{interrupt('b?')}"]}
+
+    return (
+        StateGraph(State)
+        .add_node("a", a)
+        .add_node("b", b)
+        .add_edge(START, "a")
+        .add_edge(START, "b")
+        .compile(checkpointer=checkpointer)
+    )
+
+
+def _ask_twice(checkpointer: BaseCheckpointSaver) -> CompiledStateGraph:
+    def ask_twice(state: State):
+        first = interrupt("first?")
+        second = interrupt("second?")
+        return {"log": [f"{first},{second}"]}
+
+    return (
+        StateGraph(State)
+        .add_node("ask", ask_twice)
+        .add_edge(START, "ask")
+        .compile(checkpointer=checkpointer)
+    )
+
+
+def _ask_then_fail_once(checkpointer: BaseCheckpointSaver) -> CompiledStateGraph:
+    failures = [ConnectionError("flaky api")]
+
+    def ask(state: State):
+        answer = interrupt("delete a.txt?")
+        if failures:
+            raise failures.pop()
+        return {"log": [answer]}
+
+    return (
+        StateGraph(State)
+        .add_node(
+            "ask", ask, retry_policy=RetryPolicy(initial_interval=0, jitter=False)
+        )
+        .add_edge(START, "ask")
+        .compile(checkpointer=checkpointer)
+    )
+
+
+def _pending_ids(graph: CompiledStateGraph, config: RunnableConfig) -> dict[str, str]:
+    return {i.value: i.id for i in graph.get_state(config).interrupts}
+
+
+def test_unknown_id_is_refused_before_anything_is_written(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config, durability=durability)
+    before = graph.get_state(config)
+    checkpoints = len(list(sync_checkpointer.list(config)))
+    ids = _pending_ids(graph, config)
+
+    with pytest.raises(
+        UnknownInterruptError,
+        match=f"Unknown or already-consumed interrupt: {UNKNOWN_ID}",
+    ) as refused:
+        graph.invoke(Command(resume={UNKNOWN_ID: "x"}), config, durability=durability)
+
+    assert (
+        refused.value.reason,
+        refused.value.interrupt_ids,
+        refused.value.pending_ids,
+    ) == ("not_pending", (UNKNOWN_ID,), tuple(sorted(ids.values())))
+    after = graph.get_state(config)
+    assert len(list(sync_checkpointer.list(config))) == checkpoints
+    assert (after.values, after.next, after.interrupts) == (
+        before.values,
+        before.next,
+        before.interrupts,
+    )
+    result = graph.invoke(
+        Command(resume={ids["a?"]: "yes", ids["b?"]: "no"}),
+        config,
+        durability=durability,
+    )
+    assert sorted(result["log"]) == ["a:yes", "b:no"]
+
+
+def test_an_answered_id_is_refused(sync_checkpointer: BaseCheckpointSaver) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    ids = _pending_ids(graph, config)
+    graph.invoke(Command(resume={ids["a?"]: "yes"}), config)
+
+    with pytest.raises(UnknownInterruptError, match=ids["a?"]) as refused:
+        graph.invoke(Command(resume={ids["a?"]: "again"}), config)
+
+    assert (
+        refused.value.reason,
+        refused.value.interrupt_ids,
+        refused.value.pending_ids,
+    ) == ("not_pending", (ids["a?"],), (ids["b?"],))
+
+    result = graph.invoke(Command(resume={ids["b?"]: "no"}), config)
+    assert sorted(result["log"]) == ["a:yes", "b:no"]
+
+
+def test_an_empty_map_is_refused(sync_checkpointer: BaseCheckpointSaver) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    with pytest.raises(UnknownInterruptError, match="names no interrupt") as refused:
+        graph.invoke(Command(resume={}), config)
+
+    assert (
+        refused.value.reason,
+        refused.value.interrupt_ids,
+        refused.value.pending_ids,
+    ) == ("empty_map", (), tuple(sorted(_pending_ids(graph, config).values())))
+
+
+def test_one_value_is_refused_while_several_are_pending(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    with pytest.raises(
+        UnknownInterruptError, match="multiple pending interrupts"
+    ) as refused:
+        graph.invoke(Command(resume="yes"), config)
+
+    ids = _pending_ids(graph, config)
+    assert (refused.value.reason, refused.value.pending_ids) == (
+        "ambiguous",
+        tuple(sorted(ids.values())),
+    )
+    graph.invoke(Command(resume={ids["a?"]: "yes"}), config)
+    result = graph.invoke(Command(resume="no"), config)
+    assert sorted(result["log"]) == ["a:yes", "b:no"]
+
+
+def test_one_value_resumes_a_static_breakpoint(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = (
+        StateGraph(State)
+        .add_node("a", lambda state: {"log": ["a"]})
+        .add_edge(START, "a")
+        .compile(checkpointer=sync_checkpointer, interrupt_before=["a"])
+    )
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    with pytest.raises(UnknownInterruptError, match=UNKNOWN_ID):
+        graph.invoke(Command(resume={UNKNOWN_ID: "x"}), config)
+
+    assert graph.invoke(Command(resume="go"), config) == {"log": ["a"]}
+
+
+def test_without_strict_mode_nothing_changes(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config(strict=False)
+    graph.invoke({"log": []}, config)
+
+    graph.invoke(Command(resume={UNKNOWN_ID: "x"}), config)
+    assert set(_pending_ids(graph, config)) == {"a?", "b?"}
+    with pytest.raises(RuntimeError, match="multiple pending interrupts"):
+        graph.invoke(Command(resume="yes"), config)
+
+
+def test_a_subgraph_interrupt_is_answered_by_its_id(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    def ask(state: State):
+        return {"log": [f"asked:{interrupt('sub?')}"]}
+
+    subgraph = StateGraph(State).add_node("ask", ask).add_edge(START, "ask").compile()
+    graph = (
+        StateGraph(State)
+        .add_node("sub", subgraph)
+        .add_edge(START, "sub")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    result = graph.invoke(
+        Command(resume={_pending_ids(graph, config)["sub?"]: "yes"}), config
+    )
+    assert result["log"] == ["asked:yes"]
+
+
+def test_a_second_interrupt_in_one_node_gets_its_own_id(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _ask_twice(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    state = graph.get_state(config)
+    (card,) = state.interrupts
+    assert card.id == xxh3_128_hexdigest(f"ask:{state.tasks[0].id}".encode())
+    graph.invoke(Command(resume={card.id: "1"}), config)
+    (second,) = graph.get_state(config).interrupts
+    assert second.id != card.id
+
+    result = graph.invoke(Command(resume={second.id: "2"}), config)
+    assert result["log"] == ["1,2"]
+
+
+def test_a_retried_answer_in_one_node_is_refused(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _ask_twice(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    (card,) = graph.get_state(config).interrupts
+    graph.invoke(Command(resume={card.id: "yes"}), config)
+    (second,) = graph.get_state(config).interrupts
+
+    with pytest.raises(UnknownInterruptError) as refused:
+        graph.invoke(Command(resume={card.id: "yes"}), config)
+
+    assert (
+        refused.value.reason,
+        refused.value.interrupt_ids,
+        refused.value.pending_ids,
+    ) == ("not_pending", (card.id,), (second.id,))
+    result = graph.invoke(Command(resume={second.id: "no"}), config)
+    assert result["log"] == ["yes,no"]
+
+
+def test_parallel_interrupts_in_one_node_each_take_an_answer(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    def ask(name: str) -> str:
+        return interrupt(f"{name}?")
+
+    def ask_in_parallel(state: State, config: RunnableConfig):
+        with get_executor_for_config(config) as pool:
+            b = pool.submit(ask, "b")
+            b.exception()
+            a = pool.submit(ask, "a")
+            return {"log": [f"{a.result()},{b.result()}"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("ask", ask_in_parallel)
+        .add_edge(START, "ask")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+    graph.invoke({"log": []}, config)
+    for answer in ("1", "2"):
+        (card,) = graph.get_state(config).interrupts
+        graph.invoke(Command(resume={card.id: answer}), config)
+
+    assert graph.get_state(config).next == ()
+
+
+def test_a_later_interrupt_saved_with_the_first_id_is_still_answered(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _ask_twice(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    (card,) = graph.get_state(config).interrupts
+    graph.invoke(Command(resume={card.id: "1"}), config)
+    state = graph.get_state(config)
+    sync_checkpointer.put_writes(
+        state.config,
+        [(INTERRUPT, (Interrupt(value="second?", id=card.id),))],
+        state.tasks[0].id,
+    )
+
+    result = graph.invoke(Command(resume={card.id: "2"}), config)
+    assert result["log"] == ["1,2"]
+
+
+def test_a_retried_node_reuses_its_answer(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _ask_then_fail_once(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    (card,) = graph.get_state(config).interrupts
+
+    result = graph.invoke(Command(resume={card.id: "yes"}), config)
+    assert result["log"] == ["yes"]
+
+
+def test_a_card_whose_resumed_task_raised_is_not_refused(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    failures = ["once"]
+
+    def ask_then_fail_once(state: State):
+        answer = interrupt("approve?")
+        if failures:
+            failures.pop()
+            raise ValueError("failed after the answer")
+        return {"log": [f"answered:{answer}"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("ask", ask_then_fail_once)
+        .add_edge(START, "ask")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+    graph.invoke({"log": []}, config)
+    (card,) = graph.get_state(config).interrupts
+    with pytest.raises(ValueError, match="failed after the answer"):
+        graph.invoke(Command(resume={card.id: "first"}), config)
+
+    graph.invoke(Command(resume={card.id: "second"}), config)
+    assert graph.get_state(config).next == ()
+
+
+def test_a_pinned_head_checkpoint_is_checked(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    head = graph.get_state(config).config
+    pinned = {"configurable": {**config["configurable"], **head["configurable"]}}
+
+    with pytest.raises(UnknownInterruptError):
+        graph.invoke(Command(resume={UNKNOWN_ID: "x"}), pinned)
+
+    ids = _pending_ids(graph, config)
+    result = graph.invoke(Command(resume={ids["a?"]: "yes", ids["b?"]: "no"}), pinned)
+    assert sorted(result["log"]) == ["a:yes", "b:no"]
+
+
+def test_a_superseded_checkpoint_is_not_checked(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _parallel(sync_checkpointer)
+    config = _config()
+    graph.invoke({"log": []}, config)
+    paused = graph.get_state(config).config
+    ids = _pending_ids(graph, config)
+    graph.invoke(Command(resume={ids["a?"]: "yes", ids["b?"]: "no"}), config)
+
+    pinned = {"configurable": {**config["configurable"], **paused["configurable"]}}
+    graph.invoke(Command(resume={ids["a?"]: "again", ids["b?"]: "again"}), pinned)
+
+
+@NEEDS_CONTEXTVARS
+async def test_unknown_id_is_refused_async(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _parallel(async_checkpointer)
+    config = _config()
+    await graph.ainvoke({"log": []}, config)
+
+    with pytest.raises(UnknownInterruptError, match=UNKNOWN_ID):
+        await graph.ainvoke(Command(resume={UNKNOWN_ID: "x"}), config)
+
+    state = await graph.aget_state(config)
+    assert {i.value for i in state.interrupts} == {"a?", "b?"}
+
+
+@NEEDS_CONTEXTVARS
+async def test_a_retried_node_reuses_its_answer_async(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _ask_then_fail_once(async_checkpointer)
+    config = _config()
+    await graph.ainvoke({"log": []}, config)
+    (card,) = (await graph.aget_state(config)).interrupts
+
+    result = await graph.ainvoke(Command(resume={card.id: "yes"}), config)
+    assert result["log"] == ["yes"]

libs/langgraph/uv.lock

@@ -1439,7 +1439,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.19"
+version = "1.2.20"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.19"
+version = "1.2.20"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/__init__.py

@@ -5,7 +5,7 @@ from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 from langgraph_sdk.priority import Priority
 from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
 
-__version__ = "0.4.12"
+__version__ = "0.4.13"
 
 __all__ = [
     "Auth",

libs/sdk-py/langgraph_sdk/_async/runs.py

@@ -305,7 +305,7 @@ class RunsClient:
             StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'})
             StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]})
             StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]})
-            StreamPart(event='end', data=None)
+            StreamPart(event='end', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2', 'status': 'success'})
             ```
 
         """

libs/sdk-py/langgraph_sdk/_sync/runs.py

@@ -302,7 +302,7 @@ class SyncRunsClient:
             StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'})
             StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]})
             StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]})
-            StreamPart(event='end', data=None)
+            StreamPart(event='end', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2', 'status': 'success'})
             ```
         """
         if checkpoint_during is not None:

libs/sdk-py/langgraph_sdk/schema.py

@@ -380,6 +380,8 @@ class Run(TypedDict):
     """The run metadata."""
     multitask_strategy: MultitaskStrategy
     """Strategy to handle concurrent runs on the same thread."""
+    cancel_reason: NotRequired[str | None]
+    """Why the run was cancelled, if it was. Omitted by older servers."""
 
 
 class Cron(TypedDict):
@@ -505,6 +507,7 @@ RunSelectField = Literal[
     "metadata",
     "kwargs",
     "multitask_strategy",
+    "cancel_reason",
 ]
 
 CronSelectField = Literal[

libs/sdk-py/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.19"
+version = "1.2.20"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },