release(cli): 0.4.41 (#9596)

Bumps langgraph-cli from 0.4.40 to 0.4.41.

Includes:

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

.github/workflows/_integration_test.yml

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

libs/cli/langgraph_cli/__init__.py

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

libs/langgraph/langgraph/_internal/_constants.py

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

libs/langgraph/langgraph/errors.py

@@ -28,6 +28,7 @@ __all__ = (
     "ParentCommand",
     "EmptyInputError",
     "TaskNotFound",
+    "UnknownInterruptError",
     "is_invalid_resume",
 )
 
@@ -146,6 +147,45 @@ 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/_algo.py

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

libs/langgraph/langgraph/pregel/_loop.py

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

libs/langgraph/langgraph/pregel/_retry.py

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

libs/langgraph/langgraph/pregel/remote.py

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

libs/langgraph/langgraph/types.py

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

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_remote_graph.py

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

libs/langgraph/tests/test_strict_resume.py

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

libs/langgraph/uv.lock

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

libs/prebuilt/uv.lock

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

libs/sdk-py/langgraph_sdk/__init__.py

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

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

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

libs/sdk-py/langgraph_sdk/_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/runs.py

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

libs/sdk-py/langgraph_sdk/_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

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

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}]

libs/sdk-py/uv.lock

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