release(checkpoint): 4.0.3 (#7625)

## Summary

Bumps `langgraph-checkpoint` `4.0.2` → `4.0.3` and updates all
downstream `uv.lock` files.

## Changes since 4.0.2

- fix(checkpoint): revive lc=2 JSON blobs for safe types without
allowlist (#7582)
- chore: dedup warnings (#7257)
- chore(deps): bump langsmith from 0.6.4 to 0.7.31 (#7525)

.github/workflows/_integration_test.yml

@@ -121,8 +121,8 @@ jobs:
             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.0.1" ]; then
-            echo "LANGCHAIN_OPENAI_VERSION != 1.0.1; $LANGCHAIN_OPENAI_VERSION"
+          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);")

libs/checkpoint-postgres/uv.lock

@@ -259,7 +259,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/checkpoint-sqlite/uv.lock

@@ -268,7 +268,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

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

@@ -55,6 +55,19 @@ _warned_unregistered_types: set[tuple[str, str]] = set()
 _warned_blocked_types: set[tuple[str, str]] = set()
 
 
+def _is_safe_json_type(id_list: list[str]) -> bool:
+    """Return True if an lc=2 id refers to a type in SAFE_MSGPACK_TYPES.
+
+    Safe types bypass the ``allowed_json_modules`` gate so that old "json" format
+    checkpoints (written before the msgpack migration) can be resumed without
+    requiring users to configure an explicit allowlist.
+    """
+    if len(id_list) < 2:
+        return False
+    module_name = ".".join(id_list[:-1])
+    return (module_name, id_list[-1]) in _lg_msgpack.SAFE_MSGPACK_TYPES
+
+
 def _warn_once(
     seen: set[tuple[str, str]], key: tuple[str, str], msg: str, *args: object
 ) -> None:
@@ -164,19 +177,23 @@ class JsonPlusSerializer(SerializerProtocol):
         return out
 
     def _reviver(self, value: dict[str, Any]) -> Any:
-        if self._allowed_json_modules and (
+        if (
             value.get("lc", None) == 2
             and value.get("type", None) == "constructor"
             and value.get("id", None) is not None
         ):
-            try:
-                return self._revive_lc2(value)
-            except InvalidModuleError as e:
-                logger.warning(
-                    "Object %s is not in the deserialization allowlist.\n%s",
-                    value["id"],
-                    e.message,
-                )
+            id_list = value["id"]
+            is_safe = _is_safe_json_type(id_list)
+            if self._allowed_json_modules or is_safe:
+                try:
+                    return self._revive_lc2(value)
+                except InvalidModuleError as e:
+                    if not is_safe:
+                        logger.warning(
+                            "Object %s is not in the deserialization allowlist.\n%s",
+                            value["id"],
+                            e.message,
+                        )
 
         return LC_REVIVER(value)
 
@@ -224,6 +241,13 @@ class JsonPlusSerializer(SerializerProtocol):
             method_display = "<init>"
 
         dotted = ".".join(needed)
+        # Safe types (the same set already allowed for msgpack deserialization) are
+        # permitted without an explicit allowlist — they are known-safe LangGraph and
+        # LangChain types.  This restores backwards-compat for old "json" checkpoints
+        # that pre-date the msgpack migration without reopening the broader security gate.
+        if _is_safe_json_type(list(needed)):
+            return
+
         if not self._allowed_json_modules:
             raise InvalidModuleError(
                 f"Refused to deserialize JSON constructor: {dotted} (method: {method_display}). "

libs/checkpoint/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.10"

libs/checkpoint/tests/test_jsonplus.py

@@ -333,6 +333,57 @@ def test_serde_jsonplus_bytes() -> None:
     assert serde.loads_typed(dumped) == some_bytes
 
 
+def test_lc2_json_safe_type_revives_without_allowlist() -> None:
+    """Old 'json' blobs with lc=2 for safe types must revive without an explicit allowlist.
+
+    Regression test for: https://github.com/langchain-ai/langgraph/issues/7498
+    Threads checkpointed before v1.0.1 (pre-msgpack) stored messages as lc=2 JSON
+    constructor dicts. Resuming those threads must reconstruct proper BaseMessage objects
+    rather than returning raw dicts that cause MESSAGE_COERCION_FAILURE in add_messages.
+    """
+    from langchain_core.messages import AIMessage
+
+    serde = JsonPlusSerializer()  # default: _allowed_json_modules=None
+
+    human_blob = {
+        "lc": 2,
+        "type": "constructor",
+        "id": ["langchain_core", "messages", "human", "HumanMessage"],
+        "kwargs": {"content": "hello", "type": "human"},
+    }
+    ai_blob = {
+        "lc": 2,
+        "type": "constructor",
+        "id": ["langchain_core", "messages", "ai", "AIMessage"],
+        "kwargs": {"content": "hi there", "type": "ai"},
+    }
+    result = serde.loads_typed(("json", json.dumps([human_blob, ai_blob]).encode()))
+
+    assert len(result) == 2
+    assert isinstance(result[0], HumanMessage), (
+        f"Expected HumanMessage, got {type(result[0])}: {result[0]!r}\n"
+        "lc=2 JSON blobs for safe types must deserialize without an explicit allowlist"
+    )
+    assert result[0].content == "hello"
+    assert isinstance(result[1], AIMessage)
+    assert result[1].content == "hi there"
+
+
+def test_lc2_json_unknown_type_stays_blocked_without_allowlist() -> None:
+    """lc=2 JSON blobs for types NOT in SAFE_MSGPACK_TYPES still require an allowlist."""
+    serde = JsonPlusSerializer()
+    load = {
+        "lc": 2,
+        "type": "constructor",
+        "id": ["pprint", "pprint"],
+        "kwargs": {"object": "HELLO"},
+    }
+    # No allowlist configured → raw dict returned (not raised, not reconstructed)
+    result = serde.loads_typed(("json", json.dumps(load).encode()))
+    assert isinstance(result, dict), "Unknown lc=2 type must stay as raw dict"
+    assert result.get("lc") == 2
+
+
 def test_deserde_invalid_module() -> None:
     serde = JsonPlusSerializer()
     load = {

libs/checkpoint/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.23"
+__version__ = "0.4.24"

libs/cli/langgraph_cli/_ignore.py

@@ -0,0 +1,124 @@
+"""Shared ignore-file handling for local source filtering."""
+
+import pathlib
+from dataclasses import dataclass
+
+import pathspec
+
+_ALWAYS_EXCLUDE = [
+    "__pycache__/",
+    ".git/",
+    ".venv/",
+    "venv/",
+    "node_modules/",
+    ".tox/",
+    ".mypy_cache/",
+]
+_ALWAYS_EXCLUDE_NAMES = frozenset(
+    pattern.rstrip("/").split("/")[-1] for pattern in _ALWAYS_EXCLUDE
+)
+_GLOB_CHARS = frozenset("*?[")
+
+
+@dataclass(frozen=True, slots=True)
+class _NegatedDockerignoreHints:
+    exact_dirs: frozenset[pathlib.PurePosixPath] = frozenset()
+    wildcard_prefixes: frozenset[pathlib.PurePosixPath] = frozenset()
+    recurse_all: bool = False
+
+    def requires_dir_walk(self, path: pathlib.PurePosixPath) -> bool:
+        if self.recurse_all or path in self.exact_dirs:
+            return True
+        return any(
+            path == prefix or path in prefix.parents or prefix in path.parents
+            for prefix in self.wildcard_prefixes
+        )
+
+
+def _build_ignore_spec(
+    directory: pathlib.Path, *, include_gitignore: bool = True
+) -> pathspec.PathSpec:
+    """Build a PathSpec combining built-in exclusions with ignore files.
+
+    Always excludes common non-source directories (`_ALWAYS_EXCLUDE`). On top
+    of that, patterns from `.dockerignore` are merged in. `.gitignore` patterns
+    are optional because some callers need Docker build-context semantics,
+    while archive creation wants both files.
+    """
+    lines: list[str] = list(_ALWAYS_EXCLUDE)
+    ignore_files = [".dockerignore"]
+    if include_gitignore:
+        ignore_files.append(".gitignore")
+    for name in ignore_files:
+        ignore_file = directory / name
+        if ignore_file.is_file():
+            lines.extend(ignore_file.read_text(encoding="utf-8").splitlines())
+    return pathspec.PathSpec.from_lines("gitwildmatch", lines)
+
+
+def _is_always_excluded(path: pathlib.PurePosixPath, *, is_dir: bool) -> bool:
+    """Whether `path` lives inside a built-in excluded directory."""
+    parent_parts = path.parts if is_dir else path.parts[:-1]
+    return any(part in _ALWAYS_EXCLUDE_NAMES for part in parent_parts)
+
+
+def _build_dockerignore_negation_hints(
+    directory: pathlib.Path,
+) -> _NegatedDockerignoreHints:
+    """Summarize which ignored directories must still be traversed.
+
+    Most negations only require walking a small, concrete chain of parent
+    directories (for example `!assets/keep.txt` requires entering `assets/`).
+    Broader glob negations may force a wider walk.
+    """
+    ignore_file = directory / ".dockerignore"
+    if not ignore_file.is_file():
+        return _NegatedDockerignoreHints()
+
+    exact_dirs: set[pathlib.PurePosixPath] = set()
+    wildcard_prefixes: set[pathlib.PurePosixPath] = set()
+    recurse_all = False
+
+    for raw_line in ignore_file.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or line.startswith("\\!"):
+            continue
+        if line.startswith("\\#"):
+            line = line[1:]
+        if not line.startswith("!"):
+            continue
+
+        pattern = line[1:].lstrip("/")
+        while pattern.startswith("./"):
+            pattern = pattern[2:]
+        pattern = pattern.rstrip("/")
+        parts = [part for part in pattern.split("/") if part and part != "."]
+        if not parts:
+            recurse_all = True
+            continue
+
+        wildcard_index = next(
+            (
+                idx
+                for idx, part in enumerate(parts)
+                if any(char in part for char in _GLOB_CHARS)
+            ),
+            None,
+        )
+        if wildcard_index is not None:
+            literal_parts = parts[:wildcard_index]
+            if not literal_parts:
+                recurse_all = True
+                continue
+            wildcard_prefixes.add(pathlib.PurePosixPath(*literal_parts))
+            continue
+
+        parent_parts = parts[:-1]
+        for idx in range(1, len(parent_parts) + 1):
+            exact_dirs.add(pathlib.PurePosixPath(*parent_parts[:idx]))
+
+    return _NegatedDockerignoreHints(
+        exact_dirs=frozenset(exact_dirs),
+        wildcard_prefixes=frozenset(wildcard_prefixes),
+        recurse_all=recurse_all,
+    )

libs/cli/langgraph_cli/archive.py

@@ -9,35 +9,12 @@ from contextlib import contextmanager
 import click
 import pathspec
 
+from langgraph_cli._ignore import _build_ignore_spec
 from langgraph_cli.config import Config, _assemble_local_deps
 
 _WARN_SIZE = 50 * 1024 * 1024  # 50 MB
 _MAX_SIZE = 200 * 1024 * 1024  # 200 MB
 
-_ALWAYS_EXCLUDE = [
-    "__pycache__/",
-    ".git/",
-    ".venv/",
-    "venv/",
-    "node_modules/",
-    ".tox/",
-    ".mypy_cache/",
-]
-
-
-def _build_ignore_spec(directory: pathlib.Path) -> pathspec.PathSpec:
-    """Build a PathSpec combining built-in exclusions with .dockerignore and .gitignore.
-
-    Always excludes common non-source directories (_ALWAYS_EXCLUDE).  On top of
-    that, patterns from .dockerignore and .gitignore (if present) are merged in.
-    """
-    lines: list[str] = list(_ALWAYS_EXCLUDE)
-    for name in (".dockerignore", ".gitignore"):
-        ignore_file = directory / name
-        if ignore_file.is_file():
-            lines.extend(ignore_file.read_text(encoding="utf-8").splitlines())
-    return pathspec.PathSpec.from_lines("gitwildmatch", lines)
-
 
 def _tar_filter(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo | None:
     """Strip symlinks, hardlinks, and traversal paths from archive."""

libs/cli/langgraph_cli/uv_lock.py

@@ -10,7 +10,13 @@ except ModuleNotFoundError:  # pragma: no cover - exercised on Python 3.10.
     import tomli as tomllib
 
 import click
+import pathspec
 
+from langgraph_cli._ignore import (
+    _build_dockerignore_negation_hints,
+    _build_ignore_spec,
+    _is_always_excluded,
+)
 from langgraph_cli.schemas import Config
 
 
@@ -440,16 +446,32 @@ def _container_root_for_uv_lock_package(
 
 
 def _uv_lock_package_copy_items(
-    package: UvLockPackage, plan: UvLockPlan
+    package: UvLockPackage,
+    plan: UvLockPlan,
+    ignore_spec: pathspec.PathSpec,
 ) -> tuple[tuple[pathlib.PurePosixPath, pathlib.PurePosixPath], ...]:
+    # Skip entries that .dockerignore / built-in exclusions would strip from
+    # the build context. Emitting `ADD <path>` for a file that Docker has
+    # filtered out causes the build to fail with
+    # "failed to compute cache key: <path> not found".
     if package.root != plan.project_root:
         relative_root = pathlib.PurePosixPath(
             *package.root.relative_to(plan.project_root).parts
         )
+        if _is_always_excluded(relative_root, is_dir=True) or ignore_spec.match_file(
+            f"{relative_root.as_posix()}/"
+        ):
+            raise click.UsageError(
+                f"Workspace member '{package.name}' at {relative_root} is "
+                "excluded from the Docker build context, but uv.lock requires "
+                "it to be copied into the build context. Remove the matching "
+                "pattern or drop the member from [tool.uv.workspace].members."
+            )
         return ((relative_root, plan.container_roots[package.root]),)
 
     root_container = plan.container_roots[package.root]
     workspace_member_roots = plan.all_workspace_roots - {plan.project_root}
+    negated_dockerignore_hints = _build_dockerignore_negation_hints(plan.project_root)
 
     def iter_entries(
         current_dir: pathlib.Path,
@@ -461,18 +483,32 @@ def _uv_lock_package_copy_items(
                 # and excluded entirely otherwise.
                 continue
 
-            descendant_member_roots = [
-                ws_root
-                for ws_root in workspace_member_roots
-                if child in ws_root.parents
-            ]
-            if child.is_dir() and descendant_member_roots:
-                entries.extend(iter_entries(child))
-                continue
-
             relative_child = pathlib.PurePosixPath(
                 *child.relative_to(plan.project_root).parts
             )
+            is_dir = child.is_dir()
+            if _is_always_excluded(relative_child, is_dir=is_dir):
+                continue
+            ignored = ignore_spec.match_file(
+                f"{relative_child.as_posix()}/" if is_dir else relative_child.as_posix()
+            )
+            is_workspace_parent = is_dir and any(
+                child in ws_root.parents for ws_root in workspace_member_roots
+            )
+
+            if is_workspace_parent:
+                entries.extend(iter_entries(child))
+                continue
+            if (
+                is_dir
+                and ignored
+                and negated_dockerignore_hints.requires_dir_walk(relative_child)
+            ):
+                entries.extend(iter_entries(child))
+                continue
+            if ignored:
+                continue
+
             entries.append(
                 (relative_child, root_container.joinpath(*relative_child.parts))
             )
@@ -956,10 +992,13 @@ def python_config_to_docker_uv_lock(
     docker_plan.add_raw("# -- End of uv.lock dependencies install --")
     docker_plan.add_blank()
 
+    ignore_spec = _build_ignore_spec(plan.project_root, include_gitignore=False)
     for package in plan.install_order:
         package_label = package.root.relative_to(plan.project_root).as_posix() or "."
         docker_plan.add_raw(f"# -- Adding workspace package {package_label} --")
-        for source, destination in _uv_lock_package_copy_items(package, plan):
+        for source, destination in _uv_lock_package_copy_items(
+            package, plan, ignore_spec
+        ):
             docker_plan.add_raw(copy_from_project_root(source, destination.as_posix()))
         docker_plan.add_instruction(
             "WORKDIR", plan.container_roots[package.root].as_posix()

libs/cli/tests/unit_tests/test_archive.py

@@ -99,6 +99,13 @@ class TestBuildIgnoreSpec:
         assert spec.match_file("app.log")
         assert spec.match_file("mod.pyc")
 
+    def test_can_skip_gitignore(self, tmp_path):
+        (tmp_path / ".dockerignore").write_text("*.log\n")
+        (tmp_path / ".gitignore").write_text("*.pyc\n")
+        spec = _build_ignore_spec(tmp_path, include_gitignore=False)
+        assert spec.match_file("app.log")
+        assert not spec.match_file("mod.pyc")
+
     def test_no_ignore_files_only_builtins(self, tmp_path):
         spec = _build_ignore_spec(tmp_path)
         assert spec.match_file("__pycache__/")

libs/cli/tests/unit_tests/test_config.py

@@ -4,6 +4,7 @@ import os
 import pathlib
 import tempfile
 import textwrap
+from unittest.mock import patch
 
 import click
 import pytest
@@ -1855,6 +1856,364 @@ def test_config_to_docker_uv_lock_supports_single_uv_project_root():
         assert additional_contexts == {}
 
 
+def test_config_to_docker_uv_lock_skips_dockerignore_entries():
+    """Entries filtered by .dockerignore / built-in excludes must not appear
+    as ADD lines. Docker fails to compute the cache key for paths that the
+    build context has stripped."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root = tmpdir_path / "single"
+        project_root.mkdir()
+        (project_root / "uv.lock").write_text("# uv lock file\n")
+        (project_root / "pyproject.toml").write_text(
+            textwrap.dedent(
+                """
+                [project]
+                name = "single-app"
+                version = "0.1.0"
+                dependencies = ["httpx>=0.28"]
+
+                [build-system]
+                requires = ["setuptools>=61"]
+                build-backend = "setuptools.build_meta"
+                """
+            ).strip()
+            + "\n"
+        )
+        (project_root / "langgraph.json").write_text("{}\n")
+        (project_root / "src").mkdir()
+        (project_root / "src" / "agent.py").write_text("graph = object()\n")
+        (project_root / "README.md").write_text("# hi\n")
+
+        # Built-in exclusions — must never appear as ADD lines.
+        (project_root / ".git").mkdir()
+        (project_root / ".git" / "HEAD").write_text("ref: refs/heads/main\n")
+        (project_root / ".venv").mkdir()
+        (project_root / ".venv" / "pyvenv.cfg").write_text("home = /usr\n")
+        (project_root / "__pycache__").mkdir()
+        (project_root / "__pycache__" / "x.cpython-311.pyc").write_bytes(b"\x00")
+
+        # .dockerignore excludes .gitignore and a custom path.
+        (project_root / ".dockerignore").write_text(".gitignore\nsecrets.env\n")
+        (project_root / ".gitignore").write_text("*.pyc\n")
+        (project_root / "secrets.env").write_text("TOKEN=abc\n")
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./src/agent.py:graph"},
+                "source": {"kind": "uv"},
+            }
+        )
+        docker, _ = config_to_docker(
+            project_root / "langgraph.json",
+            config,
+            base_image="langchain/langgraph-api:0.2.47",
+        )
+
+        for excluded in (
+            "ADD .git ",
+            "ADD .gitignore ",
+            "ADD .venv ",
+            "ADD __pycache__ ",
+            "ADD secrets.env ",
+        ):
+            assert excluded not in docker, (
+                f"{excluded!r} should be filtered out of Dockerfile:\n{docker}"
+            )
+
+        # The .dockerignore itself is still part of the context and should be
+        # ADDed (Docker needs it at build time, and archive.py includes it).
+        assert "ADD .dockerignore /deps/workspace/.dockerignore" in docker
+        assert "ADD src /deps/workspace/src" in docker
+        assert "ADD README.md /deps/workspace/README.md" in docker
+
+
+def test_config_to_docker_uv_lock_does_not_apply_gitignore():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root = tmpdir_path / "single"
+        project_root.mkdir()
+        (project_root / "uv.lock").write_text("# uv lock file\n")
+        (project_root / "pyproject.toml").write_text(
+            textwrap.dedent(
+                """
+                [project]
+                name = "single-app"
+                version = "0.1.0"
+                dependencies = ["httpx>=0.28"]
+
+                [build-system]
+                requires = ["setuptools>=61"]
+                build-backend = "setuptools.build_meta"
+                """
+            ).strip()
+            + "\n"
+        )
+        (project_root / "langgraph.json").write_text("{}\n")
+        (project_root / "src").mkdir()
+        (project_root / "src" / "agent.py").write_text("graph = object()\n")
+        (project_root / "README.md").write_text("# hi\n")
+        (project_root / ".gitignore").write_text("README.md\n")
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./src/agent.py:graph"},
+                "source": {"kind": "uv"},
+            }
+        )
+        docker, _ = config_to_docker(
+            project_root / "langgraph.json",
+            config,
+            base_image="langchain/langgraph-api:0.2.47",
+        )
+
+        assert "ADD README.md /deps/workspace/README.md" in docker
+
+
+def test_config_to_docker_uv_lock_skips_dockerignore_entries_in_workspace():
+    """Multi-member workspace: ignore patterns must filter root-level entries
+    AND entries encountered while recursing into directories that contain
+    workspace members (the `descendant_member_roots` branch)."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root, config_path = _write_uv_lock_workspace(
+            tmpdir_path,
+            agent_dependencies=["workspace-root", "shared", "httpx>=0.28"],
+            root_sources="[tool.uv.sources]\nshared = { workspace = true }\nworkspace-root = { workspace = true }",
+            agent_sources="[tool.uv.sources]\nshared = { workspace = true }\nworkspace-root = { workspace = true }",
+        )
+        root_src = project_root / "src" / "workspace_root"
+        root_src.mkdir(parents=True)
+        (root_src / "__init__.py").write_text("__all__ = []\n")
+        (project_root / "README.md").write_text("workspace root package\n")
+
+        # A non-member sibling of the `apps/agent` member that should be
+        # filtered out via .dockerignore. This exercises the recursion into
+        # `apps/` where `apps/agent` is kept (it's a member) but its sibling is
+        # filtered.
+        (project_root / "apps" / "scratch.txt").write_text("scratch\n")
+        # A root-level path that .dockerignore excludes.
+        (project_root / "secrets.env").write_text("TOKEN=abc\n")
+        (project_root / ".dockerignore").write_text("secrets.env\napps/scratch.txt\n")
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {
+                    "agent": "../../apps/agent/src/agent/graph.py:graph",
+                },
+                "source": {"kind": "uv", "root": "../..", "package": "agent"},
+            }
+        )
+        docker, _ = config_to_docker(
+            config_path, config, base_image="langchain/langgraph-api:0.2.47"
+        )
+
+        assert "COPY --from=uv-workspace-root src /deps/workspace/src" in docker
+        assert (
+            "COPY --from=uv-workspace-root README.md /deps/workspace/README.md"
+            in docker
+        )
+        assert (
+            "COPY --from=uv-workspace-root .dockerignore /deps/workspace/.dockerignore"
+            in docker
+        )
+        assert "secrets.env" not in docker
+        assert "apps/scratch.txt" not in docker
+        # Workspace members themselves are still copied via their own per-member
+        # COPY line — the sibling filter must not disturb this.
+        assert (
+            "COPY --from=uv-workspace-root apps/agent /deps/workspace/apps/agent"
+            in docker
+        )
+
+
+def test_config_to_docker_uv_lock_preserves_negated_dockerignore_descendants():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root = tmpdir_path / "single"
+        project_root.mkdir()
+        (project_root / "uv.lock").write_text("# uv lock file\n")
+        (project_root / "pyproject.toml").write_text(
+            textwrap.dedent(
+                """
+                [project]
+                name = "single-app"
+                version = "0.1.0"
+                dependencies = ["httpx>=0.28"]
+
+                [build-system]
+                requires = ["setuptools>=61"]
+                build-backend = "setuptools.build_meta"
+                """
+            ).strip()
+            + "\n"
+        )
+        (project_root / "langgraph.json").write_text("{}\n")
+        (project_root / "src").mkdir()
+        (project_root / "src" / "agent.py").write_text("graph = object()\n")
+        (project_root / "assets").mkdir()
+        (project_root / "assets" / "keep.txt").write_text("keep\n")
+        (project_root / "assets" / "drop.txt").write_text("drop\n")
+        (project_root / ".dockerignore").write_text("assets/\n!assets/keep.txt\n")
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./src/agent.py:graph"},
+                "source": {"kind": "uv"},
+            }
+        )
+        docker, _ = config_to_docker(
+            project_root / "langgraph.json",
+            config,
+            base_image="langchain/langgraph-api:0.2.47",
+        )
+
+        assert "ADD assets /deps/workspace/assets" not in docker
+        assert "ADD assets/keep.txt /deps/workspace/assets/keep.txt" in docker
+        assert "assets/drop.txt" not in docker
+
+
+def test_config_to_docker_uv_lock_prunes_unrelated_ignored_subtrees():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root = tmpdir_path / "single"
+        project_root.mkdir()
+        (project_root / "uv.lock").write_text("# uv lock file\n")
+        (project_root / "pyproject.toml").write_text(
+            textwrap.dedent(
+                """
+                [project]
+                name = "single-app"
+                version = "0.1.0"
+                dependencies = ["httpx>=0.28"]
+
+                [build-system]
+                requires = ["setuptools>=61"]
+                build-backend = "setuptools.build_meta"
+                """
+            ).strip()
+            + "\n"
+        )
+        (project_root / "langgraph.json").write_text("{}\n")
+        (project_root / "src").mkdir()
+        (project_root / "src" / "agent.py").write_text("graph = object()\n")
+        (project_root / "assets").mkdir()
+        (project_root / "assets" / "keep.txt").write_text("keep\n")
+        (project_root / "vendor").mkdir()
+        (project_root / "vendor" / "huge.txt").write_text("large\n")
+        (project_root / ".dockerignore").write_text(
+            "vendor/\nassets/\n!assets/keep.txt\n"
+        )
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./src/agent.py:graph"},
+                "source": {"kind": "uv"},
+            }
+        )
+
+        original_iterdir = pathlib.Path.iterdir
+
+        def guarded_iterdir(self):
+            if self == project_root / "vendor":
+                raise AssertionError("should not walk unrelated ignored subtree")
+            return original_iterdir(self)
+
+        with patch.object(
+            pathlib.Path, "iterdir", autospec=True, side_effect=guarded_iterdir
+        ):
+            docker, _ = config_to_docker(
+                project_root / "langgraph.json",
+                config,
+                base_image="langchain/langgraph-api:0.2.47",
+            )
+
+        assert "ADD assets/keep.txt /deps/workspace/assets/keep.txt" in docker
+        assert "vendor/huge.txt" not in docker
+
+
+def test_config_to_docker_uv_lock_never_reincludes_always_excluded_subtrees():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root = tmpdir_path / "single"
+        project_root.mkdir()
+        (project_root / "uv.lock").write_text("# uv lock file\n")
+        (project_root / "pyproject.toml").write_text(
+            textwrap.dedent(
+                """
+                [project]
+                name = "single-app"
+                version = "0.1.0"
+                dependencies = ["httpx>=0.28"]
+
+                [build-system]
+                requires = ["setuptools>=61"]
+                build-backend = "setuptools.build_meta"
+                """
+            ).strip()
+            + "\n"
+        )
+        (project_root / "langgraph.json").write_text("{}\n")
+        (project_root / "src").mkdir()
+        (project_root / "src" / "agent.py").write_text("graph = object()\n")
+        (project_root / ".venv" / "pkg").mkdir(parents=True)
+        (project_root / ".venv" / "pkg" / "keep.txt").write_text("keep\n")
+        (project_root / "node_modules" / "pkg").mkdir(parents=True)
+        (project_root / "node_modules" / "pkg" / "package.json").write_text("{}\n")
+        (project_root / ".dockerignore").write_text(
+            "!.venv/pkg/keep.txt\n!node_modules/pkg/package.json\n"
+        )
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./src/agent.py:graph"},
+                "source": {"kind": "uv"},
+            }
+        )
+        docker, _ = config_to_docker(
+            project_root / "langgraph.json",
+            config,
+            base_image="langchain/langgraph-api:0.2.47",
+        )
+
+        assert ".venv/pkg/keep.txt" not in docker
+        assert "node_modules/pkg/package.json" not in docker
+        assert "ADD src /deps/workspace/src" in docker
+
+
+def test_config_to_docker_uv_lock_rejects_ignored_workspace_member():
+    """A workspace member matched by .dockerignore cannot be copied into the
+    build context — uv.lock requires it, so fail loudly with a clear message."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root, config_path = _write_uv_lock_workspace(
+            tmpdir_path,
+            agent_sources="[tool.uv.sources]\nshared = { workspace = true }",
+        )
+        (project_root / ".dockerignore").write_text("libs/shared\n")
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "../../apps/agent/src/agent/graph.py:graph"},
+                "source": {"kind": "uv", "root": "../..", "package": "agent"},
+                "auth": {"path": "../../libs/shared/src/shared/auth.py:create_auth"},
+            }
+        )
+        with pytest.raises(
+            click.UsageError, match=r"Workspace member 'shared' at libs/shared"
+        ):
+            config_to_docker(
+                config_path, config, base_image="langchain/langgraph-api:0.2.47"
+            )
+
+
 def test_config_to_docker_uv_lock_rejects_invalid_source_package_type():
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = pathlib.Path(tmpdir)

libs/langgraph/langgraph/_internal/_config.py

@@ -306,12 +306,7 @@ def ensure_config(*configs: RunnableConfig | None) -> RunnableConfig:
         for k, v in config.items():
             if _is_not_empty(v) and k in CONFIG_KEYS:
                 if k == CONF:
-                    # Merge configurable dicts across configs so that values
-                    # bound via `with_config(...)` (e.g. `ls_agent_type`) are
-                    # preserved when later configs (e.g. invoke-time) only
-                    # specify a subset of keys like `thread_id`.
-                    existing = cast(dict, empty.get(k) or {})
-                    empty[k] = {**existing, **cast(dict, v)}
+                    empty[k] = cast(dict, v).copy()
                 else:
                     empty[k] = v  # type: ignore[literal-required]
         for k, v in config.items():

libs/langgraph/langgraph/_internal/_constants.py

@@ -56,6 +56,8 @@ CONFIG_KEY_CHECKPOINT_NS = sys.intern("checkpoint_ns")
 # holds the current checkpoint_ns, "" for root graph
 CONFIG_KEY_NODE_FINISHED = sys.intern("__pregel_node_finished")
 # holds a callback to be called when a node is finished
+CONFIG_KEY_TIMED_ATTEMPT_OBSERVER = sys.intern("__pregel_timed_attempt_observer")
+# holds a callback to be called when a timed node attempt starts or finishes
 CONFIG_KEY_SCRATCHPAD = sys.intern("__pregel_scratchpad")
 # holds a mutable dict for temporary storage scoped to the current task
 CONFIG_KEY_RUNNER_SUBMIT = sys.intern("__pregel_runner_submit")
@@ -106,6 +108,7 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_NS,
+    CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     CONFIG_KEY_RESUME_MAP,
     # other constants
     PUSH,

libs/langgraph/langgraph/_internal/_runnable.py

@@ -706,7 +706,12 @@ class RunnableSeq(Runnable):
                                 step.ainvoke(input, config, **kwargs), context=context
                             )
                     else:
-                        input = await step.ainvoke(input, config, **kwargs)
+                        with set_config_context(config) as context:
+                            input = await context.run(
+                                lambda: asyncio.create_task(
+                                    step.ainvoke(input, config, **kwargs)
+                                )
+                            )
                 else:
                     input = await step.ainvoke(input, config)
         # finish the root run

libs/langgraph/langgraph/_internal/_timeout.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Literal
+
+_SYNC_TIMEOUT_PREFIX = (
+    "Node timeouts are only supported for async nodes because sync Python "
+    "execution cannot be safely cancelled in-process."
+)
+
+
+def coerce_timeout(value: float | timedelta | None) -> float | None:
+    """Normalize a timeout to positive seconds, or None if unset."""
+    if value is None:
+        return None
+    seconds = value.total_seconds() if isinstance(value, timedelta) else float(value)
+    if seconds <= 0:
+        raise ValueError("timeout must be greater than 0")
+    return seconds
+
+
+def sync_timeout_unsupported(
+    name: str, *, kind: Literal["Node", "Task"] = "Node"
+) -> ValueError:
+    """Build the canonical error for using `timeout` with a sync target."""
+    return ValueError(f"{_SYNC_TIMEOUT_PREFIX} {kind} {name!r} is sync.")

libs/langgraph/langgraph/errors.py

@@ -20,6 +20,7 @@ __all__ = (
     "GraphBubbleUp",
     "GraphInterrupt",
     "NodeInterrupt",
+    "NodeTimeoutError",
     "ParentCommand",
     "EmptyInputError",
     "TaskNotFound",
@@ -125,3 +126,25 @@ class TaskNotFound(Exception):
     """Raised when the executor is unable to find a task (for distributed mode)."""
 
     pass
+
+
+class NodeTimeoutError(TimeoutError):
+    """Raised when a node invocation exceeds its configured `timeout`.
+
+    Subclasses the built-in `TimeoutError`, so existing `except TimeoutError`
+    handlers keep working. If the node has a `retry_policy` whose `retry_on`
+    permits `TimeoutError`, the attempt will be retried.
+    """
+
+    node: str
+    timeout: float
+    elapsed: float
+
+    def __init__(self, node: str, timeout: float, elapsed: float) -> None:
+        super().__init__(
+            f"Node '{node}' exceeded its timeout of {timeout:.3f}s "
+            f"(elapsed: {elapsed:.3f}s)."
+        )
+        self.node = node
+        self.timeout = timeout
+        self.elapsed = elapsed

libs/langgraph/langgraph/func/__init__.py

@@ -5,6 +5,7 @@ import inspect
 import warnings
 from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass
+from datetime import timedelta
 from typing import (
     Any,
     Generic,
@@ -22,6 +23,8 @@ from typing_extensions import Unpack
 
 from langgraph._internal import _serde
 from langgraph._internal._constants import CACHE_NS_WRITES, PREVIOUS
+from langgraph._internal._runnable import is_async_callable
+from langgraph._internal._timeout import coerce_timeout, sync_timeout_unsupported
 from langgraph._internal._typing import MISSING, DeprecatedKwargs
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
@@ -51,6 +54,7 @@ class _TaskFunction(Generic[P, T]):
         *,
         retry_policy: Sequence[RetryPolicy],
         cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
+        timeout: float | None = None,
         name: str | None = None,
     ) -> None:
         if name is not None:
@@ -67,6 +71,7 @@ class _TaskFunction(Generic[P, T]):
         self.func = func
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
+        self.timeout = timeout
         functools.update_wrapper(self, func)
 
     def __call__(self, *args: P.args, **kwargs: P.kwargs) -> SyncAsyncFuture[T]:
@@ -74,6 +79,7 @@ class _TaskFunction(Generic[P, T]):
             self.func,
             retry_policy=self.retry_policy,
             cache_policy=self.cache_policy,
+            timeout=self.timeout,
             *args,
             **kwargs,
         )
@@ -98,6 +104,7 @@ def task(
     name: str | None = None,
     retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
+    timeout: float | timedelta | None = None,
     **kwargs: Unpack[DeprecatedKwargs],
 ) -> Callable[
     [Callable[P, Awaitable[T]] | Callable[P, T]],
@@ -119,6 +126,7 @@ def task(
     name: str | None = None,
     retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
+    timeout: float | timedelta | None = None,
     **kwargs: Unpack[DeprecatedKwargs],
 ) -> (
     Callable[[Callable[P, Awaitable[T]] | Callable[P, T]], _TaskFunction[P, T]]
@@ -142,6 +150,9 @@ def task(
         name: An optional name for the task. If not provided, the function name will be used.
         retry_policy: An optional retry policy (or list of policies) to use for the task in case of a failure.
         cache_policy: An optional cache policy to use for the task. This allows caching of the task results.
+        timeout: Maximum wall-clock duration for a single task attempt, in seconds
+            (or as a `timedelta`). If exceeded, `NodeTimeoutError` is raised.
+            Supported only for async tasks.
 
     Returns:
         A callable function when used as a decorator.
@@ -196,6 +207,7 @@ def task(
         )
         if retry_policy is None:
             retry_policy = retry  # type: ignore[assignment]
+    timeout_s = coerce_timeout(timeout)
 
     retry_policies: Sequence[RetryPolicy] = (
         ()
@@ -208,8 +220,15 @@ def task(
     def decorator(
         func: Callable[P, Awaitable[T]] | Callable[P, T],
     ) -> Callable[P, SyncAsyncFuture[T]]:
+        if timeout_s is not None and not is_async_callable(func):
+            name_ = name or getattr(func, "__name__", func.__class__.__name__)
+            raise sync_timeout_unsupported(str(name_), kind="Task")
         return _TaskFunction(
-            func, retry_policy=retry_policies, cache_policy=cache_policy, name=name
+            func,
+            retry_policy=retry_policies,
+            cache_policy=cache_policy,
+            timeout=timeout_s,
+            name=name,
         )
 
     if __func_or_none__ is not None:
@@ -400,6 +419,7 @@ class entrypoint(Generic[ContextT]):
         context_schema: type[ContextT] | None = None,
         cache_policy: CachePolicy | None = None,
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
         """Initialize the entrypoint decorator."""
@@ -426,6 +446,7 @@ class entrypoint(Generic[ContextT]):
         self.cache = cache
         self.cache_policy = cache_policy
         self.retry_policy = retry_policy
+        self.timeout = coerce_timeout(timeout)
         self.context_schema = context_schema
 
     @dataclass(**_DC_KWARGS)
@@ -535,6 +556,7 @@ class entrypoint(Generic[ContextT]):
                     bound=bound,
                     triggers=[START],
                     channels=START,
+                    timeout=self.timeout,
                     writers=[
                         ChannelWrite(
                             [

libs/langgraph/langgraph/graph/_node.py

@@ -90,3 +90,4 @@ class StateNodeSpec(Generic[NodeInputT, ContextT]):
     cache_policy: CachePolicy | None
     ends: tuple[str, ...] | dict[str, str] | None = EMPTY_SEQ
     defer: bool = False
+    timeout: float | None = None

libs/langgraph/langgraph/graph/state.py

@@ -7,6 +7,7 @@ import warnings
 from collections import defaultdict
 from collections.abc import Awaitable, Callable, Hashable, Sequence
 from dataclasses import is_dataclass
+from datetime import timedelta
 from functools import partial
 from inspect import isclass, isfunction, ismethod, signature
 from types import FunctionType
@@ -45,6 +46,7 @@ from langgraph._internal._fields import (
 )
 from langgraph._internal._pydantic import create_model
 from langgraph._internal._runnable import coerce_to_runnable
+from langgraph._internal._timeout import coerce_timeout
 from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.binop import BinaryOperatorAggregate
@@ -300,6 +302,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         destinations: dict[str, str] | tuple[str, ...] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> Self:
         """Add a new node to the `StateGraph`, input schema is inferred as the state schema.
@@ -367,6 +370,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         destinations: dict[str, str] | tuple[str, ...] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> Self:
         """Add a new node to the `StateGraph` where input schema is specified.
@@ -439,6 +443,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         destinations: dict[str, str] | tuple[str, ...] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> Self:
         """Add a new node to the `StateGraph`, input schema is inferred as the state schema.
@@ -506,6 +511,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         destinations: dict[str, str] | tuple[str, ...] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> Self:
         """Add a new node to the `StateGraph`, input schema is specified.
@@ -580,6 +586,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         destinations: dict[str, str] | tuple[str, ...] | None = None,
+        timeout: float | timedelta | None = None,
         **kwargs: Unpack[DeprecatedKwargs],
     ) -> Self:
         """Add a new node to the `StateGraph`.
@@ -609,6 +616,12 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 !!! warning
 
                     This is only used for graph rendering and doesn't have any effect on the graph execution.
+            timeout: Maximum wall-clock duration for a single invocation of this
+                node, in seconds (or as a `timedelta`). When exceeded, a
+                [`NodeTimeoutError`][langgraph.errors.NodeTimeoutError] is raised
+                and the retry policy (if any) decides whether to retry. Timeouts
+                are supported only for async nodes; sync nodes cannot be safely
+                cancelled in-process.
 
         Example:
             ```python
@@ -662,6 +675,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             )
             if input_schema is None:
                 input_schema = cast(type[NodeInputT] | None, input_)
+        timeout = coerce_timeout(timeout)
 
         if not isinstance(node, str):
             action = node
@@ -757,6 +771,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 cache_policy=cache_policy,
                 ends=ends,
                 defer=defer,
+                timeout=timeout,
             )
         elif inferred_input_schema is not None:
             self.nodes[node] = StateNodeSpec(
@@ -767,6 +782,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 cache_policy=cache_policy,
                 ends=ends,
                 defer=defer,
+                timeout=timeout,
             )
         else:
             self.nodes[node] = StateNodeSpec[StateT, ContextT](
@@ -777,6 +793,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 cache_policy=cache_policy,
                 ends=ends,
                 defer=defer,
+                timeout=timeout,
             )
 
         input_schema = input_schema or inferred_input_schema
@@ -1332,6 +1349,7 @@ class CompiledStateGraph(
                 retry_policy=node.retry_policy,
                 cache_policy=node.cache_policy,
                 bound=node.runnable,  # type: ignore[arg-type]
+                timeout=node.timeout,
             )
         else:
             raise RuntimeError

libs/langgraph/langgraph/pregel/_algo.py

@@ -7,6 +7,7 @@ import threading
 from collections import defaultdict, deque
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from copy import copy
+from datetime import timedelta
 from functools import partial
 from hashlib import sha1
 from typing import (
@@ -61,6 +62,7 @@ from langgraph._internal._constants import (
     TASKS,
 )
 from langgraph._internal._scratchpad import PregelScratchpad
+from langgraph._internal._timeout import coerce_timeout
 from langgraph._internal._typing import EMPTY_SEQ, MISSING
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.topic import Topic
@@ -114,13 +116,21 @@ class PregelTaskWrites(NamedTuple):
 
 
 class Call:
-    __slots__ = ("func", "input", "retry_policy", "cache_policy", "callbacks")
+    __slots__ = (
+        "func",
+        "input",
+        "retry_policy",
+        "cache_policy",
+        "callbacks",
+        "timeout",
+    )
 
     func: Callable
     input: tuple[tuple[Any, ...], dict[str, Any]]
     retry_policy: Sequence[RetryPolicy] | None
     cache_policy: CachePolicy | None
     callbacks: Callbacks
+    timeout: float | None
 
     def __init__(
         self,
@@ -130,12 +140,14 @@ class Call:
         retry_policy: Sequence[RetryPolicy] | None,
         cache_policy: CachePolicy | None,
         callbacks: Callbacks,
+        timeout: float | timedelta | None = None,
     ) -> None:
         self.func = func
         self.input = input
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
         self.callbacks = callbacks
+        self.timeout = coerce_timeout(timeout)
 
 
 def should_interrupt(
@@ -733,6 +745,7 @@ def prepare_single_task(
                         task_path[:3],
                         writers=proc.flat_writers,
                         subgraphs=proc.subgraphs,
+                        timeout=proc.timeout,
                     )
             else:
                 return PregelTask(task_id, name, task_path[:3])
@@ -870,6 +883,7 @@ def prepare_push_task_functional(
             cache_key,
             task_id,
             in_progress_task_path,
+            timeout=call.timeout,
         )
     else:
         return PregelTask(task_id, name, in_progress_task_path)
@@ -1041,6 +1055,7 @@ def prepare_push_task_send(
             translated_task_path,
             writers=proc.flat_writers,
             subgraphs=proc.subgraphs,
+            timeout=proc.timeout,
         )
     else:
         return PregelTask(task_id, packet.node, translated_task_path)

libs/langgraph/langgraph/pregel/_call.py

@@ -8,6 +8,7 @@ import inspect
 import sys
 import types
 from collections.abc import Awaitable, Callable, Generator, Sequence
+from datetime import timedelta
 from typing import Any, Generic, TypeVar, cast
 
 from langchain_core.runnables import Runnable
@@ -20,6 +21,7 @@ from langgraph._internal._runnable import (
     is_async_callable,
     run_in_executor,
 )
+from langgraph._internal._timeout import coerce_timeout, sync_timeout_unsupported
 from langgraph.config import get_config
 from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.types import CachePolicy, RetryPolicy
@@ -255,8 +257,13 @@ def call(
     *args: Any,
     retry_policy: Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy | None = None,
+    timeout: float | timedelta | None = None,
     **kwargs: Any,
 ) -> SyncAsyncFuture[T]:
+    timeout_s = coerce_timeout(timeout)
+    if timeout_s is not None and not is_async_callable(func):
+        name = getattr(func, "__name__", func.__class__.__name__)
+        raise sync_timeout_unsupported(name, kind="Task")
     config = get_config()
     impl = config[CONF][CONFIG_KEY_CALL]
     fut = impl(
@@ -265,5 +272,6 @@ def call(
         retry_policy=retry_policy,
         cache_policy=cache_policy,
         callbacks=config["callbacks"],
+        timeout=timeout_s,
     )
     return fut

libs/langgraph/langgraph/pregel/_loop.py

@@ -831,8 +831,18 @@ class PregelLoop:
             # parent. For forks (source=update/fork), use the fork's parent
             # checkpoint ID since the fork was created after the subgraph's
             # checkpoints from the original execution.
+            #
+            # Only gate on is_time_traveling (not is_replaying). When the
+            # client resumes with an explicit checkpoint_id that happens to
+            # point at the current head (e.g. LangGraph Studio sending
+            # `checkpoint: {checkpoint_id}` alongside Command(resume=...)),
+            # is_replaying is True but is_time_traveling is False. In that
+            # case subgraphs should load their latest checkpoint normally,
+            # not go through ReplayState's before-bound lookup which would
+            # miss subgraph checkpoints created during processing of the
+            # current parent step.
             replay_state: ReplayState | None = None
-            if self.is_replaying:
+            if is_time_traveling:
                 replay_checkpoint_id = self.checkpoint["id"]
                 if (
                     self.checkpoint_metadata.get("source")

libs/langgraph/langgraph/pregel/_read.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
+from datetime import timedelta
 from functools import cached_property
 from typing import (
     Any,
@@ -11,6 +12,7 @@ from langchain_core.runnables import Runnable, RunnableConfig
 from langgraph._internal._config import merge_configs
 from langgraph._internal._constants import CONF, CONFIG_KEY_READ
 from langgraph._internal._runnable import RunnableCallable, RunnableSeq
+from langgraph._internal._timeout import coerce_timeout
 from langgraph.pregel._utils import find_subgraph_pregel
 from langgraph.pregel._write import ChannelWrite
 from langgraph.pregel.protocol import PregelProtocol
@@ -123,6 +125,11 @@ class PregelNode:
     cache_policy: CachePolicy | None
     """The cache policy to use when invoking the node."""
 
+    timeout: float | None
+    """Maximum time in seconds allowed for a single invocation of this node.
+    If exceeded, `NodeTimeoutError` is raised and the retry policy (if any)
+    decides whether to retry. Supported only for async nodes."""
+
     tags: Sequence[str] | None
     """Tags to attach to the node for tracing."""
 
@@ -145,6 +152,7 @@ class PregelNode:
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
         subgraphs: Sequence[PregelProtocol] | None = None,
+        timeout: float | timedelta | None = None,
     ) -> None:
         self.channels = channels
         self.triggers = list(triggers)
@@ -156,6 +164,7 @@ class PregelNode:
             self.retry_policy = (retry_policy,)
         else:
             self.retry_policy = retry_policy
+        self.timeout = coerce_timeout(timeout)
         self.tags = tags
         self.metadata = metadata
         if subgraphs is not None:

libs/langgraph/langgraph/pregel/_retry.py

@@ -4,12 +4,16 @@ import asyncio
 import logging
 import random
 import sys
+import threading
 import time
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Coroutine, Sequence
+from contextlib import suppress
 from dataclasses import replace
-from typing import Any
+from datetime import datetime, timedelta, timezone
+from typing import Any, Literal
 
 from langchain_core.runnables import RunnableConfig
+from typing_extensions import NotRequired, TypedDict
 
 from langgraph._internal._config import patch_configurable, recast_checkpoint_ns
 from langgraph._internal._constants import (
@@ -18,11 +22,14 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_RUNTIME,
+    CONFIG_KEY_SEND,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_THREAD_ID,
+    CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     NS_SEP,
 )
-from langgraph.errors import GraphBubbleUp, ParentCommand
+from langgraph._internal._timeout import sync_timeout_unsupported
+from langgraph.errors import GraphBubbleUp, NodeTimeoutError, ParentCommand
 from langgraph.runtime import ExecutionInfo, Runtime
 from langgraph.types import Command, PregelExecutableTask, RetryPolicy
 
@@ -30,6 +37,182 @@ logger = logging.getLogger(__name__)
 SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
 
 
+class _TimedAttemptPayload(TypedDict):
+    execution_id: str
+    task_id: str
+    task_name: str
+    attempt: int
+    run_id: str | None
+    thread_id: str | None
+    checkpoint_ns: str | None
+    started_at: datetime
+    deadline_at: datetime
+    timeout_secs: float
+    event: Literal["start", "finish"]
+    finished_at: NotRequired[datetime]
+    status: NotRequired[Literal["success", "error"]]
+    error_type: NotRequired[str | None]
+    error_message: NotRequired[str | None]
+
+
+class _TimedAttemptScope:
+    """Guarded-config window for timed attempts.
+
+    `close()` and the guarded send are serialized so writes from a cancelled
+    background task cannot slip past the timeout boundary.
+    """
+
+    __slots__ = ("_active", "_lock")
+
+    def __init__(self) -> None:
+        self._active = True
+        self._lock = threading.Lock()
+
+    def wrap_config(self, config: RunnableConfig) -> RunnableConfig:
+        configurable = config.get(CONF, {})
+        if (send := configurable.get(CONFIG_KEY_SEND)) is not None:
+            return patch_configurable(config, {CONFIG_KEY_SEND: self._guard_send(send)})
+        return config
+
+    def close(self) -> None:
+        with self._lock:
+            self._active = False
+
+    def _guard_send(
+        self, send: Callable[[Sequence[tuple[str, Any]]], None]
+    ) -> Callable[[Sequence[tuple[str, Any]]], None]:
+        def guarded_send(writes: Sequence[tuple[str, Any]]) -> None:
+            with self._lock:
+                if self._active:
+                    send(writes)
+
+        return guarded_send
+
+
+def _drain_cancelled(task: asyncio.Task[Any]) -> None:
+    # Mark the abandoned task's exception as retrieved so asyncio doesn't log it.
+    with suppress(asyncio.CancelledError):
+        task.exception()
+
+
+def _create_task_with_config_context(
+    run: Callable[[], Coroutine[Any, Any, Any]], config: RunnableConfig
+) -> asyncio.Task[Any]:
+    from langgraph._internal._runnable import set_config_context
+
+    with set_config_context(config) as context:
+        return context.run(lambda: asyncio.create_task(run()))
+
+
+def _start_timed_attempt(
+    task: PregelExecutableTask, config: RunnableConfig, timeout_s: float
+) -> _TimedAttemptPayload | None:
+    configurable = config.get(CONF, {})
+    callback = configurable.get(CONFIG_KEY_TIMED_ATTEMPT_OBSERVER)
+    if callback is None:
+        return None
+    runtime = configurable.get(CONFIG_KEY_RUNTIME)
+    execution_info = runtime.execution_info if isinstance(runtime, Runtime) else None
+    attempt = execution_info.node_attempt if execution_info is not None else 1
+    run_id = execution_info.run_id if execution_info is not None else None
+    thread_id = (
+        execution_info.thread_id
+        if execution_info is not None
+        else configurable.get(CONFIG_KEY_THREAD_ID)
+    )
+    checkpoint_ns = (
+        execution_info.checkpoint_ns
+        if execution_info is not None
+        else configurable.get(CONFIG_KEY_CHECKPOINT_NS)
+    )
+    started_at = datetime.now(timezone.utc)
+    payload: _TimedAttemptPayload = {
+        "execution_id": f"run:{run_id or '-'}|task:{task.id}|attempt:{attempt}",
+        "task_id": task.id,
+        "task_name": task.name,
+        "attempt": attempt,
+        "run_id": run_id,
+        "thread_id": thread_id,
+        "checkpoint_ns": checkpoint_ns,
+        "started_at": started_at,
+        "deadline_at": started_at + timedelta(seconds=timeout_s),
+        "timeout_secs": timeout_s,
+        "event": "start",
+    }
+    _dispatch_observer(callback, payload)
+    return payload
+
+
+def _finish_timed_attempt(
+    config: RunnableConfig,
+    payload: _TimedAttemptPayload | None,
+    error: BaseException | None = None,
+) -> None:
+    if payload is None:
+        return
+    callback = config.get(CONF, {}).get(CONFIG_KEY_TIMED_ATTEMPT_OBSERVER)
+    if callback is None:
+        return
+    finish: _TimedAttemptPayload = {
+        **payload,
+        "event": "finish",
+        "finished_at": datetime.now(timezone.utc),
+        "status": "error" if error is not None else "success",
+        "error_type": type(error).__name__ if error is not None else None,
+        "error_message": str(error) if error is not None else None,
+    }
+    _dispatch_observer(callback, finish)
+
+
+def _dispatch_observer(
+    callback: Callable[[_TimedAttemptPayload], None], payload: _TimedAttemptPayload
+) -> None:
+    try:
+        callback(payload)
+    except Exception:
+        logger.warning("Timed attempt observer failed", exc_info=True)
+
+
+async def _arun_with_timeout(
+    task: PregelExecutableTask,
+    config: RunnableConfig,
+    timeout_s: float,
+    *,
+    stream: bool,
+) -> Any:
+    scope = _TimedAttemptScope()
+    scoped_config = scope.wrap_config(config)
+    start = time.monotonic()
+    if stream:
+
+        async def run() -> Any:
+            async for _ in task.proc.astream(task.input, scoped_config):
+                pass
+
+    else:
+
+        async def run() -> Any:
+            return await task.proc.ainvoke(task.input, scoped_config)
+
+    bg = _create_task_with_config_context(run, scoped_config)
+    try:
+        return await asyncio.wait_for(asyncio.shield(bg), timeout=timeout_s)
+    except asyncio.TimeoutError as exc:
+        elapsed = time.monotonic() - start
+        scope.close()
+        task.writes.clear()
+        bg.cancel()
+        bg.add_done_callback(_drain_cancelled)
+        raise NodeTimeoutError(task.name, timeout_s, elapsed) from exc
+    except asyncio.CancelledError:
+        scope.close()
+        bg.cancel()
+        bg.add_done_callback(_drain_cancelled)
+        raise
+    finally:
+        scope.close()
+
+
 def _ensure_execution_info(
     runtime: Runtime, config: RunnableConfig, task: PregelExecutableTask
 ) -> Runtime:
@@ -90,6 +273,11 @@ def run_with_retry(
 ) -> None:
     """Run a task with retries."""
     retry_policy = task.retry_policy or retry_policy
+    if task.timeout is not None:
+        # `validate_timeout_supported` catches sync nodes at compile time;
+        # this is a runtime safety net for paths (e.g. distributed runtime)
+        # that may bypass that validation.
+        raise sync_timeout_unsupported(task.name)
     attempts = 0
     node_first_attempt_time = time.time()
     config = task.config
@@ -195,6 +383,7 @@ async def arun_with_retry(
 ) -> None:
     """Run a task asynchronously with retries."""
     retry_policy = task.retry_policy or retry_policy
+    timeout_s = task.timeout
     attempts = 0
     node_first_attempt_time = time.time()
     config = task.config
@@ -229,35 +418,47 @@ async def arun_with_retry(
                     )
                 },
             )
+        attempt_payload = (
+            _start_timed_attempt(task, config, timeout_s)
+            if timeout_s is not None
+            else None
+        )
         try:
-            # clear any writes from previous attempts
             task.writes.clear()
-            # run the task
-            if stream:
-                async for _ in task.proc.astream(task.input, config):
-                    pass
-                # if successful, end
-                break
-            else:
+            if timeout_s is None:
+                if stream:
+                    async for _ in task.proc.astream(task.input, config):
+                        pass
+                    break
                 return await task.proc.ainvoke(task.input, config)
+            result = await _arun_with_timeout(task, config, timeout_s, stream=stream)
+            _finish_timed_attempt(config, attempt_payload)
+            if stream:
+                break
+            return result
         except ParentCommand as exc:
             ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
             cmd = exc.args[0]
             # strip task_ids from namespace for comparison (ns format: "node1|node2:task_id")
             if cmd.graph in (ns, recast_checkpoint_ns(ns), task.name):
-                # this command is for the current graph, handle it
-                for w in task.writers:
-                    w.invoke(cmd, config)
+                try:
+                    for w in task.writers:
+                        w.invoke(cmd, config)
+                except Exception as writer_exc:
+                    _finish_timed_attempt(config, attempt_payload, writer_exc)
+                    raise
+                _finish_timed_attempt(config, attempt_payload)
                 break
             elif cmd.graph == Command.PARENT:
                 # this command is for the parent graph, assign it to the parent.
                 exc.args = (replace(cmd, graph=_checkpoint_ns_for_parent_command(ns)),)
-            # bubble up
+            _finish_timed_attempt(config, attempt_payload)
             raise
         except GraphBubbleUp:
-            # if interrupted, end
+            _finish_timed_attempt(config, attempt_payload)
             raise
         except Exception as exc:
+            _finish_timed_attempt(config, attempt_payload, exc)
             if SUPPORTS_EXC_NOTES:
                 exc.add_note(f"During task with name '{task.name}' and id '{task.id}'")
             if not retry_policy:

libs/langgraph/langgraph/pregel/_runner.py

@@ -14,6 +14,7 @@ from collections.abc import (
     Iterator,
     Sequence,
 )
+from datetime import timedelta
 from functools import partial
 from typing import (
     Any,
@@ -537,6 +538,7 @@ def _call(
     *,
     retry_policy: Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy | None = None,
+    timeout: float | timedelta | None = None,
     callbacks: Callbacks = None,
     futures: weakref.ref[FuturesDict],
     schedule_task: Callable[
@@ -560,6 +562,7 @@ def _call(
             retry_policy=retry_policy,
             cache_policy=cache_policy,
             callbacks=callbacks,
+            timeout=timeout,
         ),
     ):
         if fut := next(
@@ -624,6 +627,7 @@ def _acall(
     *,
     retry_policy: Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy | None = None,
+    timeout: float | timedelta | None = None,
     callbacks: Callbacks = None,
     # injected dependencies
     futures: weakref.ref[FuturesDict],
@@ -657,6 +661,7 @@ def _acall(
             input,
             retry_policy=retry_policy,
             cache_policy=cache_policy,
+            timeout=timeout,
             callbacks=callbacks,
             futures=futures,
             schedule_task=schedule_task,
@@ -678,6 +683,7 @@ async def _acall_impl(
     *,
     retry_policy: Sequence[RetryPolicy] | None = None,
     cache_policy: CachePolicy | None = None,
+    timeout: float | timedelta | None = None,
     callbacks: Callbacks = None,
     # injected dependencies
     futures: weakref.ref[FuturesDict[asyncio.Future, asyncio.Event]],
@@ -703,6 +709,7 @@ async def _acall_impl(
                 retry_policy=retry_policy,
                 cache_policy=cache_policy,
                 callbacks=callbacks,
+                timeout=timeout,
             ),
         ):
             if fut := next(

libs/langgraph/langgraph/pregel/_utils.py

@@ -4,16 +4,21 @@ import ast
 import inspect
 import re
 import textwrap
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
+from functools import partial
 from typing import Any
 
 from langchain_core.runnables import Runnable, RunnableLambda, RunnableSequence
+from langchain_core.runnables.config import run_in_executor
 from langgraph.checkpoint.base import ChannelVersions
 from typing_extensions import override
 
 from langgraph._internal._runnable import RunnableCallable, RunnableSeq
+from langgraph._internal._timeout import sync_timeout_unsupported
 from langgraph.pregel.protocol import PregelProtocol
 
+_SEQUENCE_TYPES = (RunnableSeq, RunnableSequence)
+
 
 def get_new_channel_versions(
     previous_versions: ChannelVersions, current_versions: ChannelVersions
@@ -64,6 +69,47 @@ def find_subgraph_pregel(candidate: Runnable) -> PregelProtocol | None:
     return None
 
 
+def _sequence_steps(runnable: Runnable) -> Sequence[Runnable] | None:
+    if isinstance(runnable, _SEQUENCE_TYPES):
+        return runnable.steps
+    return None
+
+
+def _has_method_override(runnable: Runnable, method_name: str) -> bool:
+    method = getattr(type(runnable), method_name, None)
+    return method is not None and method is not getattr(Runnable, method_name)
+
+
+def _is_executor_backed_afunc(afunc: Callable[..., Any] | None) -> bool:
+    return isinstance(afunc, partial) and afunc.func is run_in_executor
+
+
+def _has_native_async(runnable: Runnable) -> bool:
+    if isinstance(runnable, RunnableCallable):
+        return runnable.afunc is not None and not _is_executor_backed_afunc(
+            runnable.afunc
+        )
+    if isinstance(runnable, RunnableLambda):
+        return bool(getattr(runnable, "afunc", False))
+    return _has_method_override(runnable, "ainvoke")
+
+
+def _runnable_has_native_async(runnable: Runnable) -> bool:
+    """Return whether a runnable can be timed without running sync code."""
+
+    if (steps := _sequence_steps(runnable)) is not None:
+        for step in steps:
+            if not _runnable_has_native_async(step):
+                return False
+        return True
+    return _has_native_async(runnable)
+
+
+def validate_timeout_supported(runnable: Runnable, *, name: str) -> None:
+    if not _runnable_has_native_async(runnable):
+        raise sync_timeout_unsupported(name)
+
+
 def get_function_nonlocals(func: Callable) -> list[Any]:
     """Get the nonlocal variables accessed by a function.
 

libs/langgraph/langgraph/pregel/main.py

@@ -17,6 +17,7 @@ from collections.abc import (
     Sequence,
 )
 from dataclasses import is_dataclass, replace
+from datetime import timedelta
 from functools import partial
 from inspect import isclass
 from typing import (
@@ -95,6 +96,7 @@ from langgraph._internal._runnable import (
     RunnableSeq,
     coerce_to_runnable,
 )
+from langgraph._internal._timeout import coerce_timeout
 from langgraph._internal._typing import MISSING, DeprecatedKwargs
 from langgraph.callbacks import (
     GraphInterruptEvent,
@@ -137,7 +139,7 @@ from langgraph.pregel._messages import StreamMessagesHandler
 from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
 from langgraph.pregel._retry import RetryPolicy
 from langgraph.pregel._runner import PregelRunner
-from langgraph.pregel._utils import get_new_channel_versions
+from langgraph.pregel._utils import get_new_channel_versions, validate_timeout_supported
 from langgraph.pregel._validate import validate_graph, validate_keys
 from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.pregel.debug import get_bolded_text, get_colored_text, tasks_w_writes
@@ -186,6 +188,7 @@ class NodeBuilder:
         "_bound",
         "_retry_policy",
         "_cache_policy",
+        "_timeout",
     )
 
     _channels: str | list[str]
@@ -196,6 +199,7 @@ class NodeBuilder:
     _bound: Runnable
     _retry_policy: list[RetryPolicy]
     _cache_policy: CachePolicy | None
+    _timeout: float | None
 
     def __init__(
         self,
@@ -208,6 +212,7 @@ class NodeBuilder:
         self._bound = DEFAULT_BOUND
         self._retry_policy = []
         self._cache_policy = None
+        self._timeout = None
 
     def subscribe_only(
         self,
@@ -326,6 +331,11 @@ class NodeBuilder:
         self._cache_policy = policy
         return self
 
+    def set_timeout(self, timeout: float | timedelta | None) -> Self:
+        """Set the per-attempt timeout for this node."""
+        self._timeout = coerce_timeout(timeout)
+        return self
+
     def build(self) -> PregelNode:
         """Builds the node."""
         return PregelNode(
@@ -337,6 +347,7 @@ class NodeBuilder:
             bound=self._bound,
             retry_policy=self._retry_policy,
             cache_policy=self._cache_policy,
+            timeout=self._timeout,
         )
 
 
@@ -817,6 +828,9 @@ class Pregel(
         )
 
     def validate(self) -> Self:
+        for name, node in self.nodes.items():
+            if node.timeout is not None:
+                validate_timeout_supported(node.bound, name=name)
         validate_graph(
             self.nodes,
             {k: v for k, v in self.channels.items() if isinstance(v, BaseChannel)},

libs/langgraph/langgraph/types.py

@@ -548,6 +548,7 @@ class PregelExecutableTask:
     path: tuple[str | int | tuple, ...]
     writers: Sequence[Runnable] = ()
     subgraphs: Sequence[PregelProtocol] = ()
+    timeout: float | None = None
 
 
 class StateSnapshot(NamedTuple):

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_retry.py

@@ -1,7 +1,12 @@
+import asyncio
+import threading
+import time
 from collections import deque
+from datetime import datetime, timedelta
 from unittest.mock import Mock, patch
 
 import pytest
+from langchain_core.runnables import RunnableLambda
 from langgraph.checkpoint.memory import MemorySaver
 from typing_extensions import TypedDict
 
@@ -10,18 +15,29 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_RUNTIME,
+    CONFIG_KEY_SEND,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_THREAD_ID,
+    CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
 )
-from langgraph.graph import START, StateGraph
+from langgraph._internal._runnable import RunnableCallable
+from langgraph._internal._timeout import coerce_timeout
+from langgraph.channels.ephemeral_value import EphemeralValue
+from langgraph.channels.last_value import LastValue
+from langgraph.errors import GraphInterrupt, NodeTimeoutError, ParentCommand
+from langgraph.func import entrypoint, task
+from langgraph.graph import END, START, StateGraph
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.pregel._read import PregelNode
 from langgraph.pregel._retry import (
     _checkpoint_ns_for_parent_command,
     _ensure_execution_info,
     _should_retry_on,
+    arun_with_retry,
     run_with_retry,
 )
 from langgraph.runtime import DEFAULT_RUNTIME, ExecutionInfo, Runtime
-from langgraph.types import PregelExecutableTask, RetryPolicy
+from langgraph.types import Command, PregelExecutableTask, RetryPolicy
 
 
 def test_should_retry_on_single_exception():
@@ -567,3 +583,643 @@ def test_run_with_retry_creates_execution_info_when_missing():
     assert info.run_id == "run-abc"
     assert info.node_attempt == 1
     assert info.node_first_attempt_time is not None
+
+
+def _make_task(
+    proc, *, timeout=None, retry_policy=(), name="timed", task_id="tid", writers=()
+):
+    runtime = DEFAULT_RUNTIME.override(execution_info=None)
+    writes = deque()
+    config = {
+        "run_id": "run-x",
+        CONF: {
+            CONFIG_KEY_RUNTIME: runtime,
+            CONFIG_KEY_CHECKPOINT_ID: "cp",
+            CONFIG_KEY_CHECKPOINT_NS: f"{name}:{task_id}",
+            CONFIG_KEY_SEND: writes.extend,
+            CONFIG_KEY_TASK_ID: task_id,
+            CONFIG_KEY_THREAD_ID: "thr",
+        },
+    }
+    return PregelExecutableTask(
+        name=name,
+        input=None,
+        proc=proc,
+        writes=writes,
+        config=config,
+        triggers=[name],
+        retry_policy=retry_policy,
+        cache_key=None,
+        id=task_id,
+        path=("__pregel_pull", name),
+        writers=writers,
+        timeout=coerce_timeout(timeout),
+    )
+
+
+def test_coerce_timeout():
+    assert coerce_timeout(None) is None
+    assert coerce_timeout(1.5) == 1.5
+    assert coerce_timeout(2) == 2.0
+    assert coerce_timeout(timedelta(milliseconds=250)) == 0.25
+    with pytest.raises(ValueError, match="greater than 0"):
+        coerce_timeout(0)
+    with pytest.raises(ValueError, match="greater than 0"):
+        coerce_timeout(timedelta())
+
+
+def test_run_with_retry_rejects_sync_timeout_without_starting_proc():
+    started = False
+
+    class Proc:
+        def invoke(self, input, config):
+            nonlocal started
+            started = True
+            return input
+
+    task = _make_task(Proc(), timeout=0.05, name="sync")
+
+    with pytest.raises(ValueError, match="only supported for async nodes"):
+        run_with_retry(task, retry_policy=None)
+    assert not started
+
+
+def test_run_with_retry_without_timeout_runs_sync_directly():
+    class FastProc:
+        def invoke(self, input, config):
+            return "ok"
+
+    task = _make_task(FastProc(), timeout=None)
+    assert run_with_retry(task, retry_policy=None) == "ok"
+
+
+def test_arun_with_retry_timeout_ok_when_fast():
+    class FastProc:
+        async def ainvoke(self, input, config):
+            return "ok"
+
+    task = _make_task(FastProc(), timeout=1.0)
+
+    async def _run() -> None:
+        assert await arun_with_retry(task, retry_policy=None) == "ok"
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_retries_when_retry_on_timeout():
+    calls: list[float] = []
+
+    class FlakyProc:
+        async def ainvoke(self, input, config):
+            calls.append(time.monotonic())
+            if len(calls) < 2:
+                await asyncio.sleep(0.5)
+                return "late"
+            return "ok"
+
+    policy = RetryPolicy(
+        max_attempts=3,
+        initial_interval=0.0,
+        jitter=False,
+        retry_on=NodeTimeoutError,
+    )
+    task = _make_task(FlakyProc(), timeout=0.05, retry_policy=(policy,))
+
+    async def _run() -> None:
+        assert await arun_with_retry(task, retry_policy=None) == "ok"
+        assert len(calls) == 2
+
+    asyncio.run(_run())
+
+
+def test_entrypoint_timeout_allows_pre_timeout_child_task_to_run():
+    child_started = threading.Event()
+
+    @task()
+    def child(value: int) -> int:
+        child_started.set()
+        return value + 1
+
+    @entrypoint(timeout=0.05)
+    async def parent(value: int) -> int:
+        child(value)
+        await asyncio.sleep(0.2)
+        return value
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await parent.ainvoke(1)
+
+    asyncio.run(_run())
+    assert child_started.wait(timeout=1.0)
+
+
+def test_arun_with_retry_timeout_accepts_timedelta():
+    class SlowProc:
+        async def ainvoke(self, input, config):
+            await asyncio.sleep(0.5)
+            return input
+
+    task = _make_task(SlowProc(), timeout=timedelta(milliseconds=50))
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await arun_with_retry(task, retry_policy=None)
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_fires_async():
+    class SlowProc:
+        async def ainvoke(self, input, config):
+            await asyncio.sleep(1.0)
+            return input
+
+    task = _make_task(SlowProc(), timeout=0.05, name="aslow")
+
+    async def _run():
+        with pytest.raises(NodeTimeoutError) as excinfo:
+            await arun_with_retry(task, retry_policy=None)
+        assert excinfo.value.node == "aslow"
+        assert excinfo.value.timeout == 0.05
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_discards_stale_executor_writes():
+    release_first_attempt = threading.Event()
+
+    class FlakyAsyncProc:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def ainvoke(self, input, config):
+            self.calls += 1
+            if self.calls == 1:
+
+                def late_write() -> str:
+                    release_first_attempt.wait(timeout=1.0)
+                    config[CONF][CONFIG_KEY_SEND]([("value", "stale")])
+                    return "late"
+
+                return await asyncio.to_thread(late_write)
+            release_first_attempt.set()
+            config[CONF][CONFIG_KEY_SEND]([("value", "fresh")])
+            return "ok"
+
+    policy = RetryPolicy(
+        max_attempts=2,
+        initial_interval=0.0,
+        jitter=False,
+        retry_on=NodeTimeoutError,
+    )
+    task = _make_task(FlakyAsyncProc(), timeout=0.05, retry_policy=(policy,))
+
+    async def _run() -> None:
+        assert await arun_with_retry(task, retry_policy=None) == "ok"
+        await asyncio.sleep(0.05)
+        assert task.writes == deque([("value", "fresh")])
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_discards_pre_timeout_writes():
+    class SlowAsyncWriterProc:
+        async def ainvoke(self, input, config):
+            config[CONF][CONFIG_KEY_SEND]([("value", "stale-before-timeout")])
+            await asyncio.sleep(0.2)
+            return "late"
+
+    task = _make_task(SlowAsyncWriterProc(), timeout=0.05, name="aslow-writer")
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await arun_with_retry(task, retry_policy=None)
+        assert task.writes == deque()
+
+    asyncio.run(_run())
+
+
+def test_astream_with_retry_timeout_discards_pre_timeout_writes():
+    class SlowStreamWriterProc:
+        async def astream(self, input, config):
+            config[CONF][CONFIG_KEY_SEND]([("value", "stale-before-timeout")])
+            await asyncio.sleep(0.2)
+            if False:
+                yield None
+
+    task = _make_task(SlowStreamWriterProc(), timeout=0.05, name="astream-writer")
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await arun_with_retry(task, retry_policy=None, stream=True)
+        assert task.writes == deque()
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_cannot_be_swallowed():
+    class StubbornProc:
+        async def ainvoke(self, input, config):
+            try:
+                await asyncio.sleep(1.0)
+            except asyncio.CancelledError:
+                config[CONF][CONFIG_KEY_SEND]([("value", "stale")])
+                await asyncio.sleep(0)
+                return "late"
+            return "ok"
+
+    task = _make_task(StubbornProc(), timeout=0.05, name="stubborn")
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError) as excinfo:
+            await arun_with_retry(task, retry_policy=None)
+        assert excinfo.value.node == "stubborn"
+        await asyncio.sleep(0.05)
+        assert task.writes == deque()
+
+    asyncio.run(_run())
+
+
+def test_astream_with_retry_timeout_cannot_be_swallowed():
+    class StubbornStreamProc:
+        async def astream(self, input, config):
+            try:
+                await asyncio.sleep(1.0)
+            except asyncio.CancelledError:
+                config[CONF][CONFIG_KEY_SEND]([("value", "stale")])
+                await asyncio.sleep(0)
+                if False:
+                    yield None
+                return
+            yield "ok"
+
+    task = _make_task(StubbornStreamProc(), timeout=0.05, name="stubborn-stream")
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError) as excinfo:
+            await arun_with_retry(task, retry_policy=None, stream=True)
+        assert excinfo.value.node == "stubborn-stream"
+        await asyncio.sleep(0.05)
+        assert task.writes == deque()
+
+    asyncio.run(_run())
+
+
+class _TimeoutState(TypedDict):
+    x: int
+
+
+def test_timeout_validation_is_eager_across_apis():
+    with pytest.raises(ValueError, match="greater than 0"):
+        task(timeout=0)
+
+    with pytest.raises(ValueError, match="greater than 0"):
+        entrypoint(timeout=0)
+
+    with pytest.raises(ValueError, match="greater than 0"):
+        NodeBuilder().set_timeout(0)
+
+    with pytest.raises(ValueError, match="greater than 0"):
+        PregelNode(channels="x", triggers=["x"], timeout=0)
+
+    builder = StateGraph(_TimeoutState)
+    with pytest.raises(ValueError, match="greater than 0"):
+        builder.add_node("slow", lambda state: state, timeout=0)
+
+
+def test_timeout_rejects_sync_functional_apis_at_declaration_time():
+    with pytest.raises(ValueError, match="only supported for async nodes"):
+
+        @task(timeout=0.05)
+        def sync_task(value: int) -> int:
+            return value
+
+    with pytest.raises(ValueError, match="only supported for async nodes"):
+
+        @entrypoint(timeout=0.05)
+        def sync_entrypoint(value: int) -> int:
+            return value
+
+
+def test_state_graph_compile_rejects_sync_node_timeout():
+    def slow(state: _TimeoutState) -> _TimeoutState:
+        return {"x": state["x"] + 1}
+
+    builder = StateGraph(_TimeoutState)
+    builder.add_node("slow", slow, timeout=0.05)
+    builder.add_edge(START, "slow")
+    builder.add_edge("slow", END)
+
+    with pytest.raises(ValueError, match="only supported for async nodes"):
+        builder.compile()
+
+
+def test_pregel_validate_rejects_sync_node_timeout():
+    def slow(value: int) -> int:
+        return value + 1
+
+    with pytest.raises(ValueError, match="only supported for async nodes"):
+        Pregel(
+            nodes={
+                "slow": (
+                    NodeBuilder()
+                    .subscribe_only("input")
+                    .do(slow)
+                    .set_timeout(0.05)
+                    .write_to("output")
+                )
+            },
+            channels={
+                "input": EphemeralValue(int),
+                "output": LastValue(int),
+            },
+            input_channels="input",
+            output_channels="output",
+        )
+
+
+def test_pregel_validate_accepts_async_runnable_lambda_timeout():
+    async def slow(value: int) -> int:
+        await asyncio.sleep(0.2)
+        return value + 1
+
+    graph = Pregel(
+        nodes={
+            "slow": (
+                NodeBuilder()
+                .subscribe_only("input")
+                .do(RunnableLambda(slow))
+                .set_timeout(0.05)
+                .write_to("output")
+            )
+        },
+        channels={
+            "input": EphemeralValue(int),
+            "output": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
+    )
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await graph.ainvoke(1)
+
+    asyncio.run(_run())
+
+
+def test_pregel_validate_accepts_runnable_callable_with_sync_and_async_timeout():
+    def sync(value: int) -> int:
+        return value + 1
+
+    async def async_(value: int) -> int:
+        await asyncio.sleep(0.2)
+        return value + 1
+
+    graph = Pregel(
+        nodes={
+            "slow": (
+                NodeBuilder()
+                .subscribe_only("input")
+                .do(RunnableCallable(sync, async_))
+                .set_timeout(0.05)
+                .write_to("output")
+            )
+        },
+        channels={
+            "input": EphemeralValue(int),
+            "output": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
+    )
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await graph.ainvoke(1)
+
+    asyncio.run(_run())
+
+
+def test_state_graph_add_node_timeout_e2e():
+    async def slow(state: _TimeoutState) -> _TimeoutState:
+        await asyncio.sleep(1.0)
+        return {"x": state["x"] + 1}
+
+    builder = StateGraph(_TimeoutState)
+    builder.add_node("slow", slow, timeout=0.05)
+    builder.add_edge(START, "slow")
+    builder.add_edge("slow", END)
+    graph = builder.compile()
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await graph.ainvoke({"x": 1})
+
+    asyncio.run(_run())
+
+
+def test_state_graph_add_node_timeout_composes_with_retry():
+    """add_node(..., timeout=...) + retry_policy retries then succeeds."""
+
+    attempts: list[int] = []
+
+    async def flaky(state: _TimeoutState) -> _TimeoutState:
+        attempts.append(len(attempts))
+        if len(attempts) < 2:
+            await asyncio.sleep(0.5)
+        return {"x": state["x"] + 1}
+
+    builder = StateGraph(_TimeoutState)
+    builder.add_node(
+        "flaky",
+        flaky,
+        timeout=0.1,
+        retry_policy=RetryPolicy(
+            max_attempts=3,
+            initial_interval=0.0,
+            jitter=False,
+            retry_on=NodeTimeoutError,
+        ),
+    )
+    builder.add_edge(START, "flaky")
+    builder.add_edge("flaky", END)
+    graph = builder.compile()
+
+    async def _run() -> None:
+        result = await graph.ainvoke({"x": 0})
+        assert result == {"x": 1}
+        assert len(attempts) == 2
+
+    asyncio.run(_run())
+
+
+def test_task_decorator_timeout_e2e():
+    @task(timeout=0.05)
+    async def slow_task(x: int) -> int:
+        await asyncio.sleep(0.2)
+        return x + 1
+
+    @entrypoint()
+    async def workflow(x: int) -> int:
+        return await slow_task(x)
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await workflow.ainvoke(1)
+
+    asyncio.run(_run())
+
+
+def test_entrypoint_timeout_e2e():
+    @entrypoint(timeout=0.05)
+    async def slow_workflow(x: int) -> int:
+        await asyncio.sleep(0.2)
+        return x
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await slow_workflow.ainvoke(1)
+
+    asyncio.run(_run())
+
+
+def test_node_builder_timeout_e2e():
+    async def slow(value: int) -> int:
+        await asyncio.sleep(0.2)
+        return value + 1
+
+    graph = Pregel(
+        nodes={
+            "slow": (
+                NodeBuilder()
+                .subscribe_only("input")
+                .do(slow)
+                .set_timeout(0.05)
+                .write_to("output")
+            )
+        },
+        channels={
+            "input": EphemeralValue(int),
+            "output": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
+    )
+
+    async def _run() -> None:
+        with pytest.raises(NodeTimeoutError):
+            await graph.ainvoke(1)
+
+    asyncio.run(_run())
+
+
+def test_arun_with_retry_timeout_observer_tracks_attempts():
+    events: list[dict] = []
+
+    class FlakyProc:
+        async def ainvoke(self, input, config):
+            runtime = config[CONF][CONFIG_KEY_RUNTIME]
+            if runtime.execution_info.node_attempt == 1:
+                await asyncio.sleep(0.2)
+            return "ok"
+
+    policy = RetryPolicy(
+        max_attempts=2,
+        initial_interval=0.0,
+        jitter=False,
+        retry_on=NodeTimeoutError,
+    )
+    task = _make_task(FlakyProc(), timeout=0.05, retry_policy=(policy,), name="flaky")
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    async def _run() -> None:
+        assert await arun_with_retry(task, retry_policy=None) == "ok"
+
+    asyncio.run(_run())
+
+    starts = [payload for payload in events if payload["event"] == "start"]
+    finishes = [payload for payload in events if payload["event"] == "finish"]
+    assert [payload["attempt"] for payload in starts] == [1, 2]
+    assert [payload["attempt"] for payload in finishes] == [1, 2]
+    assert [payload["status"] for payload in finishes] == ["error", "success"]
+    assert starts[0]["execution_id"] != starts[1]["execution_id"]
+    assert starts[0]["timeout_secs"] == 0.05
+    assert starts[0]["task_name"] == "flaky"
+    assert isinstance(starts[0]["started_at"], datetime)
+    assert isinstance(starts[0]["deadline_at"], datetime)
+    assert isinstance(finishes[0]["finished_at"], datetime)
+    assert starts[0]["deadline_at"] > starts[0]["started_at"]
+
+
+def test_arun_with_retry_timeout_observer_treats_parent_command_as_non_error():
+    events: list[dict] = []
+
+    class ParentProc:
+        async def ainvoke(self, input, config):
+            raise ParentCommand(Command(graph=Command.PARENT))
+
+    task = _make_task(ParentProc(), timeout=0.05, name="parent")
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    async def _run() -> None:
+        with pytest.raises(ParentCommand):
+            await arun_with_retry(task, retry_policy=None)
+
+    asyncio.run(_run())
+
+    finish = next(payload for payload in events if payload["event"] == "finish")
+    assert finish["status"] == "success"
+    assert finish["error_type"] is None
+    assert finish["error_message"] is None
+
+
+def test_arun_with_retry_timeout_observer_finishes_when_parent_writer_errors():
+    events: list[dict] = []
+
+    class ParentProc:
+        async def ainvoke(self, input, config):
+            raise ParentCommand(Command(graph="parent", update={"value": "updated"}))
+
+    class FailingWriter:
+        def invoke(self, input, config):
+            raise ValueError("writer failed")
+
+    task = _make_task(
+        ParentProc(), timeout=0.05, name="parent", writers=(FailingWriter(),)
+    )
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    async def _run() -> None:
+        with pytest.raises(ValueError, match="writer failed"):
+            await arun_with_retry(task, retry_policy=None)
+
+    asyncio.run(_run())
+
+    finish = next(payload for payload in events if payload["event"] == "finish")
+    assert finish["status"] == "error"
+    assert finish["error_type"] == "ValueError"
+    assert finish["error_message"] == "writer failed"
+
+
+def test_arun_with_retry_timeout_observer_treats_bubble_up_as_non_error():
+    events: list[dict] = []
+
+    class BubbleProc:
+        async def ainvoke(self, input, config):
+            raise GraphInterrupt(())
+
+    task = _make_task(BubbleProc(), timeout=0.05, name="bubble")
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    async def _run() -> None:
+        with pytest.raises(GraphInterrupt):
+            await arun_with_retry(task, retry_policy=None)
+
+    asyncio.run(_run())
+
+    finish = next(payload for payload in events if payload["event"] == "finish")
+    assert finish["status"] == "success"
+    assert finish["error_type"] is None
+    assert finish["error_message"] is None

libs/langgraph/tests/test_time_travel.py

@@ -1113,6 +1113,70 @@ def test_subgraph_interrupt_replay_from_parent_then_resume(
     ]
 
 
+def test_subgraph_interrupt_resume_with_explicit_head_checkpoint_id(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Resume with Command(resume=...) plus the current head checkpoint_id
+    in config. The subgraph must continue from the interrupted node, not
+    restart from scratch. Explicit checkpoint_id triggers is_replaying but
+    this is a resume, not a time-travel, so ReplayState should not apply."""
+
+    called: list[str] = []
+
+    def step_a(state: State) -> State:
+        called.append("step_a")
+        return {"value": ["sub_a"]}
+
+    def ask_human(state: State) -> State:
+        called.append("ask_human")
+        answer = interrupt("Provide input:")
+        return {"value": [f"human:{answer}"]}
+
+    def step_b(state: State) -> State:
+        called.append("step_b")
+        return {"value": ["sub_b"]}
+
+    subgraph = (
+        StateGraph(State)
+        .add_node("step_a", step_a)
+        .add_node("ask_human", ask_human)
+        .add_node("step_b", step_b)
+        .add_edge(START, "step_a")
+        .add_edge("step_a", "ask_human")
+        .add_edge("ask_human", "step_b")
+        .compile(checkpointer=True)
+    )
+
+    graph = (
+        StateGraph(State)
+        .add_node("subgraph_node", subgraph)
+        .add_edge(START, "subgraph_node")
+        .compile(checkpointer=sync_checkpointer)
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Run until interrupt fires in subgraph
+    graph.invoke({"value": []}, config)
+    assert called == ["step_a", "ask_human"]
+
+    # Resume with explicit head checkpoint_id in config
+    head_checkpoint_id = graph.get_state(config).config["configurable"]["checkpoint_id"]
+    called.clear()
+    resume_config = {
+        "configurable": {
+            "thread_id": "1",
+            "checkpoint_id": head_checkpoint_id,
+            "checkpoint_ns": "",
+        }
+    }
+    result = graph.invoke(Command(resume="answer"), resume_config)
+
+    assert called == ["ask_human", "step_b"]
+    assert "__interrupt__" not in result
+    assert result["value"] == ["sub_a", "human:answer", "sub_b"]
+
+
 def test_subgraph_replay_loads_accumulated_state_then_resume(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/tests/test_utils.py

@@ -427,32 +427,3 @@ def test_callback_manager_copies_configurable_ids_to_tracing_metadata() -> None:
         "thread_id": "th-123",
         "user_id": "uid-1",
     }
-
-
-def test_ensure_config_merges_configurable_across_configs() -> None:
-    """`ensure_config(bound, invoke_time)` should merge `configurable` dicts.
-
-    Prior to the fix, a later config's `configurable` dict fully overwrote an
-    earlier one, causing values bound via `with_config({"configurable": {...}})`
-    (e.g. `ls_agent_type="root"` set by `create_agent`) to be dropped whenever
-    an invoke-time config supplied any other configurable key like `thread_id`.
-    """
-    bound: RunnableConfig = {
-        "configurable": {"ls_agent_type": "root", "custom_setting": "keep_me"},
-        "metadata": {"ls_integration": "langchain_create_agent"},
-    }
-    invoke_time: RunnableConfig = {
-        "configurable": {"thread_id": "t-1"},
-    }
-    merged = ensure_config(bound, invoke_time)
-    # Both the bound and invoke-time configurable keys are preserved.
-    assert merged["configurable"] == {
-        "ls_agent_type": "root",
-        "custom_setting": "keep_me",
-        "thread_id": "t-1",
-    }
-    # Invoke-time values still override bound values when they collide.
-    override: RunnableConfig = {"configurable": {"ls_agent_type": "subagent"}}
-    merged2 = ensure_config(bound, override)
-    assert merged2["configurable"]["ls_agent_type"] == "subagent"
-    assert merged2["configurable"]["custom_setting"] == "keep_me"

libs/langgraph/uv.lock

@@ -1348,7 +1348,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0"
+version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -1360,14 +1360,14 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/92/fe/20190232d9b513242899dbb0c2bb77e31b4d61e343743adbe90ebc2603d2/langchain_core-1.3.0.tar.gz", hash = "sha256:14a39f528bf459aa3aa40d0a7f7f1bae7520d435ef991ae14a4ceb74d8c49046", size = 860755, upload-time = "2026-04-17T14:51:38.298Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f4/fe/abeae8d0d2899e191d67c6c7f065f7e52a953f30b21ef327fa49084e4af9/langchain_core-1.3.1.tar.gz", hash = "sha256:41b384055799f93f34520df6bf7b80e2e5e23153cdfd46874251c6c9916ea030", size = 862403, upload-time = "2026-04-23T18:54:01.857Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f8/e2/dbfa347aa072a6dc4cd38d6f9ebfc730b4c14c258c47f480f4c5c546f177/langchain_core-1.3.0-py3-none-any.whl", hash = "sha256:baf16ee028475df177b9ab8869a751c79406d64a6f12125b93802991b566cced", size = 515140, upload-time = "2026-04-17T14:51:36.274Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/c2/8493be505921857988db068b7c027f28a9b1587b4425c6a32b1221c9c9fe/langchain_core-1.3.1-py3-none-any.whl", hash = "sha256:8b13d19d3bed3f4768df12c7f6932d2ada715f3ac9fd020c63d28c693968269e", size = 515879, upload-time = "2026-04-23T18:53:59.94Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.1.8"
+version = "1.1.9"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1548,7 +1548,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1742,7 +1742,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.10"
+version = "1.0.11"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },
@@ -1751,7 +1751,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.0.0" },
+    { name = "langchain-core", specifier = ">=1.3.1" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
 ]
 
@@ -2140,7 +2140,7 @@ wheels = [
 
 [[package]]
 name = "nbconvert"
-version = "7.17.0"
+version = "7.17.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "beautifulsoup4" },
@@ -2158,9 +2158,9 @@ dependencies = [
     { name = "pygments" },
     { name = "traitlets" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/38/47/81f886b699450d0569f7bc551df2b1673d18df7ff25cc0c21ca36ed8a5ff/nbconvert-7.17.0.tar.gz", hash = "sha256:1b2696f1b5be12309f6c7d707c24af604b87dfaf6d950794c7b07acab96dda78", size = 862855, upload-time = "2026-01-29T16:37:48.478Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/01/b1/708e53fe2e429c103c6e6e159106bcf0357ac41aa4c28772bd8402339051/nbconvert-7.17.1.tar.gz", hash = "sha256:34d0d0a7e73ce3cbab6c5aae8f4f468797280b01fd8bd2ca746da8569eddd7d2", size = 865311, upload-time = "2026-04-08T00:44:14.914Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0d/4b/8d5f796a792f8a25f6925a96032f098789f448571eb92011df1ae59e8ea8/nbconvert-7.17.0-py3-none-any.whl", hash = "sha256:4f99a63b337b9a23504347afdab24a11faa7d86b405e5c8f9881cd313336d518", size = 261510, upload-time = "2026-01-29T16:37:46.322Z" },
+    { url = "https://files.pythonhosted.org/packages/67/f8/bb0a9d5f46819c821dc1f004aa2cc29b1d91453297dbf5ff20470f00f193/nbconvert-7.17.1-py3-none-any.whl", hash = "sha256:aa85c087b435e7bf1ffd03319f658e285f2b89eccab33bc1ba7025495ab3e7c8", size = 261927, upload-time = "2026-04-08T00:44:12.845Z" },
 ]
 
 [[package]]
@@ -3018,11 +3018,11 @@ wheels = [
 
 [[package]]
 name = "python-dotenv"
-version = "1.2.1"
+version = "1.2.2"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/f0/26/19cadc79a718c5edbec86fd4919a6b6d3f681039a2f6d66d14be94e75fb9/python_dotenv-1.2.1.tar.gz", hash = "sha256:42667e897e16ab0d66954af0e60a9caa94f0fd4ecf3aaf6d2d260eec1aa36ad6", size = 44221, upload-time = "2025-10-26T15:12:10.434Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/82/ed/0301aeeac3e5353ef3d94b6ec08bbcabd04a72018415dcb29e588514bba8/python_dotenv-1.2.2.tar.gz", hash = "sha256:2c371a91fbd7ba082c2c1dc1f8bf89ca22564a087c2c287cd9b662adde799cf3", size = 50135, upload-time = "2026-03-01T16:00:26.196Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/14/1b/a298b06749107c305e1fe0f814c6c74aea7b2f1e10989cb30f544a1b3253/python_dotenv-1.2.1-py3-none-any.whl", hash = "sha256:b81ee9561e9ca4004139c6cbba3a238c32b03e4894671e181b671e8cb8425d61", size = 21230, upload-time = "2025-10-26T15:12:09.109Z" },
+    { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" },
 ]
 
 [[package]]

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -82,6 +82,7 @@ from langchain_core.tools.base import (
     _is_injected_arg_type,
     get_all_basemodel_annotations,
 )
+from langgraph._internal._constants import CONF, CONFIG_KEY_READ
 from langgraph._internal._runnable import RunnableCallable
 from langgraph.errors import GraphBubbleUp
 from langgraph.graph.message import REMOVE_ALL_MESSAGES
@@ -800,7 +801,7 @@ class ToolNode(RunnableCallable):
         # Construct ToolRuntime instances at the top level for each tool call
         tool_runtimes = []
         for call, cfg in zip(tool_calls, config_list, strict=False):
-            state = self._extract_state(input)
+            state = self._extract_state(input, cfg)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -835,7 +836,7 @@ class ToolNode(RunnableCallable):
         # Construct ToolRuntime instances at the top level for each tool call
         tool_runtimes = []
         for call, cfg in zip(tool_calls, config_list, strict=False):
-            state = self._extract_state(input)
+            state = self._extract_state(input, cfg)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -859,14 +860,30 @@ class ToolNode(RunnableCallable):
 
     def _combine_tool_outputs(
         self,
-        outputs: list[ToolMessage | Command],
+        outputs: list[ToolMessage | Command | list[ToolMessage | Command]],
         input_type: Literal["list", "dict", "tool_calls"],
     ) -> list[Command | list[ToolMessage] | dict[str, list[ToolMessage]]]:
+        # Flatten list entries from tools that returned multiple items
+        flat_outputs: list[ToolMessage | Command]
+        if any(isinstance(output, list) for output in outputs):
+            flat_outputs = []
+            for output in outputs:
+                if isinstance(output, list):
+                    flat_outputs.extend(output)
+                else:
+                    flat_outputs.append(output)
+        else:
+            flat_outputs = cast("list[ToolMessage | Command]", outputs)
+
         # preserve existing behavior for non-command tool outputs for backwards
         # compatibility
-        if not any(isinstance(output, Command) for output in outputs):
+        if not any(isinstance(output, Command) for output in flat_outputs):
             # TypedDict, pydantic, dataclass, etc. should all be able to load from dict
-            return outputs if input_type == "list" else {self._messages_key: outputs}
+            return (
+                flat_outputs
+                if input_type == "list"
+                else {self._messages_key: flat_outputs}
+            )
 
         # LangGraph will automatically handle list of Command and non-command node
         # updates
@@ -876,7 +893,7 @@ class ToolNode(RunnableCallable):
 
         # combine all parent commands with goto into a single parent command
         parent_command: Command | None = None
-        for output in outputs:
+        for output in flat_outputs:
             if isinstance(output, Command):
                 if (
                     output.graph is Command.PARENT
@@ -906,7 +923,7 @@ class ToolNode(RunnableCallable):
         request: ToolCallRequest,
         input_type: Literal["list", "dict", "tool_calls"],
         config: RunnableConfig,
-    ) -> ToolMessage | Command:
+    ) -> ToolMessage | Command | list[Command | ToolMessage]:
         """Execute tool call with configured error handling.
 
         Args:
@@ -915,7 +932,7 @@ class ToolNode(RunnableCallable):
             config: Runnable configuration.
 
         Returns:
-            ToolMessage or Command.
+            ToolMessage, Command, or list of Command/ToolMessage.
 
         Raises:
             Exception: If tool fails and handle_tool_errors is False.
@@ -947,6 +964,11 @@ class ToolNode(RunnableCallable):
                     call["name"], exc, call["args"], filtered_errors
                 ) from exc
 
+            # Inside try so validation errors route through _handle_tool_errors
+            return self._normalize_tool_response(
+                response, request.tool_call, input_type
+            )
+
         # GraphInterrupt is a special exception that will always be raised.
         # It can be triggered in the following scenarios,
         # Where GraphInterrupt(GraphBubbleUp) is raised from an `interrupt` invocation
@@ -988,23 +1010,12 @@ class ToolNode(RunnableCallable):
                 status="error",
             )
 
-        # Process successful response
-        if isinstance(response, Command):
-            # Validate Command before returning to handler
-            return self._validate_tool_command(response, request.tool_call, input_type)
-        if isinstance(response, ToolMessage):
-            response.content = cast("str | list", msg_content_output(response.content))
-            return response
-
-        msg = f"Tool {call['name']} returned unexpected type: {type(response)}"
-        raise TypeError(msg)
-
     def _run_one(
         self,
         call: ToolCall,
         input_type: Literal["list", "dict", "tool_calls"],
         tool_runtime: ToolRuntime,
-    ) -> ToolMessage | Command:
+    ) -> ToolMessage | Command | list[Command | ToolMessage]:
         """Execute single tool call with wrap_tool_call wrapper if configured.
 
         Args:
@@ -1059,7 +1070,7 @@ class ToolNode(RunnableCallable):
         request: ToolCallRequest,
         input_type: Literal["list", "dict", "tool_calls"],
         config: RunnableConfig,
-    ) -> ToolMessage | Command:
+    ) -> ToolMessage | Command | list[Command | ToolMessage]:
         """Execute tool call asynchronously with configured error handling.
 
         Args:
@@ -1068,7 +1079,7 @@ class ToolNode(RunnableCallable):
             config: Runnable configuration.
 
         Returns:
-            ToolMessage or Command.
+            ToolMessage, Command, or list of Command/ToolMessage.
 
         Raises:
             Exception: If tool fails and handle_tool_errors is False.
@@ -1100,6 +1111,11 @@ class ToolNode(RunnableCallable):
                     call["name"], exc, call["args"], filtered_errors
                 ) from exc
 
+            # Inside try so validation errors route through _handle_tool_errors
+            return self._normalize_tool_response(
+                response, request.tool_call, input_type
+            )
+
         # GraphInterrupt is a special exception that will always be raised.
         # It can be triggered in the following scenarios,
         # Where GraphInterrupt(GraphBubbleUp) is raised from an `interrupt` invocation
@@ -1141,23 +1157,12 @@ class ToolNode(RunnableCallable):
                 status="error",
             )
 
-        # Process successful response
-        if isinstance(response, Command):
-            # Validate Command before returning to handler
-            return self._validate_tool_command(response, request.tool_call, input_type)
-        if isinstance(response, ToolMessage):
-            response.content = cast("str | list", msg_content_output(response.content))
-            return response
-
-        msg = f"Tool {call['name']} returned unexpected type: {type(response)}"
-        raise TypeError(msg)
-
     async def _arun_one(
         self,
         call: ToolCall,
         input_type: Literal["list", "dict", "tool_calls"],
         tool_runtime: ToolRuntime,
-    ) -> ToolMessage | Command:
+    ) -> ToolMessage | Command | list[Command | ToolMessage]:
         """Execute single tool call asynchronously with awrap_tool_call wrapper if configured.
 
         Args:
@@ -1273,18 +1278,37 @@ class ToolNode(RunnableCallable):
         return None
 
     def _extract_state(
-        self, input: list[AnyMessage] | dict[str, Any] | BaseModel
+        self,
+        input: list[AnyMessage] | dict[str, Any] | BaseModel,
+        config: RunnableConfig,
     ) -> list[AnyMessage] | dict[str, Any] | BaseModel:
-        """Extract state from input, handling ToolCallWithContext if present.
+        """Extract state from input.
 
-        Args:
-            input: The input which may be raw state or ToolCallWithContext.
+        Three input shapes:
 
-        Returns:
-            The actual state to pass to wrap_tool_call wrappers.
+        - `ToolCallWithContext` dict — legacy Send payload carrying an inlined
+          state snapshot; return `input["state"]`.
+        - list of `ToolCall` dicts — new Send payload with no inlined state;
+          hydrate state from channels via `CONFIG_KEY_READ`.
+        - regular graph state (dict/list/BaseModel) — return `input` as-is.
         """
         if isinstance(input, dict) and input.get("__type") == "tool_call_with_context":
             return input["state"]
+        if (
+            isinstance(input, list)
+            and input
+            and isinstance(input[-1], dict)
+            and input[-1].get("type") == "tool_call"
+        ):
+            read = config.get(CONF, {}).get(CONFIG_KEY_READ)
+            if read is None:
+                return {}
+            # Pregel installs CONFIG_KEY_READ as
+            # `functools.partial(local_read, scratchpad, channels, managed, task)`.
+            # Match the previous inlined-state contract by reading channels only;
+            # managed values have their own injection path (`ToolRuntime.context`).
+            channels = read.args[1]
+            return cast("dict[str, Any]", read(list(channels), True))
         return input
 
     def _inject_tool_args(
@@ -1404,11 +1428,84 @@ class ToolNode(RunnableCallable):
         tool_call_copy["args"] = {**stripped_args, **injected_args}
         return tool_call_copy
 
+    def _normalize_tool_response(
+        self,
+        response: Any,
+        tool_call: ToolCall,
+        input_type: Literal["list", "dict", "tool_calls"],
+    ) -> ToolMessage | Command | list[Command | ToolMessage]:
+        """Validate and normalize a tool's raw return value."""
+        if isinstance(response, Command):
+            return self._validate_tool_command(response, tool_call, input_type)
+        if isinstance(response, ToolMessage):
+            response.content = cast("str | list", msg_content_output(response.content))
+            return response
+        if isinstance(response, list):
+            if all(isinstance(r, (Command, ToolMessage)) for r in response):
+                return self._validate_tool_command_list(response, tool_call, input_type)
+            msg = (
+                f"Tool {tool_call['name']} returned a list with invalid element "
+                "types: expected all Command or ToolMessage"
+            )
+            raise TypeError(msg)
+        msg = f"Tool {tool_call['name']} returned unexpected type: {type(response)}"
+        raise TypeError(msg)
+
+    def _validate_tool_command_list(
+        self,
+        response: list[Command | ToolMessage],
+        tool_call: ToolCall,
+        input_type: Literal["list", "dict", "tool_calls"],
+    ) -> list[Command | ToolMessage]:
+        """Validate a list of Command/ToolMessage returned by a single tool call.
+
+        Requires exactly one terminating ToolMessage (matching the outer tool_call_id)
+        across the list — either as a top-level element or nested in a
+        Command.update["messages"].
+        """
+        expected_id = tool_call["id"]
+
+        terminator_count = 0
+        for item in response:
+            if isinstance(item, ToolMessage):
+                if item.tool_call_id == expected_id:
+                    terminator_count += 1
+            elif isinstance(item, Command) and isinstance(item.update, dict):
+                for msg in item.update.get(self._messages_key, []):
+                    if isinstance(msg, ToolMessage) and msg.tool_call_id == expected_id:
+                        terminator_count += 1
+
+        if terminator_count != 1:
+            msg = (
+                f"Tool {tool_call['name']} returned a list with "
+                f"{terminator_count} messages bound to tool_call_id "
+                f"{expected_id!r}; expected exactly one terminating ToolMessage."
+            )
+            raise ValueError(msg)
+
+        # Per-Command normalization still runs, but the list-level count above
+        # already guarantees exactly one terminator, so individual Commands may
+        # lack one.
+        validated: list[Command | ToolMessage] = []
+        for item in response:
+            if isinstance(item, Command):
+                validated.append(
+                    self._validate_tool_command(
+                        item, tool_call, input_type, require_terminator=False
+                    )
+                )
+            else:
+                item.content = cast("str | list", msg_content_output(item.content))
+                validated.append(item)
+        return validated
+
     def _validate_tool_command(
         self,
         command: Command,
         call: ToolCall,
         input_type: Literal["list", "dict", "tool_calls"],
+        *,
+        require_terminator: bool = True,
     ) -> Command:
         if isinstance(command.update, dict):
             # input type is dict when ToolNode is invoked with a dict input
@@ -1458,7 +1555,11 @@ class ToolNode(RunnableCallable):
 
         # validate that we always have a ToolMessage matching the tool call in
         # Command.update if command is sent to the CURRENT graph
-        if updated_command.graph is None and not has_matching_tool_message:
+        if (
+            require_terminator
+            and updated_command.graph is None
+            and not has_matching_tool_message
+        ):
             example_update = (
                 '`Command(update={"messages": '
                 '[ToolMessage("Success", tool_call_id=tool_call_id), ...]}, ...)`'

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "1.0.10"
+version = "1.0.11"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"
@@ -25,7 +25,7 @@ classifiers = [
 ]
 dependencies = [
     "langgraph-checkpoint>=2.1.0,<5.0.0",
-    "langchain-core>=1.0.0",
+    "langchain-core>=1.3.1",
 ]
 
 [project.urls]

libs/prebuilt/tests/test_on_tool_call.py

@@ -1320,6 +1320,98 @@ async def test_state_extraction_with_tool_call_with_context_async() -> None:
     assert "tool_call" not in state_seen[0]
 
 
+def _config_with_channel_read(
+    channel_values: dict[str, object],
+    store: BaseStore | None = None,
+) -> RunnableConfig:
+    """Build a config that mimics `CONFIG_KEY_READ` as Pregel installs it.
+
+    Pregel always installs a `functools.partial(local_read, scratchpad,
+    channels, managed, task)`, and `ToolNode` introspects that partial to
+    learn channel names. The stub matches the shape: partial whose second and
+    third positional args are `channels` and `managed` mappings.
+    """
+    import functools
+
+    channels_stub = {k: None for k in channel_values}
+    managed_stub: dict[str, object] = {}
+
+    # Shape matches pregel's real partial:
+    # functools.partial(local_read, scratchpad, channels, managed, task)
+    def _read(scratchpad, channels, managed, task, select, fresh):  # noqa: ARG001
+        if isinstance(select, str):
+            return channel_values[select]
+        return {k: channel_values[k] for k in select if k in channel_values}
+
+    read = functools.partial(_read, None, channels_stub, managed_stub, None)
+    cfg = _create_config_with_runtime(store)
+    cfg["configurable"]["__pregel_read"] = read
+    return cfg
+
+
+def test_list_form_send_hydrates_state_from_channel_read() -> None:
+    """Send('tools', [tool_call]) with no inlined state should hydrate
+    ToolRuntime.state from CONFIG_KEY_READ (full state read)."""
+    state_seen = []
+
+    def state_inspector_handler(
+        request: ToolCallRequest,
+        execute: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        state_seen.append(request.state)
+        return execute(request)
+
+    channel_values = {
+        "messages": [AIMessage("from channels")],
+        "files": {"/a.md": "body"},
+    }
+
+    tool_node = ToolNode([add], wrap_tool_call=state_inspector_handler)
+
+    tool_call: ToolCall = {
+        "name": "add",
+        "args": {"a": 1, "b": 2},
+        "id": "call_1",
+        "type": "tool_call",
+    }
+
+    tool_node.invoke([tool_call], config=_config_with_channel_read(channel_values))
+
+    assert len(state_seen) == 1
+    got = state_seen[0]
+    assert got == channel_values
+    assert "messages" in got and "files" in got
+
+
+async def test_list_form_send_hydrates_state_async() -> None:
+    state_seen = []
+
+    def state_inspector_handler(
+        request: ToolCallRequest,
+        execute: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        state_seen.append(request.state)
+        return execute(request)
+
+    channel_values = {"messages": [AIMessage("from channels")], "files": {}}
+
+    tool_node = ToolNode([add], wrap_tool_call=state_inspector_handler)
+
+    tool_call: ToolCall = {
+        "name": "add",
+        "args": {"a": 1, "b": 2},
+        "id": "call_1",
+        "type": "tool_call",
+    }
+
+    await tool_node.ainvoke(
+        [tool_call], config=_config_with_channel_read(channel_values)
+    )
+
+    assert len(state_seen) == 1
+    assert state_seen[0] == channel_values
+
+
 def test_tool_call_request_is_frozen() -> None:
     """Test that ToolCallRequest raises deprecation warnings on direct attribute reassignment."""
     tool_call: ToolCall = {"name": "add", "args": {"a": 1, "b": 2}, "id": "call_1"}

libs/prebuilt/tests/test_tool_node.py

@@ -2223,3 +2223,195 @@ def test_tool_node_injected_state_overwrites_llm_value() -> None:
     )
     tool_message = result["messages"][-1]
     assert tool_message.content == "PUBLIC_DATA"
+
+
+class _ReturningTool(BaseTool):
+    """A tool that returns a configured value verbatim."""
+
+    name: str = "list_tool"
+    description: str = "Returns a configured value"
+    return_value: Any = None
+
+    def _run(self, **kwargs: Any) -> Any:
+        return self.return_value
+
+    async def _arun(self, **kwargs: Any) -> Any:
+        return self.return_value
+
+
+def _list_tool_call(outer_id: str = "call-1") -> dict[str, Any]:
+    return {"name": "list_tool", "args": {}, "id": outer_id, "type": "tool_call"}
+
+
+def _invoke_returning(
+    return_value: Any,
+    *,
+    outer_id: str = "call-1",
+    handle_tool_errors: bool = True,
+) -> Any:
+    node = ToolNode(
+        [_ReturningTool(return_value=return_value)],
+        handle_tool_errors=handle_tool_errors,
+    )
+    return node.invoke(
+        {"messages": [AIMessage("", tool_calls=[_list_tool_call(outer_id)])]},
+        config=_create_config_with_runtime(),
+    )
+
+
+def test_tool_node_list_return_command_and_tool_message() -> None:
+    """Valid: tool returns [Command(update={...}), ToolMessage(...)]."""
+    outer_id = "call-1"
+    result = _invoke_returning(
+        [
+            Command(update={"foo": "bar"}),
+            ToolMessage(content="done", tool_call_id=outer_id),
+        ]
+    )
+    assert isinstance(result, list)
+    commands = [r for r in result if isinstance(r, Command)]
+    assert len(commands) == 1
+    assert commands[0].update == {"foo": "bar"}
+    non_commands = [r for r in result if not isinstance(r, Command)]
+    assert len(non_commands) == 1
+    assert isinstance(non_commands[0], dict)
+    msgs = non_commands[0]["messages"]
+    assert len(msgs) == 1
+    assert isinstance(msgs[0], ToolMessage)
+    assert msgs[0].content == "done"
+    assert msgs[0].tool_call_id == outer_id
+
+
+def test_tool_node_list_return_nested_terminator() -> None:
+    """Valid: terminator nested inside Command.update['messages']."""
+    outer_id = "call-1"
+    result = _invoke_returning(
+        [
+            Command(update={"foo": "bar"}),
+            Command(
+                update={
+                    "messages": [ToolMessage(content="done", tool_call_id=outer_id)]
+                }
+            ),
+        ]
+    )
+    assert isinstance(result, list)
+    commands = [r for r in result if isinstance(r, Command)]
+    assert len(commands) == 2
+    updates = [c.update for c in commands]
+    assert {"foo": "bar"} in updates
+    msgs_update = next(u for u in updates if "messages" in (u or {}))
+    assert any(
+        isinstance(m, ToolMessage) and m.tool_call_id == outer_id
+        for m in msgs_update["messages"]
+    )
+
+
+def test_tool_node_list_return_parent_goto_with_terminator() -> None:
+    """Valid: [Command(graph=PARENT, goto=[Send(...)]), ToolMessage(...)]."""
+    outer_id = "call-1"
+    result = _invoke_returning(
+        [
+            Command(graph=Command.PARENT, goto=[Send("child", {})]),
+            ToolMessage(content="ok", tool_call_id=outer_id),
+        ]
+    )
+    assert isinstance(result, list)
+    parent_cmds = [
+        r for r in result if isinstance(r, Command) and r.graph is Command.PARENT
+    ]
+    assert len(parent_cmds) == 1
+    assert isinstance(parent_cmds[0].goto, list)
+    assert any(isinstance(s, Send) for s in parent_cmds[0].goto)
+    non_commands = [r for r in result if not isinstance(r, Command)]
+    assert len(non_commands) == 1
+
+
+def test_tool_node_list_return_no_terminator_raises() -> None:
+    """Invalid: list with no terminating ToolMessage."""
+    with pytest.raises(ValueError, match="0 messages bound to tool_call_id"):
+        _invoke_returning([Command(update={"foo": "bar"})], handle_tool_errors=False)
+
+
+def test_tool_node_list_return_multiple_terminators_raises() -> None:
+    """Invalid: list with two terminating ToolMessages."""
+    outer_id = "call-1"
+    with pytest.raises(ValueError, match="2 messages bound to tool_call_id"):
+        _invoke_returning(
+            [
+                ToolMessage(content="a", tool_call_id=outer_id),
+                ToolMessage(content="b", tool_call_id=outer_id),
+            ],
+            handle_tool_errors=False,
+        )
+
+
+def test_tool_node_list_return_validation_error_handled() -> None:
+    """handle_tool_errors=True converts validation errors to an error ToolMessage."""
+    result = _invoke_returning([Command(update={"foo": "bar"})])
+    assert isinstance(result, dict)
+    msg = result["messages"][0]
+    assert isinstance(msg, ToolMessage)
+    assert msg.status == "error"
+    assert "0 messages bound to tool_call_id" in msg.content
+
+
+async def test_tool_node_list_return_async_smoke() -> None:
+    """Async path parallels sync for the happy case."""
+    outer_id = "call-1"
+    node = ToolNode(
+        [
+            _ReturningTool(
+                return_value=[
+                    Command(update={"foo": "bar"}),
+                    ToolMessage(content="done", tool_call_id=outer_id),
+                ]
+            )
+        ]
+    )
+    result = await node.ainvoke(
+        {"messages": [AIMessage("", tool_calls=[_list_tool_call(outer_id)])]},
+        config=_create_config_with_runtime(),
+    )
+    assert isinstance(result, list)
+    commands = [r for r in result if isinstance(r, Command)]
+    assert len(commands) == 1 and commands[0].update == {"foo": "bar"}
+
+
+def test_tool_node_list_return_mixed_with_regular_tool() -> None:
+    """List-returning tool and a regular tool dispatched from the same AIMessage."""
+    list_tool_id = "call-list"
+    regular_tool_id = "call-regular"
+    list_tool = _ReturningTool(
+        return_value=[
+            Command(update={"foo": "bar"}),
+            ToolMessage(content="list done", tool_call_id=list_tool_id),
+        ]
+    )
+
+    def regular_tool(x: int) -> str:
+        """A normal tool."""
+        return f"regular: {x}"
+
+    tool_calls = [
+        {"name": "list_tool", "args": {}, "id": list_tool_id, "type": "tool_call"},
+        {
+            "name": "regular_tool",
+            "args": {"x": 7},
+            "id": regular_tool_id,
+            "type": "tool_call",
+        },
+    ]
+    node = ToolNode([list_tool, regular_tool])
+    result = node.invoke(
+        {"messages": [AIMessage("", tool_calls=tool_calls)]},
+        config=_create_config_with_runtime(),
+    )
+    assert isinstance(result, list)
+    commands = [r for r in result if isinstance(r, Command)]
+    assert len(commands) == 1
+    assert commands[0].update == {"foo": "bar"}
+    all_msgs = [m for r in result if isinstance(r, dict) for m in r["messages"]]
+    tool_call_ids = {m.tool_call_id for m in all_msgs}
+    assert list_tool_id in tool_call_ids
+    assert regular_tool_id in tool_call_ids

libs/prebuilt/uv.lock

@@ -249,7 +249,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0"
+version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -261,14 +261,14 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/92/fe/20190232d9b513242899dbb0c2bb77e31b4d61e343743adbe90ebc2603d2/langchain_core-1.3.0.tar.gz", hash = "sha256:14a39f528bf459aa3aa40d0a7f7f1bae7520d435ef991ae14a4ceb74d8c49046", size = 860755, upload-time = "2026-04-17T14:51:38.298Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f4/fe/abeae8d0d2899e191d67c6c7f065f7e52a953f30b21ef327fa49084e4af9/langchain_core-1.3.1.tar.gz", hash = "sha256:41b384055799f93f34520df6bf7b80e2e5e23153cdfd46874251c6c9916ea030", size = 862403, upload-time = "2026-04-23T18:54:01.857Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f8/e2/dbfa347aa072a6dc4cd38d6f9ebfc730b4c14c258c47f480f4c5c546f177/langchain_core-1.3.0-py3-none-any.whl", hash = "sha256:baf16ee028475df177b9ab8869a751c79406d64a6f12125b93802991b566cced", size = 515140, upload-time = "2026-04-17T14:51:36.274Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/c2/8493be505921857988db068b7c027f28a9b1587b4425c6a32b1221c9c9fe/langchain_core-1.3.1-py3-none-any.whl", hash = "sha256:8b13d19d3bed3f4768df12c7f6932d2ada715f3ac9fd020c63d28c693968269e", size = 515879, upload-time = "2026-04-23T18:53:59.94Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.1.8"
+version = "1.1.9"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -352,7 +352,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -490,7 +490,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.10"
+version = "1.0.11"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -535,7 +535,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.0.0" },
+    { name = "langchain-core", specifier = ">=1.3.1" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
 ]
 

libs/sdk-py/uv.lock

@@ -262,7 +262,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0"
+version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -274,14 +274,14 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/92/fe/20190232d9b513242899dbb0c2bb77e31b4d61e343743adbe90ebc2603d2/langchain_core-1.3.0.tar.gz", hash = "sha256:14a39f528bf459aa3aa40d0a7f7f1bae7520d435ef991ae14a4ceb74d8c49046", size = 860755, upload-time = "2026-04-17T14:51:38.298Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f4/fe/abeae8d0d2899e191d67c6c7f065f7e52a953f30b21ef327fa49084e4af9/langchain_core-1.3.1.tar.gz", hash = "sha256:41b384055799f93f34520df6bf7b80e2e5e23153cdfd46874251c6c9916ea030", size = 862403, upload-time = "2026-04-23T18:54:01.857Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f8/e2/dbfa347aa072a6dc4cd38d6f9ebfc730b4c14c258c47f480f4c5c546f177/langchain_core-1.3.0-py3-none-any.whl", hash = "sha256:baf16ee028475df177b9ab8869a751c79406d64a6f12125b93802991b566cced", size = 515140, upload-time = "2026-04-17T14:51:36.274Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/c2/8493be505921857988db068b7c027f28a9b1587b4425c6a32b1221c9c9fe/langchain_core-1.3.1-py3-none-any.whl", hash = "sha256:8b13d19d3bed3f4768df12c7f6932d2ada715f3ac9fd020c63d28c693968269e", size = 515879, upload-time = "2026-04-23T18:53:59.94Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.1.8"
+version = "1.1.9"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -365,7 +365,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.0.3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -413,7 +413,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.10"
+version = "1.0.11"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },
@@ -422,7 +422,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.0.0" },
+    { name = "langchain-core", specifier = ">=1.3.1" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
 ]