prototype: seed existing add_messages threads; deprecate message edit ops

Existing threads. A field that changes from `add_messages` (or any plain
list) to a log channel keeps its thread: the stored list becomes the log's
first entries. The seed is committed inline with the first checkpoint saved
after the change, whether that comes from graph input, a resume or
`update_state`, and the channel's version is bumped if that step didn't
change it. Checkpoints from before the change still read as the list they
stored (`get_state`, `Pregel.read_log`); `InMemorySaver` treats their
version as an empty log.

Deprecation. `add_messages` warns (`LangGraphDeprecatedSinceV13`, removal
in 2.0; both placeholders) on the four edit ops: `RemoveMessage`,
`REMOVE_ALL_MESSAGES`, replace-by-id, and `Overwrite` on an `add_messages`
channel. Appends don't warn. `MessagesLog` errors now say what to do
instead.

Fix: the default `read_log` took a step's marker from the newest
checkpoint carrying its version, which holds the marker without inline
entries; it now uses the checkpoint that committed the version. This broke
reads after `update_state` followed by graph input on savers without native
log support.

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

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import copy
+import json
 import logging
 from collections.abc import AsyncIterator, Collection, Iterator, Mapping, Sequence
 from typing import (
@@ -26,6 +27,8 @@ from langgraph.checkpoint.serde.types import (
     RESUME,
     SCHEDULED,
     ChannelProtocol,
+    LogAppend,
+    LogStep,
 )
 
 V = TypeVar("V", int, float, str)
@@ -182,6 +185,93 @@ 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]:
@@ -231,6 +321,14 @@ 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,
         *,
@@ -604,6 +702,109 @@ 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,12 +21,15 @@ 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__)
 
@@ -66,6 +69,8 @@ class InMemorySaver(
         ```
     """
 
+    supports_log = True
+
     # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
     storage: defaultdict[
         str,
@@ -93,6 +98,7 @@ 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]
@@ -131,6 +137,11 @@ 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
@@ -149,6 +160,87 @@ 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]:
@@ -255,7 +347,6 @@ 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,
@@ -266,9 +357,9 @@ class InMemorySaver(
                         ),
                     },
                     metadata=self.serde.loads_typed(metadata),
-                    pending_writes=[
-                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
-                    ],
+                    pending_writes=self._pending_writes(
+                        thread_id, checkpoint_ns, checkpoint_id
+                    ),
                     parent_config=(
                         {
                             "configurable": {
@@ -285,7 +376,6 @@ 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={
@@ -302,9 +392,9 @@ class InMemorySaver(
                         ),
                     },
                     metadata=self.serde.loads_typed(metadata),
-                    pending_writes=[
-                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
-                    ],
+                    pending_writes=self._pending_writes(
+                        thread_id, checkpoint_ns, checkpoint_id
+                    ),
                     parent_config=(
                         {
                             "configurable": {
@@ -388,10 +478,6 @@ 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(
@@ -422,9 +508,9 @@ class InMemorySaver(
                             if parent_checkpoint_id
                             else None
                         ),
-                        pending_writes=[
-                            (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
-                        ],
+                        pending_writes=self._pending_writes(
+                            thread_id, checkpoint_ns, checkpoint_id
+                        ),
                     )
 
     def put(
@@ -451,8 +537,22 @@ 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"")
             )
@@ -500,6 +600,19 @@ 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
@@ -528,6 +641,72 @@ 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`.
@@ -619,6 +798,17 @@ 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

@@ -0,0 +1,401 @@
+"""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,6 +34,8 @@ 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,
 )
@@ -322,11 +324,29 @@ 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,
@@ -659,6 +679,14 @@ 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,6 +31,52 @@ 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/config.py

@@ -154,9 +154,12 @@ def _get_pip_cleanup_lines(
     pip_installer: Literal["uv", "pip"],
 ) -> str:
     commands = [
-        "mkdir -p /api/langgraph_api /api/langgraph_runtime /api/langgraph_license && "
-        "touch /api/langgraph_api/__init__.py /api/langgraph_runtime/__init__.py /api/langgraph_license/__init__.py",
-        f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir --no-deps -e /api",
+        f"""# -- Ensure user deps didn't inadvertently overwrite langgraph-api
+RUN mkdir -p /api/langgraph_api /api/langgraph_runtime /api/langgraph_license && \
+touch /api/langgraph_api/__init__.py /api/langgraph_runtime/__init__.py /api/langgraph_license/__init__.py
+RUN PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir --no-deps -e /api
+# -- End of ensuring user deps didn't inadvertently overwrite langgraph-api --
+# -- Removing build deps from the final image ~<:===~~~ --"""
     ]
     if to_uninstall:
         for pack in to_uninstall:
@@ -165,31 +168,30 @@ def _get_pip_cleanup_lines(
                     f"Invalid build tool: {pack}; must be one of {', '.join(_BUILD_TOOLS)}"
                 )
         packs_str = " ".join(sorted(to_uninstall))
-        commands.append(f"pip uninstall -y {packs_str}")
+        commands.append(f"RUN pip uninstall -y {packs_str}")
         # Ensure the directories are removed entirely
         packages_rm = " ".join(
             f"/usr/local/lib/python*/site-packages/{pack}*" for pack in to_uninstall
         )
         if "pip" in to_uninstall:
             packages_rm += ' && find /usr/local/bin -name "pip*" -delete || true'
-        commands.append(f"rm -rf {packages_rm}")
+        commands.append(f"RUN rm -rf {packages_rm}")
         wolfi_packages_rm = " ".join(
             f"/usr/lib/python*/site-packages/{pack}*" for pack in to_uninstall
         )
         if "pip" in to_uninstall:
             wolfi_packages_rm += ' && find /usr/bin -name "pip*" -delete || true'
-        commands.append(f"rm -rf {wolfi_packages_rm}")
+        commands.append(f"RUN rm -rf {wolfi_packages_rm}")
         if pip_installer == "uv":
             commands.append(
-                f"uv pip uninstall --system {packs_str} && rm /usr/bin/uv /usr/bin/uvx"
+                f"RUN uv pip uninstall --system {packs_str} && rm /usr/bin/uv /usr/bin/uvx"
             )
-    elif pip_installer == "uv":
-        commands.append("rm /usr/bin/uv /usr/bin/uvx")
-    # One RUN, since each layer is costly on copy-based snapshotters; subshells scope each `|| true`.
-    return (
-        "# -- Restore langgraph-api over user deps, then remove build deps --\n"
-        "RUN " + " && \\\n    ".join(f"({command})" for command in commands)
-    )
+    else:
+        if pip_installer == "uv":
+            commands.append(
+                "RUN rm /usr/bin/uv /usr/bin/uvx\n# -- End of build deps removal --"
+            )
+    return "\n".join(commands)
 
 
 def _parse_version(version_str: str) -> tuple[int, int]:
@@ -1606,10 +1608,7 @@ ADD {relpath} /deps/{name}
                 pip_installer=pip_installer,
             ),
             "",
-            # The JS install step already set this WORKDIR.
-            f"WORKDIR {local_deps.working_dir}"
-            if local_deps.working_dir and not js_inst_str
-            else "",
+            f"WORKDIR {local_deps.working_dir}" if local_deps.working_dir else "",
         ]
     )
 
@@ -1691,7 +1690,7 @@ def node_config_to_docker(
         "",
         os.linesep.join(env_vars),
         "",
-        f"WORKDIR {build_workdir}" if build_workdir != install_workdir else "",
+        f"WORKDIR {build_workdir}",
         "",
         build_step,
     ]

libs/cli/langgraph_cli/uv_lock.py

@@ -65,7 +65,6 @@ class UvLockPlan:
 @dataclass(slots=True)
 class DockerBuildPlan:
     lines: list[str]
-    workdir: str | None = None
 
     def add_blank(self) -> None:
         self.lines.append("")
@@ -74,11 +73,6 @@ class DockerBuildPlan:
         self.lines.append(line)
 
     def add_instruction(self, opcode: str, value: str | None = None) -> None:
-        # Repeating the current WORKDIR changes nothing but still costs a build step.
-        if opcode == "WORKDIR":
-            if value == self.workdir:
-                return
-            self.workdir = value
         if value:
             self.lines.append(f"{opcode} {value}")
         else:

libs/cli/tests/unit_tests/test_config.py

@@ -1231,6 +1231,7 @@ ENV LANGGRAPH_AUTH='{"path": "./graphs/auth.mts:auth"}'
 ENV LANGGRAPH_UI='{"agent": "./graphs/agent.ui.jsx"}'
 ENV LANGGRAPH_UI_CONFIG='{"shared": ["nuqs"]}'
 ENV LANGSERVE_GRAPHS='{"agent": "./graphs/agent.js:graph"}'
+WORKDIR /deps/unit_tests
 RUN (test ! -f /api/langgraph_api/js/build.mts && echo "Prebuild script not found, skipping") || tsx /api/langgraph_api/js/build.mts"""
 
     assert clean_empty_lines(actual_docker_stdin) == expected_docker_stdin
@@ -1421,6 +1422,7 @@ ENV LANGGRAPH_AUTH='{"path": "./graphs/auth.mts:auth"}'
 ENV LANGGRAPH_UI='{"agent": "./graphs/agent.ui.jsx"}'
 ENV LANGGRAPH_UI_CONFIG='{"shared": ["nuqs"]}'
 ENV LANGSERVE_GRAPHS='{"agent": "./graphs/agent.js:graph"}'
+WORKDIR /deps/unit_tests
 RUN (test ! -f /api/langgraph_api/js/build.mts && echo "Prebuild script not found, skipping") || tsx /api/langgraph_api/js/build.mts"""
 
     assert clean_empty_lines(actual_docker_stdin) == expected_docker_stdin
@@ -1548,7 +1550,8 @@ ENV NODE_VERSION=20
 WORKDIR /deps/outer-unit_tests/unit_tests
 RUN npm i && tsx /api/langgraph_api/js/build.mts
 # -- End of JS dependencies install --
-{FORMATTED_CLEANUP_LINES}"""
+{FORMATTED_CLEANUP_LINES}
+WORKDIR /deps/outer-unit_tests/unit_tests"""
 
     assert clean_empty_lines(actual_docker_stdin) == expected_docker_stdin
     assert additional_contexts == {}
@@ -1592,7 +1595,8 @@ ENV NODE_VERSION=22
 WORKDIR /deps/outer-unit_tests/unit_tests
 RUN npm i && tsx /api/langgraph_api/js/build.mts
 # -- End of JS dependencies install --
-{FORMATTED_CLEANUP_LINES}"""
+{FORMATTED_CLEANUP_LINES}
+WORKDIR /deps/outer-unit_tests/unit_tests"""
 
     assert clean_empty_lines(actual_docker_stdin) == expected_docker_stdin
     assert additional_contexts == {}
@@ -2148,8 +2152,7 @@ def test_config_to_docker_uv_lock_detects_js_pm_from_target_package_root():
             config_path, config, base_image="langchain/langgraph-api:0.2.47"
         )
 
-        # Package install, JS install and runtime all share this dir: one WORKDIR.
-        assert docker.splitlines().count("WORKDIR /deps/workspace/apps/agent") == 1
+        assert "WORKDIR /deps/workspace/apps/agent" in docker
         assert (
             "RUN pnpm i --frozen-lockfile && tsx /api/langgraph_api/js/build.mts"
             in docker
@@ -2800,7 +2803,7 @@ def test_config_retain_build_tools():
         "/usr/local/lib/python*/site-packages/" + pckg + "*" in docker_true
         for pckg in _BUILD_TOOLS
     )
-    assert "(pip uninstall -y pip setuptools wheel)" not in docker_true
+    assert "RUN pip uninstall -y pip setuptools wheel" not in docker_true
     config_false = validate_config(
         {**copy.deepcopy(base_config), "keep_pkg_tools": False}
     )
@@ -2811,7 +2814,7 @@ def test_config_retain_build_tools():
         "/usr/local/lib/python*/site-packages/" + pckg + "*" in docker_false
         for pckg in _BUILD_TOOLS
     )
-    assert "(pip uninstall -y pip setuptools wheel)" in docker_false
+    assert "RUN pip uninstall -y pip setuptools wheel" in docker_false
     config_list = validate_config(
         {**copy.deepcopy(base_config), "keep_pkg_tools": ["pip", "setuptools"]}
     )
@@ -2826,15 +2829,8 @@ def test_config_retain_build_tools():
         "/usr/local/lib/python*/site-packages/" + pckg + "*" in docker_list
         for pckg in ("pip", "setuptools")
     )
-    assert "(pip uninstall -y wheel)" in docker_list
-    assert "(pip uninstall -y pip setuptools)" not in docker_list
-
-
-def test_pip_cleanup_is_a_single_run_with_scoped_fallbacks():
-    cleanup = _get_pip_cleanup_lines("uv pip install --system", _BUILD_TOOLS, "uv")
-    assert cleanup.count("RUN ") == 1
-    # Each `|| true` must close its own subshell so it can't mask later steps.
-    assert cleanup.count("|| true)") == cleanup.count("|| true") == 2
+    assert "RUN pip uninstall -y wheel" in docker_list
+    assert "RUN pip uninstall -y pip setuptools" not in docker_list
 
 
 # config_to_compose

libs/langgraph/langgraph/channels/__init__.py

@@ -4,6 +4,7 @@ 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,
@@ -22,6 +23,9 @@ __all__ = (
     "EphemeralValue",
     "BinaryOperatorAggregate",
     "DeltaChannel",
+    "Log",
+    "LogRef",
+    "MessagesLog",
     "NamedBarrierValue",
     "NamedBarrierValueAfterFinish",
     # topics

libs/langgraph/langgraph/channels/binop.py

@@ -1,4 +1,5 @@
 import collections.abc
+import warnings
 from collections.abc import Callable, Sequence
 from typing import Any, Generic
 
@@ -29,6 +30,26 @@ 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).
 
@@ -140,6 +161,8 @@ 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

@@ -0,0 +1,328 @@
+"""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/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
+from langgraph.warnings import LangGraphDeprecatedSinceV10, LangGraphDeprecatedSinceV13
 
 __all__ = (
     "add_messages",
@@ -41,7 +41,27 @@ 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]:
@@ -57,6 +77,7 @@ 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)
 
 
@@ -213,9 +234,11 @@ 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()
@@ -235,6 +258,10 @@ 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/_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
+from typing import Any, Literal, cast
 
 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 _DeltaSnapshot
+from langgraph.checkpoint.serde.types import LogStep, _DeltaSnapshot
 
 from langgraph._internal._config import (
     DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT,
@@ -35,6 +35,7 @@ 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
@@ -325,6 +326,10 @@ 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
@@ -455,12 +460,39 @@ 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.
 
@@ -487,12 +519,20 @@ 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
@@ -505,6 +545,7 @@ 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] = {}
@@ -522,12 +563,20 @@ 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,6 +3,7 @@ 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 (
@@ -12,6 +13,7 @@ from contextlib import (
     ExitStack,
 )
 from datetime import datetime, timezone
+from functools import partial
 from types import TracebackType
 from typing import (
     Any,
@@ -19,7 +21,7 @@ from typing import (
     TypeVar,
     cast,
 )
-from uuid import UUID, uuid5
+from uuid import UUID, uuid4, uuid5
 
 from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
@@ -33,6 +35,7 @@ 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
 
@@ -72,6 +75,7 @@ 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 (
@@ -356,6 +360,9 @@ 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]:
@@ -513,6 +520,7 @@ 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,
                 {
@@ -522,13 +530,21 @@ 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(
-                    self.checkpointer_put_writes,
+                    put_writes_fn,
                     config,
                     writes_to_save,
                     task_id,
@@ -536,13 +552,16 @@ class PregelLoop:
                 )
             else:
                 fut = self.submit(
-                    self.checkpointer_put_writes,
+                    put_writes_fn,
                     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) for c, _ in writes_to_save
+                isinstance(self.specs.get(c), (DeltaChannel, Log))
+                for c, _ in writes_to_save
             ):
                 self._delta_write_futs.append(fut)
             # ERROR_SOURCE_NODE is only appended by commit() when the task
@@ -556,6 +575,24 @@ 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
@@ -579,6 +616,7 @@ 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"
             ):
@@ -1311,6 +1349,8 @@ 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:
@@ -1383,6 +1423,51 @@ 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
@@ -1709,6 +1794,13 @@ 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,
@@ -1821,6 +1913,10 @@ 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
@@ -1865,7 +1961,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (
-            [(str(tid), k, v) for tid, k, v in saved.pending_writes]
+            [
+                (str(tid), k, self._log_write_from_saver(k, v))
+                for tid, k, v in saved.pending_writes
+            ]
             if saved.pending_writes is not None
             else []
         )
@@ -1965,6 +2064,13 @@ 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,
@@ -2077,6 +2183,10 @@ 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
@@ -2121,7 +2231,10 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (
-            [(str(tid), k, v) for tid, k, v in saved.pending_writes]
+            [
+                (str(tid), k, self._log_write_from_saver(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,6 +26,7 @@ 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
@@ -537,6 +538,13 @@ 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/main.py

@@ -49,6 +49,8 @@ from langgraph.checkpoint.base import (
     ChannelVersions,
     Checkpoint,
     CheckpointTuple,
+    LogPage,
+    log_range,
 )
 from langgraph.store.base import BaseStore
 from pydantic import BaseModel, TypeAdapter
@@ -110,6 +112,7 @@ 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
@@ -1245,6 +1248,7 @@ class Pregel(
             saved.checkpoint,
             saver=saver,
             config=saved.config,
+            load_logs=False,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
@@ -1370,6 +1374,7 @@ class Pregel(
             saved.checkpoint,
             saver=saver,
             config=saved.config,
+            load_logs=False,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
@@ -1482,6 +1487,97 @@ 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/warnings.py

@@ -7,6 +7,7 @@ __all__ = (
     "LangGraphDeprecatedSinceV05",
     "LangGraphDeprecatedSinceV10",
     "LangGraphDeprecatedSinceV11",
+    "LangGraphDeprecatedSinceV13",
 )
 
 
@@ -67,3 +68,10 @@ 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/tests/test_log_channel.py

@@ -0,0 +1,718 @@
+"""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/sdk-py/langgraph_sdk/_async/threads.py

@@ -21,6 +21,7 @@ from langgraph_sdk.schema import (
     ThreadSelectField,
     ThreadSortBy,
     ThreadState,
+    ThreadStateSelectField,
     ThreadStatus,
     ThreadStreamMode,
     ThreadUpdateStateResponse,
@@ -489,6 +490,7 @@ class ThreadsClient:
         checkpoint_id: str | None = None,  # deprecated
         *,
         subgraphs: bool = False,
+        select: list[ThreadStateSelectField] | None = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> ThreadState:
@@ -499,6 +501,9 @@ class ThreadsClient:
             checkpoint: The checkpoint to get the state of.
             checkpoint_id: (deprecated) The checkpoint ID to get the state of.
             subgraphs: Include subgraphs states.
+            select: Fields to include in the response. Use `values.<key>` to
+                return a single state key (e.g. `values.messages`), which lets the
+                server skip loading the rest of the state. Defaults to all fields.
             headers: Optional custom headers to include with the request.
             params: Optional query parameters to include with the request.
 
@@ -591,16 +596,30 @@ class ThreadsClient:
                     }
             }
             ```
+
+            Fetch only the `messages` state key:
+
+            ```python
+            thread_state = await client.threads.get_state(
+                thread_id="my_thread_id",
+                select=["values.messages"],
+            )
+            ```
         """
         if checkpoint:
+            payload: dict[str, Any] = {"checkpoint": checkpoint, "subgraphs": subgraphs}
+            if select:
+                payload["select"] = select
             return await self.http.post(
                 f"/threads/{_quote_path_param(thread_id)}/state/checkpoint",
-                json={"checkpoint": checkpoint, "subgraphs": subgraphs},
+                json=payload,
                 headers=headers,
                 params=params,
             )
         elif checkpoint_id:
-            get_params = {"subgraphs": subgraphs}
+            get_params: dict[str, Any] = {"subgraphs": subgraphs}
+            if select:
+                get_params["select"] = select
             if params:
                 get_params = {**get_params, **dict(params)}
             return await self.http.get(
@@ -609,7 +628,9 @@ class ThreadsClient:
                 headers=headers,
             )
         else:
-            get_params = {"subgraphs": subgraphs}
+            get_params: dict[str, Any] = {"subgraphs": subgraphs}
+            if select:
+                get_params["select"] = select
             if params:
                 get_params = {**get_params, **dict(params)}
             return await self.http.get(
@@ -692,6 +713,7 @@ class ThreadsClient:
         before: str | Checkpoint | None = None,
         metadata: Mapping[str, Any] | None = None,
         checkpoint: Checkpoint | None = None,
+        select: list[ThreadStateSelectField] | None = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> list[ThreadState]:
@@ -703,6 +725,9 @@ class ThreadsClient:
             limit: The maximum number of states to return.
             before: Return states before this checkpoint.
             metadata: Filter states by metadata key-value pairs.
+            select: Fields to include in the response. Use `values.<key>` to
+                return a single state key (e.g. `values.messages`), which lets the
+                server skip loading the rest of the state. Defaults to all fields.
             headers: Optional custom headers to include with the request.
             params: Optional query parameters to include with the request.
 
@@ -729,6 +754,8 @@ class ThreadsClient:
             payload["metadata"] = metadata
         if checkpoint:
             payload["checkpoint"] = checkpoint
+        if select:
+            payload["select"] = select
         return await self.http.post(
             f"/threads/{_quote_path_param(thread_id)}/history",
             json=payload,

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

@@ -21,6 +21,7 @@ from langgraph_sdk.schema import (
     ThreadSelectField,
     ThreadSortBy,
     ThreadState,
+    ThreadStateSelectField,
     ThreadStatus,
     ThreadStreamMode,
     ThreadUpdateStateResponse,
@@ -480,6 +481,7 @@ class SyncThreadsClient:
         checkpoint_id: str | None = None,  # deprecated
         *,
         subgraphs: bool = False,
+        select: list[ThreadStateSelectField] | None = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> ThreadState:
@@ -489,6 +491,9 @@ class SyncThreadsClient:
             thread_id: The ID of the thread to get the state of.
             checkpoint: The checkpoint to get the state of.
             subgraphs: Include subgraphs states.
+            select: Fields to include in the response. Use `values.<key>` to
+                return a single state key (e.g. `values.messages`), which lets the
+                server skip loading the rest of the state. Defaults to all fields.
             headers: Optional custom headers to include with the request.
 
         Returns:
@@ -581,16 +586,29 @@ class SyncThreadsClient:
             }
             ```
 
+            Fetch only the `messages` state key:
+
+            ```python
+            thread_state = client.threads.get_state(
+                thread_id="my_thread_id",
+                select=["values.messages"],
+            )
+            ```
         """
         if checkpoint:
+            payload: dict[str, Any] = {"checkpoint": checkpoint, "subgraphs": subgraphs}
+            if select:
+                payload["select"] = select
             return self.http.post(
                 f"/threads/{_quote_path_param(thread_id)}/state/checkpoint",
-                json={"checkpoint": checkpoint, "subgraphs": subgraphs},
+                json=payload,
                 headers=headers,
                 params=params,
             )
         elif checkpoint_id:
-            get_params = {"subgraphs": subgraphs}
+            get_params: dict[str, Any] = {"subgraphs": subgraphs}
+            if select:
+                get_params["select"] = select
             if params:
                 get_params = {**get_params, **dict(params)}
             return self.http.get(
@@ -599,7 +617,9 @@ class SyncThreadsClient:
                 headers=headers,
             )
         else:
-            get_params = {"subgraphs": subgraphs}
+            get_params: dict[str, Any] = {"subgraphs": subgraphs}
+            if select:
+                get_params["select"] = select
             if params:
                 get_params = {**get_params, **dict(params)}
             return self.http.get(
@@ -679,6 +699,7 @@ class SyncThreadsClient:
         before: str | Checkpoint | None = None,
         metadata: Mapping[str, Any] | None = None,
         checkpoint: Checkpoint | None = None,
+        select: list[ThreadStateSelectField] | None = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> list[ThreadState]:
@@ -690,6 +711,9 @@ class SyncThreadsClient:
             limit: The maximum number of states to return.
             before: Return states before this checkpoint.
             metadata: Filter states by metadata key-value pairs.
+            select: Fields to include in the response. Use `values.<key>` to
+                return a single state key (e.g. `values.messages`), which lets the
+                server skip loading the rest of the state. Defaults to all fields.
             headers: Optional custom headers to include with the request.
 
         Returns:
@@ -717,6 +741,8 @@ class SyncThreadsClient:
             payload["metadata"] = metadata
         if checkpoint:
             payload["checkpoint"] = checkpoint
+        if select:
+            payload["select"] = select
         return self.http.post(
             f"/threads/{_quote_path_param(thread_id)}/history",
             json=payload,

libs/sdk-py/langgraph_sdk/schema.py

@@ -480,6 +480,21 @@ ThreadSelectField = Literal[
     "interrupts",
 ]
 
+ThreadStateSelectField = (
+    Literal[
+        "values",
+        "next",
+        "tasks",
+        "checkpoint",
+        "metadata",
+        "created_at",
+        "parent_checkpoint",
+        "interrupts",
+    ]
+    | str
+)
+"""Also accepts `values.<key>` for a single state key, e.g. `values.messages`."""
+
 RunSelectField = Literal[
     "run_id",
     "thread_id",

libs/sdk-py/tests/test_threads_client.py

@@ -11,6 +11,14 @@ from langgraph_sdk.client import (
     SyncThreadsClient,
     ThreadsClient,
 )
+from langgraph_sdk.schema import Checkpoint
+
+CHECKPOINT: Checkpoint = {
+    "thread_id": "thread_123",
+    "checkpoint_ns": "",
+    "checkpoint_id": "cp_1",
+    "checkpoint_map": None,
+}
 
 
 @pytest.mark.asyncio
@@ -56,3 +64,121 @@ def test_sync_threads_update_return_minimal():
         )
 
     assert result is None
+
+
+@pytest.mark.asyncio
+async def test_async_threads_get_state_select():
+    requests: list[httpx.Request] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json={})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        threads_client = ThreadsClient(HttpClient(client))
+        await threads_client.get_state("thread_123", select=["values.messages", "next"])
+        await threads_client.get_state(
+            "thread_123", checkpoint_id="cp_1", select=["values.messages"]
+        )
+        await threads_client.get_state(
+            "thread_123",
+            checkpoint=CHECKPOINT,
+            select=["values.messages"],
+        )
+        await threads_client.get_state("thread_123")
+
+    latest, by_id, by_checkpoint, default = requests
+    assert latest.method == "GET"
+    assert latest.url.path == "/threads/thread_123/state"
+    assert latest.url.params.get_list("select") == ["values.messages", "next"]
+    assert by_id.method == "GET"
+    assert by_id.url.path == "/threads/thread_123/state/cp_1"
+    assert by_id.url.params.get_list("select") == ["values.messages"]
+    assert by_checkpoint.method == "POST"
+    assert by_checkpoint.url.path == "/threads/thread_123/state/checkpoint"
+    assert json.loads(by_checkpoint.content) == {
+        "checkpoint": CHECKPOINT,
+        "subgraphs": False,
+        "select": ["values.messages"],
+    }
+    assert "select" not in default.url.params
+
+
+@pytest.mark.asyncio
+async def test_async_threads_get_history_select():
+    bodies: list[dict] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/threads/thread_123/history"
+        bodies.append(json.loads(request.content))
+        return httpx.Response(200, json=[])
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        threads_client = ThreadsClient(HttpClient(client))
+        await threads_client.get_history("thread_123", select=["values.messages"])
+        await threads_client.get_history("thread_123")
+
+    assert bodies == [{"limit": 10, "select": ["values.messages"]}, {"limit": 10}]
+
+
+def test_sync_threads_get_state_select():
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json={})
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        threads_client = SyncThreadsClient(SyncHttpClient(client))
+        threads_client.get_state("thread_123", select=["values.messages", "next"])
+        threads_client.get_state(
+            "thread_123", checkpoint_id="cp_1", select=["values.messages"]
+        )
+        threads_client.get_state(
+            "thread_123",
+            checkpoint=CHECKPOINT,
+            select=["values.messages"],
+        )
+        threads_client.get_state("thread_123")
+
+    latest, by_id, by_checkpoint, default = requests
+    assert latest.method == "GET"
+    assert latest.url.path == "/threads/thread_123/state"
+    assert latest.url.params.get_list("select") == ["values.messages", "next"]
+    assert by_id.method == "GET"
+    assert by_id.url.path == "/threads/thread_123/state/cp_1"
+    assert by_id.url.params.get_list("select") == ["values.messages"]
+    assert by_checkpoint.method == "POST"
+    assert by_checkpoint.url.path == "/threads/thread_123/state/checkpoint"
+    assert json.loads(by_checkpoint.content) == {
+        "checkpoint": CHECKPOINT,
+        "subgraphs": False,
+        "select": ["values.messages"],
+    }
+    assert "select" not in default.url.params
+
+
+def test_sync_threads_get_history_select():
+    bodies: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/threads/thread_123/history"
+        bodies.append(json.loads(request.content))
+        return httpx.Response(200, json=[])
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        threads_client = SyncThreadsClient(SyncHttpClient(client))
+        threads_client.get_history("thread_123", select=["values.messages"])
+        threads_client.get_history("thread_123")
+
+    assert bodies == [{"limit": 10, "select": ["values.messages"]}, {"limit": 10}]