feat(langgraph): append-only TranscriptChannel; edits and removals fork

libs/langgraph/langgraph/_internal/_constants.py

@@ -74,8 +74,6 @@ CONFIG_KEY_RUNTIME = sys.intern("__pregel_runtime")
 # holds a `Runtime` instance with context, store, stream writer, etc.
 CONFIG_KEY_RESUME_MAP = sys.intern("__pregel_resume_map")
 # holds a mapping of task ns -> resume value for resuming tasks
-CONFIG_KEY_STRICT_RESUME = sys.intern("__pregel_strict_resume")
-# when True, the root graph refuses a resume that can't be matched to its pending interrupts
 CONFIG_KEY_STREAM_MESSAGES_V2 = sys.intern("__pregel_stream_messages_v2")
 # when True, attach StreamMessagesHandlerV2 so content-block (v2) events
 # flow through stream_mode="messages"; set by StreamingHandler only.
@@ -136,7 +134,6 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     CONFIG_KEY_RESUME_MAP,
-    CONFIG_KEY_STRICT_RESUME,
     CONFIG_KEY_STREAM_MESSAGES_V2,
     # other constants
     PUSH,

libs/langgraph/langgraph/channels/transcript.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+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 InvalidUpdateError
+
+__all__ = ("TranscriptChannel", "TranscriptEntry", "TranscriptRef")
+
+EntryKind = Literal["append", "reset"]
+
+
+@dataclass(frozen=True, slots=True)
+class TranscriptEntry:
+    """One log entry. `append` carries a message, `reset` the full list."""
+
+    seq: int
+    kind: EntryKind
+    item_id: str = ""
+    value: Any = None
+
+
+@dataclass(slots=True)
+class TranscriptRef:
+    """Checkpoint value of a `TranscriptChannel`: a pointer into a log.
+
+    Written: `entries` holds entries with seq in (base_head, head].
+    Read: `entries` holds every entry in [1, head], fork ancestors included.
+    """
+
+    transcript_id: str
+    head: int
+    base_head: int = 0
+    entries: list[TranscriptEntry] = field(default_factory=list)
+    parent_transcript_id: str = ""
+    fork_seq: int = 0
+    is_tip: bool = True
+
+
+def _coerce(update: Any) -> list[Any]:
+    from langchain_core.messages import (
+        BaseMessageChunk,
+        convert_to_messages,
+        message_chunk_to_message,
+    )
+
+    items = update if isinstance(update, list) else [update]
+    out = []
+    for m in convert_to_messages(items):
+        m = message_chunk_to_message(m) if isinstance(m, BaseMessageChunk) else m
+        if m.id is None:
+            m.id = str(uuid.uuid4())
+        out.append(m)
+    return out
+
+
+def _is_remove(m: Any) -> bool:
+    from langchain_core.messages import RemoveMessage
+
+    return isinstance(m, RemoveMessage)
+
+
+def _remove_all_id() -> str:
+    from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+    return REMOVE_ALL_MESSAGES
+
+
+class TranscriptChannel(BaseChannel[list, Any, TranscriptRef]):
+    """Message list persisted as an append-only log.
+
+    The checkpoint stores `(transcript_id, head)`. A checkpointer that
+    supports transcripts persists new entries atomically with the checkpoint
+    and returns them on read, so storage grows linearly and loading any
+    checkpoint costs O(len(transcript)).
+
+    Writes contain only new or changed messages:
+
+    * a new id appends;
+    * an existing id (edit) or a `RemoveMessage` forks: a new transcript
+      shares the prefix before the edited message and replays what follows;
+    * `REMOVE_ALL_MESSAGES` or `Overwrite` writes one `reset` entry.
+
+    Loading a checkpoint that is not its transcript's tip also forks.
+    """
+
+    __slots__ = (
+        "value",
+        "pos",
+        "anchor",
+        "resets",
+        "transcript_id",
+        "head",
+        "base_head",
+        "pending",
+        "parent_transcript_id",
+        "fork_seq",
+    )
+
+    def __init__(self, typ: Any = list) -> None:
+        super().__init__(typ)
+        self.value: list[Any] = []
+        # id -> index in value
+        self.pos: dict[str, int] = {}
+        # id -> seq of the entry that introduced it (an append or a reset)
+        self.anchor: dict[str, int] = {}
+        # seqs of reset entries
+        self.resets: set[int] = set()
+        self.transcript_id = str(uuid.uuid4())
+        self.head = 0
+        # Highest seq known committed for this channel's lineage.
+        self.base_head = 0
+        self.pending: list[TranscriptEntry] = []
+        # Set only for a fork this channel created and has not seen committed.
+        self.parent_transcript_id = ""
+        self.fork_seq = 0
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, TranscriptChannel)
+
+    @property
+    def ValueType(self) -> Any:
+        return list
+
+    @property
+    def UpdateType(self) -> Any:
+        return Any
+
+    def _new(self) -> Self:
+        new = self.__class__(self.typ)
+        new.key = self.key
+        return new
+
+    def copy(self) -> Self:
+        new = self._new()
+        new.value = list(self.value)
+        new.pos = dict(self.pos)
+        new.anchor = dict(self.anchor)
+        new.resets = set(self.resets)
+        new.transcript_id = self.transcript_id
+        new.head = self.head
+        new.base_head = self.base_head
+        new.pending = list(self.pending)
+        new.parent_transcript_id = self.parent_transcript_id
+        new.fork_seq = self.fork_seq
+        return new
+
+    def from_checkpoint(self, checkpoint: Any) -> Self:
+        new = self._new()
+        if checkpoint is MISSING:
+            return new
+        if not isinstance(checkpoint, TranscriptRef):
+            # Plain list from a regular channel: start a transcript from it.
+            new._reset(_coerce(list(checkpoint)))
+            return new
+        for e in checkpoint.entries:
+            if e.kind == "reset":
+                new._load_reset(e.seq, e.value)
+            else:
+                new._load_append(e.seq, e.value)
+        new.head = new.base_head = checkpoint.head
+        if checkpoint.is_tip:
+            new.transcript_id = checkpoint.transcript_id
+        else:
+            new.parent_transcript_id = checkpoint.transcript_id
+            new.fork_seq = checkpoint.head
+        return new
+
+    def get(self) -> list[Any]:
+        return self.value
+
+    def is_available(self) -> bool:
+        return True
+
+    def update(self, values: Sequence[Any]) -> bool:
+        if not values:
+            return False
+        for v in values:
+            is_overwrite, overwrite = _get_overwrite(v)
+            if is_overwrite:
+                self._reset(_coerce(list(overwrite)))
+                continue
+            msgs = _coerce(v)
+            remove_all = [
+                i
+                for i, m in enumerate(msgs)
+                if _is_remove(m) and m.id == _remove_all_id()
+            ]
+            if remove_all:
+                self._reset(msgs[remove_all[-1] + 1 :])
+                continue
+            for m in msgs:
+                self._write(m)
+        return True
+
+    def _write(self, m: Any) -> None:
+        i = self.pos.get(m.id)
+        if i is None:
+            if _is_remove(m):
+                raise InvalidUpdateError(
+                    f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
+                )
+            self._append(m)
+            return
+        current = self.value[i]
+        if current is m or (not _is_remove(m) and current == m):
+            # Resend of a message already in the transcript.
+            return
+        replacement = [] if _is_remove(m) else [m]
+        self._fork_at(m.id, replacement)
+
+    def _fork_at(self, item_id: str, replacement: list[Any]) -> None:
+        """Fork before `item_id`, swap it for `replacement`, replay the rest."""
+        # The fork point must already be committed, so it is at most base_head.
+        fork = min(self.anchor[item_id] - 1, self.base_head)
+        prefix = [m for m in self.value if self.anchor[m.id] <= fork]
+        # Entries in (fork, head] that brought in the tail, in order. After a
+        # reset no older item survives, so a reset group is the whole list.
+        groups: list[tuple[int, list[Any]]] = []
+        for m in self.value:
+            a = self.anchor[m.id]
+            if a <= fork:
+                continue
+            if groups and groups[-1][0] == a:
+                groups[-1][1].append(m)
+            else:
+                groups.append((a, [m]))
+
+        # Items surviving the latest reset after the fork point all anchor at
+        # or after it, so replay starts with that reset, even when empty.
+        latest_reset = max((r for r in self.resets if r > fork), default=None)
+        if latest_reset is not None and (not groups or groups[0][0] != latest_reset):
+            groups.insert(0, (latest_reset, []))
+        old_resets = self.resets
+
+        if self.parent_transcript_id:
+            # Fork created this run: base_head == fork_seq, so branch off its parent.
+            parent = self.parent_transcript_id
+        elif fork == 0:
+            parent = ""
+        else:
+            parent = self.transcript_id
+        resets = {r for r in self.resets if r <= fork}
+        anchors = {m.id: self.anchor[m.id] for m in prefix}
+
+        self.transcript_id = str(uuid.uuid4())
+        self.parent_transcript_id = parent
+        self.fork_seq = self.head = self.base_head = fork
+        self.pending = []
+        self.value = prefix
+        self.pos = {m.id: i for i, m in enumerate(prefix)}
+        self.anchor = anchors
+        self.resets = resets
+        for a, group in groups:
+            swapped = [
+                n for m in group for n in (replacement if m.id == item_id else [m])
+            ]
+            if a in old_resets:
+                self._reset(swapped)
+            else:
+                for m in swapped:
+                    self._append(m)
+
+    def _append(self, m: Any) -> None:
+        self.head += 1
+        self.pending.append(TranscriptEntry(self.head, "append", m.id, m))
+        self._load_append(self.head, m)
+
+    def _reset(self, msgs: list[Any]) -> None:
+        self.head += 1
+        self.pending.append(TranscriptEntry(self.head, "reset", "", list(msgs)))
+        self._load_reset(self.head, msgs)
+
+    def _load_append(self, seq: int, m: Any) -> None:
+        self.pos[m.id] = len(self.value)
+        self.anchor[m.id] = seq
+        self.value.append(m)
+
+    def _load_reset(self, seq: int, msgs: list[Any]) -> None:
+        self.value = list(msgs)
+        self.pos = {m.id: i for i, m in enumerate(self.value)}
+        self.anchor = dict.fromkeys(self.pos, seq)
+        self.resets.add(seq)
+
+    def checkpoint(self) -> TranscriptRef:
+        return TranscriptRef(
+            transcript_id=self.transcript_id,
+            head=self.head,
+            base_head=self.base_head,
+            entries=list(self.pending),
+            parent_transcript_id=self.parent_transcript_id,
+            fork_seq=self.fork_seq,
+        )

libs/langgraph/langgraph/errors.py

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

libs/langgraph/langgraph/pregel/_loop.py

@@ -48,7 +48,6 @@ from langgraph._internal._constants import (
     CONFIG_KEY_RUNTIME,
     CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_STREAM,
-    CONFIG_KEY_STRICT_RESUME,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_THREAD_ID,
     ERROR,
@@ -78,7 +77,6 @@ from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import (
     EmptyInputError,
     GraphInterrupt,
-    UnknownInterruptError,
 )
 from langgraph.managed.base import (
     ManagedValueMapping,
@@ -1026,45 +1024,17 @@ class PregelLoop:
                         "Cannot use Command(resume=...) without checkpointer"
                     )
 
-                strict = (
-                    self.config[CONF].get(CONFIG_KEY_STRICT_RESUME, False)
-                    # Nested calls inherit the parent's configurable.
-                    and not self.is_nested
-                    # A finished task can rerun and interrupt again on replay, while
-                    # its earlier output still makes its interrupt look answered.
-                    and not self._addressed_checkpoint_superseded
-                )
                 if resume_is_map := (
                     isinstance(resume, dict)
                     and all(is_xxh3_128_hexdigest(k) for k in resume)
                 ):
-                    if strict:
-                        pending = self._pending_interrupts()
-                        if not resume:
-                            raise UnknownInterruptError(
-                                "The resume map names no interrupt",
-                                reason="empty_map",
-                                pending_ids=sorted(pending),
-                            )
-                        if unknown := sorted(resume.keys() - pending):
-                            raise UnknownInterruptError(
-                                f"Unknown or already-consumed interrupt: {unknown[0]}",
-                                reason="not_pending",
-                                interrupt_ids=unknown,
-                                pending_ids=sorted(pending),
-                            )
                     self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
                 else:
-                    if len(pending := self._pending_interrupts()) > 1:
-                        message = (
+                    if len(self._pending_interrupts()) > 1:
+                        raise RuntimeError(
                             "When there are multiple pending interrupts, you must specify the interrupt id when resuming. "
-                            "Docs: https://docs.langchain.com/oss/python/langgraph/interrupts#handling-multiple-interrupts."
+                            "Docs: https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation."
                         )
-                        if strict:
-                            raise UnknownInterruptError(
-                                message, reason="ambiguous", pending_ids=sorted(pending)
-                            )
-                        raise RuntimeError(message)
 
             writes: defaultdict[str, list[tuple[str, Any]]] = defaultdict(list)
             # group writes by task ID

libs/langgraph/langgraph/pregel/remote.py

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

libs/langgraph/tests/test_channels_transcript.py

@@ -0,0 +1,119 @@
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+
+from langgraph._internal._typing import MISSING
+from langgraph.channels.transcript import TranscriptChannel, TranscriptRef
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+
+class FakeStore:
+    """Commits refs like a transcript-aware checkpointer and reads them back."""
+
+    def __init__(self) -> None:
+        self.transcripts: dict[str, dict] = {}
+
+    def commit(self, ch: TranscriptChannel) -> TranscriptRef:
+        ref = ch.checkpoint()
+        t = self.transcripts.setdefault(
+            ref.transcript_id,
+            {"parent": ref.parent_transcript_id, "fork": ref.fork_seq, "entries": {}},
+        )
+        for e in ref.entries:
+            t["entries"][e.seq] = e
+        return ref
+
+    def _read(self, tid: str, head: int) -> list:
+        t = self.transcripts[tid]
+        out = self._read(t["parent"], min(head, t["fork"])) if t["parent"] else []
+        return out + [t["entries"][s] for s in sorted(t["entries"]) if s <= head]
+
+    def load(self, ref: TranscriptRef, *, tip: bool = True) -> TranscriptChannel:
+        return TranscriptChannel().from_checkpoint(
+            TranscriptRef(
+                ref.transcript_id,
+                ref.head,
+                entries=self._read(ref.transcript_id, ref.head),
+                is_tip=tip,
+            )
+        )
+
+
+def contents(ch: TranscriptChannel) -> list[str]:
+    return [m.content for m in ch.get()]
+
+
+def test_appends_and_resends() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0"), AIMessage("a0", id="a0")]])
+    ch = store.load(store.commit(ch))
+    ch.update([[HumanMessage("h1", id="h1")]])
+    ref = store.commit(ch)
+    assert [e.seq for e in ref.entries] == [3]
+    ch = store.load(ref)
+    assert contents(ch) == ["h0", "a0", "h1"]
+
+    ch.update([[AIMessage("a0", id="a0"), HumanMessage("h1", id="h1")]])
+    assert ch.checkpoint().entries == []
+
+
+def test_edit_forks_and_keeps_original() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0"), AIMessage("a0", id="a0")]])
+    ch = store.load(store.commit(ch))
+    ch.update([[HumanMessage("h1", id="h1"), AIMessage("a1", id="a1")]])
+    original = store.commit(ch)
+    ch = store.load(original)
+
+    ch.update([[AIMessage("a0-edited", id="a0")]])
+    edited = store.commit(ch)
+    assert edited.parent_transcript_id == original.transcript_id
+    assert edited.fork_seq == 1
+    assert contents(store.load(edited)) == ["h0", "a0-edited", "h1", "a1"]
+    assert contents(store.load(original, tip=False)) == ["h0", "a0", "h1", "a1"]
+
+    # A second fork before the first is committed branches off the same parent.
+    ch.update([[RemoveMessage(id="h1")]])
+    removed = store.commit(ch)
+    assert removed.parent_transcript_id == original.transcript_id
+    assert contents(store.load(removed)) == ["h0", "a0-edited", "a1"]
+
+
+def test_edit_before_any_commit_has_no_parent() -> None:
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("x", id="x"), AIMessage("y", id="y")]])
+    ch.update([[AIMessage("y2", id="y")]])
+    ref = ch.checkpoint()
+    assert (ref.parent_transcript_id, ref.fork_seq) == ("", 0)
+    assert contents(ch) == ["x", "y2"]
+
+
+def test_edit_inside_reset_replays_reset() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0")]])
+    ch = store.load(store.commit(ch))
+    ch.update(
+        [[RemoveMessage(id=REMOVE_ALL_MESSAGES), HumanMessage("p", id="p"), AIMessage("q", id="q")]]
+    )
+    ch = store.load(store.commit(ch))
+    ch.update([[AIMessage("q2", id="q")]])
+    ref = store.commit(ch)
+    assert [(e.seq, e.kind) for e in ref.entries] == [(2, "reset")]
+    assert contents(store.load(ref)) == ["p", "q2"]
+
+
+def test_load_non_tip_forks() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0")]])
+    first = store.commit(ch)
+    ch = store.load(first)
+    ch.update([[AIMessage("a0", id="a0")]])
+    store.commit(ch)
+
+    branch = store.load(first, tip=False)
+    branch.update([[AIMessage("other", id="o")]])
+    ref = store.commit(branch)
+    assert ref.parent_transcript_id == first.transcript_id
+    assert contents(store.load(ref)) == ["h0", "other"]

libs/langgraph/tests/test_remote_graph.py

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

libs/langgraph/tests/test_strict_resume.py

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

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

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

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

@@ -21,7 +21,6 @@ from langgraph_sdk.schema import (
     ThreadSelectField,
     ThreadSortBy,
     ThreadState,
-    ThreadStateSelectField,
     ThreadStatus,
     ThreadStreamMode,
     ThreadUpdateStateResponse,
@@ -490,7 +489,6 @@ 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:
@@ -501,9 +499,6 @@ 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.
 
@@ -596,30 +591,16 @@ 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=payload,
+                json={"checkpoint": checkpoint, "subgraphs": subgraphs},
                 headers=headers,
                 params=params,
             )
         elif checkpoint_id:
-            get_params: dict[str, Any] = {"subgraphs": subgraphs}
-            if select:
-                get_params["select"] = select
+            get_params = {"subgraphs": subgraphs}
             if params:
                 get_params = {**get_params, **dict(params)}
             return await self.http.get(
@@ -628,9 +609,7 @@ class ThreadsClient:
                 headers=headers,
             )
         else:
-            get_params: dict[str, Any] = {"subgraphs": subgraphs}
-            if select:
-                get_params["select"] = select
+            get_params = {"subgraphs": subgraphs}
             if params:
                 get_params = {**get_params, **dict(params)}
             return await self.http.get(
@@ -713,7 +692,6 @@ 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]:
@@ -725,9 +703,6 @@ 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.
 
@@ -754,8 +729,6 @@ 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/runs.py

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

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

@@ -21,7 +21,6 @@ from langgraph_sdk.schema import (
     ThreadSelectField,
     ThreadSortBy,
     ThreadState,
-    ThreadStateSelectField,
     ThreadStatus,
     ThreadStreamMode,
     ThreadUpdateStateResponse,
@@ -481,7 +480,6 @@ 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:
@@ -491,9 +489,6 @@ 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:
@@ -586,29 +581,16 @@ 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=payload,
+                json={"checkpoint": checkpoint, "subgraphs": subgraphs},
                 headers=headers,
                 params=params,
             )
         elif checkpoint_id:
-            get_params: dict[str, Any] = {"subgraphs": subgraphs}
-            if select:
-                get_params["select"] = select
+            get_params = {"subgraphs": subgraphs}
             if params:
                 get_params = {**get_params, **dict(params)}
             return self.http.get(
@@ -617,9 +599,7 @@ class SyncThreadsClient:
                 headers=headers,
             )
         else:
-            get_params: dict[str, Any] = {"subgraphs": subgraphs}
-            if select:
-                get_params["select"] = select
+            get_params = {"subgraphs": subgraphs}
             if params:
                 get_params = {**get_params, **dict(params)}
             return self.http.get(
@@ -699,7 +679,6 @@ 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]:
@@ -711,9 +690,6 @@ 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:
@@ -741,8 +717,6 @@ 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

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

libs/sdk-py/tests/test_threads_client.py

@@ -11,14 +11,6 @@ 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
@@ -64,121 +56,3 @@ 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}]