feat: support list of hooks for pre_model_hook and post_model_hook

Allow `pre_model_hook` and `post_model_hook` in `create_react_agent` to accept a list of `RunnableLike` callables in addition to a single hook. When a list is provided the hooks are composed in order: each hook receives the graph state merged with all prior hooks' updates, and the final merged update is returned to the graph. This mirrors the middleware-stack pattern familiar from HTTP frameworks (Express, FastAPI, Starlette) and allows hook logic to be written as small, reusable units that can be composed without manually threading state between them.
2026-10-06 08:25:14 +02:00 · 2026-05-08 16:02:24 -04:00
31 changed files with 400 additions and 1264 deletions
@@ -279,7 +279,7 @@ wheels = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0"
+version = "3.1.0a4"
 description = "Library with a Postgres implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"
@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-  "langgraph-checkpoint>=4.1.0,<5.0.0",
+  "langgraph-checkpoint>=4.1.0a4,<5.0.0",
  "orjson>=3.11.5",
  "psycopg>=3.2.0",
  "psycopg-pool>=3.2.0",
@@ -244,7 +244,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
@@ -257,9 +257,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
 ]

 [[package]]
@@ -276,7 +276,7 @@ wheels = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -324,7 +324,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0"
+version = "3.1.0a4"
 source = { editable = "." }
 dependencies = [
    { name = "langgraph-checkpoint" },
@@ -1234,11 +1234,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.0"
+version = "3.1.0a1"
 description = "Library with a SQLite implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"
@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-    "langgraph-checkpoint>=4.1.0,<5.0.0",
+    "langgraph-checkpoint>=4.1.0a4,<5.0.0",
    "aiosqlite>=0.20",
    "sqlite-vec>=0.1.6",
 ]
@@ -253,11 +253,10 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
-    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -266,26 +265,14 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.15"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -333,7 +320,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.0"
+version = "3.1.0a1"
 source = { editable = "." }
 dependencies = [
    { name = "aiosqlite" },
@@ -1161,11 +1148,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -60,29 +60,20 @@ class CheckpointMetadata(TypedDict, total=False):
    """
    run_id: str
    """The ID of the run that created this checkpoint."""
-    counters_since_delta_snapshot: dict[str, tuple[int, int]]
-    """Per-channel counters since the last `_DeltaSnapshot` was written.
+    delta_updates_since_snapshot: dict[str, int]
+    """Per-channel update count since the last `_DeltaSnapshot` was written.

    !!! warning "Beta"

        This metadata field backs `DeltaChannel` (beta). The key name and
        contents may change while the delta-channel design stabilizes.

-    Maps channel name -> `(updates, supersteps)`:
-
-    - index 0 (`updates`): number of supersteps that wrote to this channel
-      since its last snapshot blob.
-    - index 1 (`supersteps`): total supersteps elapsed since this channel's
-      last snapshot, regardless of whether the channel was written.
-
-    A snapshot fires when EITHER `updates >= ch.snapshot_frequency` OR
-    `supersteps >= DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT` (system-wide bound,
-    default 5000, env `LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT`).
-    The supersteps bound prevents unbounded ancestor walks on threads where
-    a delta channel exists but is no longer being updated.
-
-    Absent on threads that don't use delta channels. Persisted as a
-    2-element list in JSON (no native tuple).
+    Maps channel name → number of supersteps that wrote to this channel
+    since its last snapshot blob. Used by `pregel.create_checkpoint` to
+    decide when to write the next snapshot (when the count reaches the
+    channel's `snapshot_frequency`, snapshot fires and the count resets
+    to 0). Absent on threads that don't use delta channels. Version-format
+    independent — works for int, float, and string version schemes.
    """


@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.10"
@@ -268,7 +268,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
@@ -281,9 +281,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
 ]

 [[package]]
@@ -300,7 +300,7 @@ wheels = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -1463,11 +1463,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -1 +1 @@
-__version__ = "0.4.26"
+__version__ = "0.4.25"
@@ -35,12 +35,6 @@ DISALLOWED_BUILD_COMMAND_CHARS = [
 # This blocks background execution (cmd &) while allowing command
 # chaining (cmd1 && cmd2) which is common in build commands.
 _SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)")
-_API_VERSION_PATTERN = re.compile(
-    r"^(?P<major>\d+)"
-    r"(?:\.(?P<minor>\d+))?"
-    r"(?:\.(?P<patch>\d+))?"
-    r"(?:(?:\.|)(?:[A-Za-z][0-9A-Za-z]*))?$"
-)


 def has_disallowed_build_command_content(command: str) -> bool:
@@ -129,18 +123,6 @@ def _parse_node_version(version_str: str) -> int:
        ) from None


-def _parse_api_version_parts(version_str: str) -> tuple[int, ...]:
-    """Parse an API version into numeric components.
-
-    Supports optional prerelease suffixes, e.g. `0.9.0rc1`.
-    """
-    version_core = version_str.split("-", 1)[0]
-    match = _API_VERSION_PATTERN.fullmatch(version_core)
-    if not match:
-        raise ValueError("Version must be major or major.minor or major.minor.patch.")
-    return tuple(int(part) for part in match.groups() if part is not None)
-
-
 def _is_node_graph(spec: str | dict) -> bool:
    """Check if a graph is a Node.js graph based on the file extension."""
    if isinstance(spec, dict):
@@ -194,12 +176,12 @@ def validate_config(config: Config) -> Config:
            )
    if api_version:
        try:
-            parts = _parse_api_version_parts(api_version)
+            parts = tuple(map(int, api_version.split("-")[0].split(".")))
            if len(parts) > 3:
                raise ValueError(
                    "Version must be major or major.minor or major.minor.patch."
                )
-        except (TypeError, ValueError):
+        except TypeError:
            raise click.UsageError(
                f"Invalid version format: {api_version}.\n\n"
                "Pin to a minor version, e.g.:\n"
@@ -2944,23 +2944,6 @@ def test_docker_tag_with_api_version(in_config: bool):
    assert tag == f"langchain/langgraph-server:{version}-py3.11"


-@pytest.mark.parametrize("in_config", [False, True])
-@pytest.mark.parametrize("version", ["0.9.0rc1", "0.9.0.dev1"])
-def test_docker_tag_with_prerelease_api_version(version: str, in_config: bool):
-    """Test docker_tag with prerelease and dev api_version values."""
-    config = validate_config(
-        {
-            "python_version": "3.11",
-            "dependencies": ["."],
-            "graphs": {"agent": "./agent.py:graph"},
-            "api_version": version if in_config else None,
-        }
-    )
-
-    tag = docker_tag(config, api_version=version if not in_config else None)
-    assert tag == f"langchain/langgraph-api:{version}-py3.11"
-
-
 def test_config_to_docker_with_api_version():
    """Test config_to_docker function with api_version parameter."""

@@ -5,7 +5,7 @@ description = "uv workspace monorepo example for LangGraph CLI integration test"
 requires-python = ">=3.11"
 dependencies = [
    "langgraph>=0.6.0,<2",
-    "langchain-core>=1.3.3",
+    "langchain-core>=0.2.14",
 ]

 [tool.uv.workspace]
@@ -21,7 +21,7 @@ dependencies = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.3" },
+    { name = "langchain-core", specifier = ">=0.2.14" },
    { name = "langgraph", specifier = ">=0.6.0,<2" },
    { name = "shared", editable = "libs/shared" },
 ]
@@ -215,11 +215,10 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
-    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -228,21 +227,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.15"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]

 [[package]]
@@ -724,7 +711,7 @@ dependencies = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.3" },
+    { name = "langchain-core", specifier = ">=0.2.14" },
    { name = "langgraph", specifier = ">=0.6.0,<2" },
 ]

@@ -5,7 +5,7 @@ description = "Simple single-package uv example for LangGraph CLI integration te
 requires-python = ">=3.11"
 dependencies = [
    "langgraph>=0.6.0,<2",
-    "langchain-core>=1.3.3",
+    "langchain-core>=0.2.14",
 ]

 [build-system]
@@ -191,11 +191,10 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
-    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -204,21 +203,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.15"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]

 [[package]]
@@ -627,7 +614,7 @@ dependencies = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.3" },
+    { name = "langchain-core", specifier = ">=0.2.14" },
    { name = "langgraph", specifier = ">=0.6.0,<2" },
 ]

@@ -946,11 +946,10 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.3"
+version = "1.3.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch", marker = "python_full_version >= '3.11'" },
-    { name = "langchain-protocol", marker = "python_full_version >= '3.11'" },
    { name = "langsmith", marker = "python_full_version >= '3.11'" },
    { name = "packaging", marker = "python_full_version >= '3.11'" },
    { name = "pydantic", marker = "python_full_version >= '3.11'" },
@@ -959,21 +958,9 @@ dependencies = [
    { name = "typing-extensions", marker = "python_full_version >= '3.11'" },
    { name = "uuid-utils", marker = "python_full_version >= '3.11'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
+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" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1f/01/4771b7ab2af1d1aba5b710bd8f13d9225c609425214b357590a17b01be77/langchain_core-1.3.3-py3-none-any.whl", hash = "sha256:18aae8506f37da7f74398492279a7d6efcee4f8e23c4c41c7af080eeb7ef7bd1", size = 543857, upload-time = "2026-05-05T19:02:34.52Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.15"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions", marker = "python_full_version >= '3.11'" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { 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" },
 ]

 [[package]]
@@ -2320,11 +2307,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -29,9 +29,6 @@ from langgraph._internal._constants import (
 )

 DEFAULT_RECURSION_LIMIT = int(getenv("LANGGRAPH_DEFAULT_RECURSION_LIMIT", "10007"))
-DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT = int(
-    getenv("LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT", "5000")
-)


 def recast_checkpoint_ns(ns: str) -> str:
@@ -32,7 +32,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
        change in future releases. Threads written with `DeltaChannel` today
        are expected to remain readable, but the surrounding contract
        (`BaseCheckpointSaver.get_delta_channel_history`, the
-        `_DeltaSnapshot` blob shape, the `counters_since_delta_snapshot`
+        `_DeltaSnapshot` blob shape, the `delta_updates_since_snapshot`
        metadata field) is not yet stable.

    The reducer receives the current accumulated value and a batch of writes
@@ -47,12 +47,9 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
    This lets LangGraph replay checkpointed writes in larger batches than they
    were originally produced without changing reconstructed state.

-    Snapshot cadence is driven by two counters: per-channel update count and
-    total supersteps since last snapshot. `create_checkpoint` writes a full
-    `_DeltaSnapshot` blob when EITHER the update count reaches
-    `snapshot_frequency` OR the supersteps count reaches the system-wide
-    `DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT` bound (default 5000), bounding
-    replay depth even for channels that stop receiving writes.
+    Snapshot cadence is driven by per-channel update count. `create_checkpoint`
+    writes a full `_DeltaSnapshot` blob every `snapshot_frequency` updates to
+    this channel, bounding replay depth.

    Parameters:
        reducer: `(state, list[writes]) -> new_state`. Must be deterministic
@@ -6,7 +6,7 @@ import typing
 import warnings
 from collections import defaultdict
 from collections.abc import Awaitable, Callable, Hashable, Sequence
-from dataclasses import dataclass, is_dataclass
+from dataclasses import is_dataclass
 from datetime import timedelta
 from functools import partial
 from inspect import isclass, isfunction, ismethod, signature
@@ -95,17 +95,6 @@ __all__ = ("StateGraph", "CompiledStateGraph")
 logger = logging.getLogger(__name__)

 _CHANNEL_BRANCH_TO = "branch:to:{}"
-_DEFAULT_ERROR_HANDLER_NODE = "__default_error_handler__"
-
-
-@dataclass(slots=True)
-class _NodeDefaults:
-    """Default node policies applied to every node at compile time."""
-
-    retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None
-    cache_policy: CachePolicy | None = None
-    error_handler: StateNode[Any, Any] | None = None
-    timeout: TimeoutPolicy | None = None


 def _warn_invalid_state_schema(schema: type[Any] | Any) -> None:
@@ -262,77 +251,10 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
        self.output_schema = cast(type[OutputT], output_schema or state_schema)
        self.context_schema = context_schema

-        self._node_defaults: _NodeDefaults = _NodeDefaults()
-
        self._add_schema(self.state_schema)
        self._add_schema(self.input_schema, allow_managed=False)
        self._add_schema(self.output_schema, allow_managed=False)

-    def set_node_defaults(
-        self,
-        *,
-        retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
-        cache_policy: CachePolicy | None = None,
-        error_handler: StateNode[Any, ContextT] | None = None,
-        timeout: float | timedelta | TimeoutPolicy | None = None,
-    ) -> Self:
-        """Set default node policies that apply to every node in this graph.
-
-        Per-node values passed to `add_node` always take precedence over these
-        defaults. Defaults are applied at `compile()` time. Policies set here
-        are **not** inherited by subgraphs.
-
-        `retry_policy` and `timeout` defaults apply to **all** nodes,
-        including error-handler nodes. `cache_policy` and `error_handler`
-        defaults only apply to regular nodes -- caching error-handler results
-        is unsafe, and handlers must never catch themselves.
-
-        Args:
-            retry_policy: Default retry policy for nodes that don't specify
-                their own via `add_node(..., retry_policy=...)`. Also applies
-                to error-handler nodes.
-            cache_policy: Default cache policy for nodes that don't specify
-                their own via `add_node(..., cache_policy=...)`. Does **not**
-                apply to error-handler nodes.
-            error_handler: Default error handler invoked when any regular node
-                raises and does not have its own `error_handler` set via
-                `add_node`. The handler is **not** invoked when an
-                error-handler node itself raises -- handler failures fail the
-                run.
-            timeout: Default timeout policy for nodes that don't specify their
-                own via `add_node(..., timeout=...)`. Also applies to
-                error-handler nodes. Accepts a `TimeoutPolicy`, a number of
-                seconds (`float`), or a `timedelta`.
-
-        Returns:
-            Self: The builder instance, for chaining.
-
-        Example:
-            ```python
-            graph = (
-                StateGraph(State)
-                .set_node_defaults(
-                    retry_policy=RetryPolicy(max_attempts=3),
-                    error_handler=my_fallback_handler,
-                )
-                .add_node("a", node_a)
-                .add_node("b", node_b, retry_policy=custom_retry)  # overrides default
-                .add_edge(START, "a")
-                .compile()
-            )
-            ```
-        """
-        defaults = self._node_defaults
-        if retry_policy is not None:
-            defaults.retry_policy = retry_policy
-        if cache_policy is not None:
-            defaults.cache_policy = cache_policy
-        if error_handler is not None:
-            defaults.error_handler = error_handler
-        if timeout is not None:
-            defaults.timeout = coerce_timeout_policy(timeout)
-        return self
-
    @property
    def _all_edges(self) -> set[tuple[str, str]]:
        return self.edges | {
@@ -1271,63 +1193,10 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                key for key, val in self.channels.items() if not is_managed_value(val)
            ]
        )
-        # Apply builder defaults to node specs. Per-node values always win.
-        # Error-handler routing and cache_policy are only assigned to regular
-        # nodes. Retry and timeout defaults also apply to error-handler nodes.
-        defaults = self._node_defaults
-        default_handler_name: str | None = None
-        if defaults.error_handler is not None:
-            if _DEFAULT_ERROR_HANDLER_NODE in self.nodes:
-                raise ValueError(
-                    f"Auto-generated default error handler node "
-                    f"`{_DEFAULT_ERROR_HANDLER_NODE}` already exists."
-                )
-            default_handler_name = _DEFAULT_ERROR_HANDLER_NODE
-            self.nodes[default_handler_name] = StateNodeSpec[Any, ContextT](
-                coerce_to_runnable(
-                    defaults.error_handler,  # type: ignore[arg-type]
-                    name=default_handler_name,
-                    trace=False,
-                ),
-                metadata=None,
-                input_schema=self.state_schema,
-                retry_policy=None,
-                cache_policy=None,
-                is_error_handler=True,
-            )
-
-        # Apply builder defaults to node specs. Per-node values always win.
-        for spec in self.nodes.values():
-            # error_handler: regular nodes only — handlers must never
-            # catch themselves or other handlers.
-            if (
-                not spec.is_error_handler
-                and default_handler_name is not None
-                and spec.error_handler_node is None
-            ):
-                spec.error_handler_node = default_handler_name
-            # retry: all nodes — handlers should be retried on transient
-            # failures just like regular nodes.
-            if defaults.retry_policy is not None and spec.retry_policy is None:
-                spec.retry_policy = defaults.retry_policy
-            # cache: regular nodes only — caching an error-handler result
-            # is unsafe because the input (failed-node state) may differ
-            # across failures even when the cache key matches.
-            if (
-                not spec.is_error_handler
-                and defaults.cache_policy is not None
-                and spec.cache_policy is None
-            ):
-                spec.cache_policy = defaults.cache_policy
-            # timeout: all nodes — a stuck handler should be cancelled the
-            # same way a stuck regular node would be.
-            if defaults.timeout is not None and spec.timeout is None:
-                spec.timeout = defaults.timeout
-
        node_error_handler_map = {
            node_name: spec.error_handler_node
            for node_name, spec in self.nodes.items()
-            if not spec.is_error_handler and spec.error_handler_node is not None
+            if spec.error_handler_node is not None
        }

        compiled = CompiledStateGraph[StateT, ContextT, InputT, OutputT](
@@ -8,11 +8,11 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
    Checkpoint,
+    CheckpointMetadata,
 )
 from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.types import _DeltaSnapshot

-from langgraph._internal._config import DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.delta import DeltaChannel
@@ -36,26 +36,21 @@ def empty_checkpoint() -> Checkpoint:

 def delta_channels_to_snapshot(
    channels: Mapping[str, BaseChannel],
-    counters_since_delta_snapshot: Mapping[str, tuple[int, int]],
+    counts: Mapping[str, int],
 ) -> set[str]:
    """Return the set of DeltaChannel names that should snapshot now.

-    A channel snapshots when EITHER its accumulated update count reaches
-    `snapshot_frequency` OR the total supersteps since its last snapshot
-    reaches `DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT`. This is a pure
+    A channel snapshots when its accumulated update count (since the last
+    snapshot) reaches or exceeds `snapshot_frequency`. This is a pure
    predicate — no mutation.
    """
-    result: set[str] = set()
-    for name, ch in channels.items():
-        if not isinstance(ch, DeltaChannel) or not ch.is_available():
-            continue
-        updates, supersteps = counters_since_delta_snapshot.get(name, (0, 0))
-        if (
-            updates >= ch.snapshot_frequency
-            or supersteps >= DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT
-        ):
-            result.add(name)
-    return result
+    return {
+        name
+        for name, ch in channels.items()
+        if isinstance(ch, DeltaChannel)
+        and ch.is_available()
+        and counts.get(name, 0) >= ch.snapshot_frequency
+    }


 def create_checkpoint(
@@ -74,7 +69,7 @@ def create_checkpoint(
    is written into `channel_values[k]`. Other delta channels are omitted
    from `channel_values` — the ancestor walk reconstructs their state
    from `checkpoint_writes`. Callers compute the set via
-    `delta_channels_to_snapshot(channels, counters)`; defaults to empty
+    `delta_channels_to_snapshot(channels, counts)`; defaults to empty
    (no snapshots) when not provided.
    """
    ts = datetime.now(timezone.utc).isoformat()
@@ -236,3 +231,17 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
        versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
        updated_channels=checkpoint.get("updated_channels", None),
    )
+
+
+def read_delta_updates_since_snapshot(
+    metadata: CheckpointMetadata | None,
+) -> dict[str, int]:
+    """Read the per-channel update counter from checkpoint metadata.
+
+    Returns an empty dict for missing/None metadata; the dict is
+    `total=False` on `CheckpointMetadata`, so absence means "no prior
+    delta-channel activity tracked."
+    """
+    if not metadata:
+        return {}
+    return dict(metadata.get("delta_updates_since_snapshot", {}) or {})
@@ -203,12 +203,6 @@ class PregelLoop:
    # `__enter__`; stays `None` only when no checkpointer.
    _delta_write_futs: list[Any] | None = None

-    # Same pattern as `_delta_write_futs` but for error-handler writes.
-    # When `put_writes` persists an ERROR_SOURCE_NODE marker, the future is
-    # appended here.  `schedule_error_handler` / `aschedule_error_handler`
-    # drain this list so the write is durable before the handler starts.
-    _error_handler_write_futs: list[Any] | None = None
-
    # Exit-mode accumulator: every delta-channel write produced during this
    # run (input writes from `_first` + per-superstep writes captured in
    # `after_tick`). At exit, `_put_exit_delta_writes` filters out channels
@@ -480,13 +474,6 @@ class PregelLoop:
                isinstance(self.specs.get(c), DeltaChannel) for c, _ in writes_to_save
            ):
                self._delta_write_futs.append(fut)
-            # ERROR_SOURCE_NODE is only appended by commit() when the task
-            # has an error handler (_should_route_to_error_handler), so this
-            # check naturally limits future collection to those tasks.
-            if self._error_handler_write_futs is not None and any(
-                c == ERROR_SOURCE_NODE for c, _ in writes
-            ):
-                self._error_handler_write_futs.append(fut)
        # output writes
        if hasattr(self, "tasks"):
            self.output_writes(task_id, writes)
@@ -566,7 +553,7 @@ class PregelLoop:
            self.tasks[pushed.id] = pushed
            # match any pending writes to the new task
            if not self.is_replaying:
-                self._reapply_writes_to_succeeded_nodes({pushed.id: pushed})
+                self._match_writes({pushed.id: pushed})
            # return the new task, to be started if not run before
            return pushed

@@ -644,8 +631,7 @@ class PregelLoop:

        # if there are pending writes from a previous loop, apply them
        if not self.is_replaying and self.checkpoint_pending_writes:
-            self._reapply_writes_to_succeeded_nodes(self.tasks)
-            self._resume_error_handlers_if_applicable()
+            self._match_writes(self.tasks)

        # before execution, check if we should interrupt
        if self.interrupt_before and should_interrupt(
@@ -712,88 +698,13 @@ class PregelLoop:

    # private

-    def _reapply_writes_to_succeeded_nodes(
-        self, tasks: Mapping[str, PregelExecutableTask]
-    ) -> None:
-        """Restore successful channel writes from checkpoint to in-memory tasks.
-
-        Skips control signals (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME)
-        so that failed/interrupted tasks remain with empty writes and will be
-        re-executed (or routed to error handlers) by the runner.
-        """
+    def _match_writes(self, tasks: Mapping[str, PregelExecutableTask]) -> None:
        for tid, k, v in self.checkpoint_pending_writes:
            if k in (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME):
                continue
            if task := tasks.get(tid):
                task.writes.append((k, v))

-    def _resume_error_handlers_if_applicable(self) -> None:
-        """On resume, schedule error handlers for tasks that failed in a prior run.
-
-        Called right after ``_reapply_writes_to_succeeded_nodes`` during ``tick()``.
-        At that point, ``_reapply_writes_to_succeeded_nodes`` has already skipped
-        ERROR / ERROR_SOURCE_NODE writes, so a previously-failed task still has
-        empty ``writes``.  Without intervention the runner (which executes only
-        tasks where ``not t.writes``) would re-run the original node.
-
-        This method prevents that re-execution for nodes that have an error
-        handler:
-
-        1. Scan ``checkpoint_pending_writes`` for ERROR_SOURCE_NODE markers
-           persisted by a prior ``commit()``.  Each marker means "this task
-           already failed and was routed to an error handler".
-        2. For each such task, write ``(ERROR, error)`` into ``task.writes``
-           so the task is no longer empty — the runner will skip it.
-        3. Prepare a fresh error-handler task and add it to ``self.tasks``.
-           Because the handler task starts with empty ``writes``, the runner
-           will pick it up and execute it.
-        """
-        # Phase 1: collect task-ids that have ERROR_SOURCE_NODE + ERROR pairs.
-        failed: dict[str, BaseException] = {}
-        for tid, chan, val in self.checkpoint_pending_writes:
-            if chan == ERROR_SOURCE_NODE:
-                error = next(
-                    (
-                        v
-                        for t, c, v in self.checkpoint_pending_writes
-                        if t == tid and c == ERROR
-                    ),
-                    None,
-                )
-                if error is not None:
-                    failed[tid] = error
-        # Phase 2: mark originals as done, schedule handler tasks.
-        for task_id, error in failed.items():
-            task = self.tasks.get(task_id)
-            if task is None:
-                continue
-            handler_node = self.nodes[task.name].error_handler_node
-            if not handler_node:
-                continue
-            # Non-empty writes → runner's `not t.writes` filter skips this task.
-            task.writes.append((ERROR, error))
-            # The handler task starts with empty writes → runner will execute it.
-            handler_task = prepare_node_error_handler_task(
-                task,
-                handler_node_name=handler_node,
-                failed_error=error,
-                checkpoint=self.checkpoint,
-                pending_writes=self.checkpoint_pending_writes,
-                processes=self.nodes,
-                channels=self.channels,
-                managed=self.managed,
-                config=task.config,
-                step=self.step,
-                stop=self.stop,
-                store=self.store,
-                checkpointer=self.checkpointer,
-                manager=self.manager,
-                retry_policy=self.retry_policy,
-                cache_policy=self.cache_policy,
-            )
-            if handler_task is not None:
-                self.tasks[handler_task.id] = handler_task
-
    def _pending_interrupts(self) -> set[str]:
        """Return the set of interrupt ids that are pending without corresponding resume values."""
        # mapping of task ids to interrupt ids
@@ -1067,41 +978,35 @@ class PregelLoop:
        if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
            # checkpoint already saved
            return
-        # Per-delta-channel counter bookkeeping.
-        #
-        # Each delta channel tracks a (updates, supersteps) tuple:
-        # - `updates` increments only when the channel is written this step.
-        # - `supersteps` increments every superstep regardless.
+        # Per-delta-channel update bookkeeping.
        #
        # `_put_checkpoint` is called once per superstep with a fresh
        # metadata dict (source="input"|"loop"|"fork") — those are the
-        # intermediate calls that bump counters. In exit mode,
+        # intermediate calls that bump the count by +1 for each delta
+        # channel touched that step. In exit mode,
        # `_suppress_interrupt`(will rename to _on_loop_exit soon)
        # additionally calls `_put_checkpoint(self.checkpoint_metadata)` AT
        # EXIT to commit the final checkpoint — this runs *after* the last
        # intermediate call already counted the last superstep. So the
        # exit call must NOT bump again or it would double-count the last
-        # superstep.
+        # superstep. (Sync/async durability does not call `_put_checkpoint`
+        # at exit, so the issue only surfaces in exit mode. force_delta_snapshot
+        # used to mask this latent bug by resetting every count to 0.)
        if not exiting:
-            prev_counters = dict(
-                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
+            prev_counts = dict(
+                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
            )
-            new_counters: dict[str, tuple[int, int]] = {}
-            updated = self.updated_channels or set()
-            for ch_name, ch in self.channels.items():
-                if not isinstance(ch, DeltaChannel):
-                    continue
-                u, s = prev_counters.get(ch_name, (0, 0))
-                s += 1
-                if ch_name in updated:
-                    u += 1
-                new_counters[ch_name] = (u, s)
+            new_counts = dict(prev_counts)
+            if self.updated_channels:
+                for ch_name in self.updated_channels:
+                    if isinstance(self.channels.get(ch_name), DeltaChannel):
+                        new_counts[ch_name] = new_counts.get(ch_name, 0) + 1
            metadata["step"] = self.step
            metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
            self.checkpoint_metadata = metadata
        else:
-            new_counters = dict(
-                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
+            new_counts = dict(
+                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
            )
        # do checkpoint?
        do_checkpoint = self._checkpointer_put_after_previous is not None and (
@@ -1109,7 +1014,7 @@ class PregelLoop:
        )
        # create new checkpoint
        channels_to_snapshot = (
-            delta_channels_to_snapshot(self.channels, new_counters)
+            delta_channels_to_snapshot(self.channels, new_counts)
            if do_checkpoint
            else set()
        )
@@ -1125,12 +1030,11 @@ class PregelLoop:
            channels_to_snapshot=channels_to_snapshot,
        )
        for k in channels_to_snapshot:
-            new_counters[k] = (0, 0)
-        non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
-        if non_zero:
-            self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
-        elif "counters_since_delta_snapshot" in self.checkpoint_metadata:
-            del self.checkpoint_metadata["counters_since_delta_snapshot"]
+            new_counts[k] = 0
+        if new_counts:
+            self.checkpoint_metadata["delta_updates_since_snapshot"] = new_counts
+        elif "delta_updates_since_snapshot" in self.checkpoint_metadata:
+            del self.checkpoint_metadata["delta_updates_since_snapshot"]
        # sanitize TASK channel in the checkpoint before saving (durability=="exit")
        if TASKS in self.checkpoint["channel_values"] and any(
            isinstance(channel, UntrackedValue) for channel in self.channels.values()
@@ -1205,10 +1109,8 @@ class PregelLoop:
        ):
            return

-        counters = dict(
-            self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
-        )
-        channels_to_snapshot = delta_channels_to_snapshot(self.channels, counters)
+        counts = self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+        channels_to_snapshot = delta_channels_to_snapshot(self.channels, counts)

        pending = [
            (step, tid, ch, v)
@@ -1543,10 +1445,12 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
        handler_node = self.nodes[failed_task.name].error_handler_node
        if not handler_node:
            return None
-        # ensure error + ERROR_SOURCE_NODE writes are durable before handler runs
-        if self._error_handler_write_futs:
-            futs, self._error_handler_write_futs = self._error_handler_write_futs, []
-            concurrent.futures.wait(futs)
+        writes = list(failed_task.writes)
+        writes.append((ERROR_SOURCE_NODE, failed_task.name))
+        self.put_writes(
+            failed_task.id,
+            writes,
+        )
        handler_task = prepare_node_error_handler_task(
            failed_task,
            handler_node_name=handler_node,
@@ -1569,7 +1473,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            return None
        self.tasks[handler_task.id] = handler_task
        if not self.is_replaying:
-            self._reapply_writes_to_succeeded_nodes({handler_task.id: handler_task})
+            self._match_writes({handler_task.id: handler_task})
        for task in self.match_cached_writes():
            self.output_writes(task.id, task.writes, cached=True)
        return handler_task
@@ -1651,7 +1555,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            else []
        )
        self._delta_write_futs = []
-        self._error_handler_write_futs = []
        self._exit_delta_writes = (
            [] if self.durability == "exit" and self.checkpointer is not None else None
        )
@@ -1797,10 +1700,12 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
        handler_node = self.nodes[failed_task.name].error_handler_node
        if not handler_node:
            return None
-        # ensure error + ERROR_SOURCE_NODE writes are durable before handler runs
-        if self._error_handler_write_futs:
-            futs, self._error_handler_write_futs = self._error_handler_write_futs, []
-            await asyncio.gather(*futs)
+        writes = list(failed_task.writes)
+        writes.append((ERROR_SOURCE_NODE, failed_task.name))
+        self.put_writes(
+            failed_task.id,
+            writes,
+        )
        handler_task = prepare_node_error_handler_task(
            failed_task,
            handler_node_name=handler_node,
@@ -1823,7 +1728,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            return None
        self.tasks[handler_task.id] = handler_task
        if not self.is_replaying:
-            self._reapply_writes_to_succeeded_nodes({handler_task.id: handler_task})
+            self._match_writes({handler_task.id: handler_task})
        for task in await self.amatch_cached_writes():
            self.output_writes(task.id, task.writes, cached=True)
        return handler_task
@@ -1908,7 +1813,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            else []
        )
        self._delta_write_futs = []
-        self._error_handler_write_futs = []
        self._exit_delta_writes = (
            [] if self.durability == "exit" and self.checkpointer is not None else None
        )
@@ -31,7 +31,6 @@ from langgraph._internal._constants import (
    CONFIG_KEY_CALL,
    CONFIG_KEY_SCRATCHPAD,
    ERROR,
-    ERROR_SOURCE_NODE,
    INTERRUPT,
    NO_WRITES,
    RESUME,
@@ -598,7 +597,7 @@ class PregelRunner:
                if self._should_route_to_error_handler(task) and not isinstance(
                    exception, GraphBubbleUp
                ):
-                    task.writes.append((ERROR_SOURCE_NODE, task.name))
+                    # Mark early in commit path; loop-side routing may happen later.
                    self._handled_exception_ids.add(id(exception))
                self.put_writes()(task.id, task.writes)  # type: ignore[misc]
        else:
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph"
-version = "1.2.0"
+version = "1.2.0a7"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -24,10 +24,10 @@ classifiers = [
    'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.4.0,<2",
-    "langgraph-checkpoint>=4.1.0,<5.0.0",
+    "langchain-core>=1.4.0a2,<2",
+    "langgraph-checkpoint>=4.1.0a4,<5.0.0",
    "langgraph-sdk>=0.3.0,<0.4.0",
-    "langgraph-prebuilt>=1.1.0,<1.2.0",
+    "langgraph-prebuilt>=1.1.0a2,<1.2.0",
    "xxhash>=3.5.0",
    "pydantic>=2.7.4",
 ]
@@ -160,8 +160,8 @@ async def test_exit_resumed_run_sub_freq() -> None:


 async def test_exit_count_parity_sync_vs_exit() -> None:
-    """Sync and exit durability produce the same update count in
-    counters_since_delta_snapshot after an equivalent run."""
+    """Sync and exit durability produce the same delta_updates_since_snapshot
+    after an equivalent run."""
    for durability in ("sync", "exit"):
        saver = InMemorySaver()
        graph = _build_graph(saver)
@@ -175,13 +175,9 @@ async def test_exit_count_parity_sync_vs_exit() -> None:

        head = saver.get_tuple(config)
        assert head is not None
-        counters = head.metadata.get("counters_since_delta_snapshot", {})
-        updates, supersteps = counters.get("messages", (0, 0))
-        assert updates == 2, (
-            f"durability={durability}: expected updates=2, got {updates}"
-        )
-        assert supersteps >= 2, (
-            f"durability={durability}: expected supersteps>=2, got {supersteps}"
+        counts = head.metadata.get("delta_updates_since_snapshot", {})
+        assert counts.get("messages") == 2, (
+            f"durability={durability}: expected count=2, got {counts}"
        )


@@ -200,9 +196,8 @@ async def test_exit_snapshot_fires_at_frequency() -> None:
    )
    head = saver.get_tuple(config)
    assert head is not None
-    counters1 = head.metadata.get("counters_since_delta_snapshot", {})
-    updates1 = counters1.get("messages", (0, 0))[0]
-    assert updates1 == 2
+    count1 = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
+    assert count1 == 2

    graph.invoke(
        {"messages": [HumanMessage(content="m2", id="h2")]},
@@ -211,9 +206,8 @@ async def test_exit_snapshot_fires_at_frequency() -> None:
    )
    head = saver.get_tuple(config)
    assert head is not None
-    counters2 = head.metadata.get("counters_since_delta_snapshot", {})
-    updates2 = counters2.get("messages", (0, 0))[0]
-    assert updates2 == 0, f"Expected reset to 0 after snapshot, got {updates2}"
+    count2 = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
+    assert count2 == 0, f"Expected reset to 0 after snapshot, got {count2}"
    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)


@@ -290,7 +284,7 @@ async def test_exit_multi_run_replay_chain() -> None:

 async def test_exit_metadata_round_trip() -> None:
    """K=5 consecutive exit runs with snapshot_frequency=5. Verify metadata
-    counters_since_delta_snapshot increments correctly across runs."""
+    delta_updates_since_snapshot increments correctly across runs."""
    freq = 5
    saver = InMemorySaver()
    graph = _build_graph(saver, freq=freq)
@@ -304,16 +298,15 @@ async def test_exit_metadata_round_trip() -> None:
        )
        head = saver.get_tuple(config)
        assert head is not None
-        counters = head.metadata.get("counters_since_delta_snapshot", {})
-        updates = counters.get("messages", (0, 0))[0]
+        count = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
        cumulative = i * 2
        if cumulative >= freq:
-            assert updates == 0 or updates == cumulative % freq or updates < freq, (
-                f"After run {i}: updates={updates} should have reset or be partial"
+            assert count == 0 or count == cumulative % freq or count < freq, (
+                f"After run {i}: count={count} should have reset or be partial"
            )
        else:
-            assert updates == cumulative, (
-                f"After run {i}: expected {cumulative}, got {updates}"
+            assert count == cumulative, (
+                f"After run {i}: expected {cumulative}, got {count}"
            )


@@ -1,195 +0,0 @@
-"""Tests for the supersteps-since-last-snapshot bound on DeltaChannel.
-
-Validates that a delta channel which stops receiving writes is still
-force-snapshotted after DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT supersteps,
-preventing unbounded ancestor walks.
-"""
-
-from typing import Annotated, Any
-from unittest.mock import patch
-
-import pytest
-from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-from typing_extensions import TypedDict
-
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph import END, START, StateGraph
-from langgraph.pregel._checkpoint import delta_channels_to_snapshot
-
-pytestmark = pytest.mark.anyio
-
-
-def _simple_reducer(current: list, updates: list) -> list:
-    """Flatten updates into current list (each update is itself a list)."""
-    result = list(current)
-    for u in updates:
-        if isinstance(u, list):
-            result.extend(u)
-        else:
-            result.append(u)
-    return result
-
-
-def _build_two_channel_graph(
-    checkpointer: InMemorySaver,
-    *,
-    freq_a: int = 10_000,
-    freq_b: int = 10_000,
-    n_loops: int = 1,
-) -> Any:
-    """Graph with two delta channels A and B.
-
-    The node only writes to channel A; B is never written by the node.
-    `n_loops` controls how many supersteps the graph runs (via chained nodes).
-    """
-    ch_a = DeltaChannel(_simple_reducer, list, snapshot_frequency=freq_a)
-    ch_b = DeltaChannel(_simple_reducer, list, snapshot_frequency=freq_b)
-    State = TypedDict(  # noqa: UP013
-        "State",
-        {"a": Annotated[list, ch_a], "b": Annotated[list, ch_b]},
-    )  # type: ignore[call-overload]
-
-    builder = StateGraph(State)
-
-    for i in range(n_loops):
-        name = f"step_{i}"
-
-        def node_fn(state: dict, _i: int = i) -> dict:
-            return {"a": [f"a-val-{_i}"]}
-
-        builder.add_node(name, node_fn)
-        if i == 0:
-            builder.add_edge(START, name)
-        else:
-            builder.add_edge(f"step_{i - 1}", name)
-        if i == n_loops - 1:
-            builder.add_edge(name, END)
-
-    return builder.compile(checkpointer=checkpointer)
-
-
-async def test_forced_snapshot_single_run() -> None:
-    """A single invoke with enough supersteps triggers snapshot on the
-    unwritten channel B via the supersteps bound."""
-    max_ss = 3
-    with patch(
-        "langgraph.pregel._checkpoint.DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT",
-        max_ss,
-    ):
-        saver = InMemorySaver()
-        graph = _build_two_channel_graph(saver, n_loops=4)
-        config = {"configurable": {"thread_id": "single-run-ss"}}
-
-        graph.invoke({"a": ["seed-a"], "b": ["seed-b"]}, config)
-
-        head = saver.get_tuple(config)
-        assert head is not None
-        assert isinstance(head.checkpoint["channel_values"].get("b"), _DeltaSnapshot), (
-            "Channel B should have been force-snapshotted via supersteps bound"
-        )
-
-        state = graph.get_state(config)
-        assert state.values["b"] == ["seed-b"]
-        assert "seed-a" in state.values["a"]
-
-
-async def test_forced_snapshot_accumulates_across_runs() -> None:
-    """Supersteps counter for an unwritten channel persists across separate
-    invoke() calls. After enough runs, the channel is force-snapshotted."""
-    max_ss = 5
-    with patch(
-        "langgraph.pregel._checkpoint.DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT",
-        max_ss,
-    ):
-        saver = InMemorySaver()
-        graph = _build_two_channel_graph(saver, n_loops=1)
-        config = {"configurable": {"thread_id": "multi-run-ss"}}
-
-        graph.invoke({"a": ["init-a"], "b": ["init-b"]}, config)
-
-        for i in range(1, 6):
-            graph.invoke({"a": [f"run-{i}"]}, config)
-
-            head = saver.get_tuple(config)
-            assert head is not None
-            counters = head.metadata.get("counters_since_delta_snapshot", {})
-            b_counters = counters.get("b", (0, 0))
-
-            if b_counters == (0, 0):
-                assert isinstance(
-                    head.checkpoint["channel_values"].get("b"), _DeltaSnapshot
-                ), f"Run {i}: counter reset but no snapshot blob for B"
-                break
-        else:
-            pytest.fail("Channel B was never force-snapshotted after multiple runs")
-
-        state = graph.get_state(config)
-        assert state.values["b"] == ["init-b"]
-        assert "init-a" in state.values["a"]
-
-
-async def test_predicate_fires_on_supersteps_overflow() -> None:
-    """Unit test: delta_channels_to_snapshot fires when supersteps >= MAX
-    even when updates == 0."""
-    ch = DeltaChannel(_simple_reducer, list, snapshot_frequency=10_000)
-    ch.key = "x"
-    ch_instance = ch.from_checkpoint(None)
-
-    channels = {"x": ch_instance}
-    counters: dict[str, tuple[int, int]] = {"x": (0, 5000)}
-
-    result = delta_channels_to_snapshot(channels, counters)
-    assert "x" in result
-
-    counters_below: dict[str, tuple[int, int]] = {"x": (0, 4999)}
-    result2 = delta_channels_to_snapshot(channels, counters_below)
-    assert "x" not in result2
-
-
-async def test_counter_reset_after_supersteps_snapshot() -> None:
-    """After the supersteps bound triggers a snapshot, the counters for
-    that channel reset. Verify by using a bound higher than one run's
-    supersteps so we can see the counter in an intermediate state."""
-    max_ss = 15
-    with patch(
-        "langgraph.pregel._checkpoint.DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT",
-        max_ss,
-    ):
-        saver = InMemorySaver()
-        graph = _build_two_channel_graph(saver, n_loops=4)
-        config = {"configurable": {"thread_id": "counter-reset"}}
-
-        graph.invoke({"a": ["seed-a"], "b": ["seed-b"]}, config)
-
-        head = saver.get_tuple(config)
-        assert head is not None
-        counters = head.metadata.get("counters_since_delta_snapshot", {})
-        b_counters = counters.get("b", (0, 0))
-        run1_supersteps = b_counters[1]
-        assert run1_supersteps > 0, "Should have some supersteps"
-        assert b_counters[0] == 1, "B written once (input step)"
-
-        graph.invoke({"a": ["more-a"]}, config)
-        head2 = saver.get_tuple(config)
-        assert head2 is not None
-        counters2 = head2.metadata.get("counters_since_delta_snapshot", {})
-        b_counters2 = counters2.get("b", (0, 0))
-        run2_supersteps = b_counters2[1]
-        assert run2_supersteps > run1_supersteps, "Supersteps should accumulate"
-        assert b_counters2[0] == 1, "B written once total (only original input)"
-
-        graph.invoke({"a": ["even-more"]}, config)
-        head3 = saver.get_tuple(config)
-        assert head3 is not None
-        assert isinstance(
-            head3.checkpoint["channel_values"].get("b"), _DeltaSnapshot
-        ), "B should have snapshotted at supersteps >= max_ss"
-        counters3 = head3.metadata.get("counters_since_delta_snapshot", {})
-        b_counters3 = counters3.get("b", (0, 0))
-        assert b_counters3[1] < max_ss, (
-            f"After snapshot, supersteps should have reset, got {b_counters3}"
-        )
-
-        state = graph.get_state(config)
-        assert state.values["b"] == ["seed-b"]
@@ -15,7 +15,7 @@ from langchain_core.callbacks import AsyncCallbackManagerForLLMRun, BaseCallback
 from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
 from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage, HumanMessage
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
-from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnableParallel
+from langchain_core.runnables import RunnableLambda, RunnableParallel
 from langgraph.checkpoint.memory import InMemorySaver, MemorySaver
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from typing_extensions import TypedDict
@@ -2280,525 +2280,3 @@ def test_node_without_error_handler_still_fails_run():

    with pytest.raises(ValueError, match="no handler"):
        graph.invoke({"foo": ""})
-
-
-# ---------------------------------------------------------------------------
-# set_node_defaults()
-# ---------------------------------------------------------------------------
-
-
-def test_set_node_defaults_error_handler_catches_all_nodes():
-    class State(TypedDict):
-        route: str
-        foo: Annotated[list[str], operator.add]
-
-    def route_node(state: State) -> Command:
-        return Command(goto=state["route"])
-
-    def fail_a(state: State) -> State:
-        raise RuntimeError("a failed")
-
-    def fail_b(state: State) -> State:
-        raise RuntimeError("b failed")
-
-    captured: dict[str, list[str]] = {"nodes": []}
-
-    def default_handler(state: State, error: NodeError) -> State:
-        captured["nodes"].append(error.node)
-        return {"foo": [f"handled_{error.node}"]}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=default_handler)
-        .add_node("route_node", route_node)
-        .add_node("fail_a", fail_a)
-        .add_node("fail_b", fail_b)
-        .add_edge(START, "route_node")
-        .add_conditional_edges(
-            "route_node", lambda s: s["route"], path_map=["fail_a", "fail_b"]
-        )
-        .compile()
-    )
-
-    result_a = graph.invoke({"route": "fail_a", "foo": []})
-    result_b = graph.invoke({"route": "fail_b", "foo": []})
-    assert result_a["foo"] == ["handled_fail_a"]
-    assert result_b["foo"] == ["handled_fail_b"]
-    assert "fail_a" in captured["nodes"]
-    assert "fail_b" in captured["nodes"]
-
-
-def test_set_node_defaults_error_handler_overridden_by_node_handler():
-    class State(TypedDict):
-        route: str
-        foo: Annotated[list[str], operator.add]
-
-    def route_node(state: State) -> Command:
-        return Command(goto=state["route"])
-
-    def fail_a(state: State) -> State:
-        raise RuntimeError("a failed")
-
-    def fail_b(state: State) -> State:
-        raise RuntimeError("b failed")
-
-    captured: dict[str, list[str]] = {"handler": []}
-
-    def node_handler(state: State, error: NodeError) -> State:
-        captured["handler"].append(f"node:{error.node}")
-        return {"foo": [f"node_handled_{error.node}"]}
-
-    def default_handler(state: State, error: NodeError) -> State:
-        captured["handler"].append(f"default:{error.node}")
-        return {"foo": [f"default_handled_{error.node}"]}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=default_handler)
-        .add_node("route_node", route_node)
-        .add_node("fail_a", fail_a, error_handler=node_handler)
-        .add_node("fail_b", fail_b)
-        .add_edge(START, "route_node")
-        .add_conditional_edges(
-            "route_node", lambda s: s["route"], path_map=["fail_a", "fail_b"]
-        )
-        .compile()
-    )
-
-    result_a = graph.invoke({"route": "fail_a", "foo": []})
-    assert result_a["foo"] == ["node_handled_fail_a"]
-    assert "node:fail_a" in captured["handler"]
-    assert "default:fail_a" not in captured["handler"]
-
-    result_b = graph.invoke({"route": "fail_b", "foo": []})
-    assert result_b["foo"] == ["default_handled_fail_b"]
-    assert "default:fail_b" in captured["handler"]
-
-
-def test_set_node_defaults_error_handler_skips_per_node_handler_nodes():
-    """If a per-node error handler itself raises, the default handler must NOT
-    catch it -- the run should fail."""
-
-    class State(TypedDict):
-        foo: str
-
-    def always_failing(state: State) -> State:
-        raise RuntimeError("node boom")
-
-    def broken_handler(state: State, error: NodeError) -> State:
-        raise RuntimeError("handler boom")
-
-    def default_handler(state: State, error: NodeError) -> State:
-        return {"foo": "default recovered"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=default_handler)
-        .add_node("always_failing", always_failing, error_handler=broken_handler)
-        .add_edge(START, "always_failing")
-        .compile()
-    )
-
-    with pytest.raises(RuntimeError, match="handler boom"):
-        graph.invoke({"foo": ""})
-
-
-def test_set_node_defaults_error_handler_failure_fails_run():
-    """When the default handler itself raises, the run fails (no infinite
-    recursion, no double-routing)."""
-
-    class State(TypedDict):
-        foo: str
-
-    def always_failing(state: State) -> State:
-        raise RuntimeError("node boom")
-
-    def broken_default_handler(state: State, error: NodeError) -> State:
-        raise RuntimeError("default handler boom")
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=broken_default_handler)
-        .add_node("always_failing", always_failing)
-        .add_edge(START, "always_failing")
-        .compile()
-    )
-
-    with pytest.raises(RuntimeError, match="default handler boom"):
-        graph.invoke({"foo": ""})
-
-
-def test_set_node_defaults_error_handler_receives_runnable_config():
-    class State(TypedDict):
-        foo: str
-
-    def always_failing(state: State) -> State:
-        raise RuntimeError("boom")
-
-    captured: dict[str, Any] = {}
-
-    def default_handler(
-        state: State, error: NodeError, config: RunnableConfig
-    ) -> State:
-        captured["thread_id"] = config["configurable"].get("thread_id")
-        return {"foo": "handled"}
-
-    checkpointer = MemorySaver()
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=default_handler)
-        .add_node("always_failing", always_failing)
-        .add_edge(START, "always_failing")
-        .compile(checkpointer=checkpointer)
-    )
-
-    thread_id = str(uuid4())
-    result = graph.invoke(
-        {"foo": ""}, config={"configurable": {"thread_id": thread_id}}
-    )
-    assert result["foo"] == "handled"
-    assert captured["thread_id"] == thread_id
-
-
-def test_set_node_defaults_error_handler_collides_with_user_node():
-    class State(TypedDict):
-        foo: str
-
-    def default_handler(state: State, error: NodeError) -> State:
-        return {"foo": "handled"}
-
-    builder = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=default_handler)
-        .add_node("__default_error_handler__", lambda s: s)
-        .add_edge(START, "__default_error_handler__")
-    )
-
-    with pytest.raises(ValueError, match="__default_error_handler__"):
-        builder.compile()
-
-
-def test_set_node_defaults_retry_policy():
-    class State(TypedDict):
-        foo: str
-
-    attempts = 0
-
-    def flaky_node(state: State) -> State:
-        nonlocal attempts
-        attempts += 1
-        if attempts < 3:
-            raise ValueError("not yet")
-        return {"foo": "ok"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(
-            retry_policy=RetryPolicy(
-                max_attempts=3, initial_interval=0.01, jitter=False, retry_on=ValueError
-            )
-        )
-        .add_node("flaky", flaky_node)
-        .add_edge(START, "flaky")
-        .compile()
-    )
-
-    with patch("time.sleep"):
-        result = graph.invoke({"foo": ""})
-
-    assert result["foo"] == "ok"
-    assert attempts == 3
-
-
-def test_set_node_defaults_retry_policy_per_node_wins():
-    class State(TypedDict):
-        foo: str
-
-    attempts = 0
-
-    def flaky_node(state: State) -> State:
-        nonlocal attempts
-        attempts += 1
-        if attempts < 2:
-            raise ValueError("not yet")
-        return {"foo": "ok"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(
-            retry_policy=RetryPolicy(
-                max_attempts=1, initial_interval=0.01, jitter=False, retry_on=ValueError
-            )
-        )
-        .add_node(
-            "flaky",
-            flaky_node,
-            retry_policy=RetryPolicy(
-                max_attempts=3,
-                initial_interval=0.01,
-                jitter=False,
-                retry_on=ValueError,
-            ),
-        )
-        .add_edge(START, "flaky")
-        .compile()
-    )
-
-    with patch("time.sleep"):
-        result = graph.invoke({"foo": ""})
-
-    assert result["foo"] == "ok"
-    assert attempts == 2
-
-
-@pytest.mark.anyio
-async def test_set_node_defaults_timeout():
-    class State(TypedDict):
-        foo: str
-
-    async def slow_node(state: State) -> State:
-        await asyncio.sleep(10)
-        return {"foo": "should-not-happen"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(timeout=TimeoutPolicy(run_timeout=0.05))
-        .add_node("slow", slow_node)
-        .add_edge(START, "slow")
-        .compile()
-    )
-
-    from langgraph.errors import NodeTimeoutError
-
-    with pytest.raises(NodeTimeoutError):
-        await graph.ainvoke({"foo": ""})
-
-
-@pytest.mark.anyio
-async def test_set_node_defaults_timeout_per_node_wins():
-    """Per-node timeout overrides the default; a generous per-node timeout
-    allows a node to complete even when the builder default is very short."""
-
-    class State(TypedDict):
-        foo: str
-
-    async def quick_node(state: State) -> State:
-        await asyncio.sleep(0.05)
-        return {"foo": "done"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(timeout=TimeoutPolicy(run_timeout=0.01))
-        .add_node("quick", quick_node, timeout=TimeoutPolicy(run_timeout=5.0))
-        .add_edge(START, "quick")
-        .compile()
-    )
-
-    result = await graph.ainvoke({"foo": ""})
-    assert result["foo"] == "done"
-
-
-def test_set_node_defaults_chaining():
-    """set_node_defaults() is chainable and can be called in any order relative to add_node."""
-
-    class State(TypedDict):
-        foo: str
-
-    def always_failing(state: State) -> State:
-        raise RuntimeError("boom")
-
-    def handler(state: State, error: NodeError) -> State:
-        return {"foo": "handled"}
-
-    graph = (
-        StateGraph(State)
-        .add_node("a", always_failing)
-        .add_edge(START, "a")
-        .set_node_defaults(
-            retry_policy=RetryPolicy(
-                max_attempts=1, initial_interval=0.01, jitter=False
-            ),
-            error_handler=handler,
-        )
-        .compile()
-    )
-
-    result = graph.invoke({"foo": ""})
-    assert result["foo"] == "handled"
-
-
-def test_set_node_defaults_combined_retry_and_error_handler():
-    """Retries are exhausted first, then the error handler runs."""
-
-    class State(TypedDict):
-        foo: str
-
-    attempts = 0
-    captured: dict[str, Any] = {}
-
-    def always_failing(state: State) -> State:
-        nonlocal attempts
-        attempts += 1
-        raise ValueError("Always fails")
-
-    def handler(state: State, error: NodeError) -> State:
-        captured["error"] = str(error.error)
-        return {"foo": "handled"}
-
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(
-            retry_policy=RetryPolicy(
-                max_attempts=2,
-                initial_interval=0.01,
-                jitter=False,
-                retry_on=ValueError,
-            ),
-            error_handler=handler,
-        )
-        .add_node("fail", always_failing)
-        .add_edge(START, "fail")
-        .compile()
-    )
-
-    with patch("time.sleep"):
-        result = graph.invoke({"foo": ""})
-
-    assert result["foo"] == "handled"
-    assert attempts == 2
-    assert captured["error"] == "Always fails"
-
-
-def test_error_handler_resumes_after_crash():
-    """If the error handler crashes, resuming should re-schedule the handler
-    (not re-execute the original failed node)."""
-
-    class State(TypedDict):
-        foo: str
-
-    call_count = {"node": 0, "handler": 0}
-    captured_errors: list[NodeError] = []
-
-    def failing_node(state: State) -> State:
-        call_count["node"] += 1
-        raise RuntimeError("boom")
-
-    handler_should_fail = [True]
-
-    def handler(state: State, error: NodeError) -> State:
-        call_count["handler"] += 1
-        captured_errors.append(error)
-        if handler_should_fail[0]:
-            raise RuntimeError("handler crash")
-        return {"foo": "recovered"}
-
-    checkpointer = MemorySaver()
-    graph = (
-        StateGraph(State)
-        .set_node_defaults(error_handler=handler)
-        .add_node("fail", failing_node)
-        .add_edge(START, "fail")
-        .compile(checkpointer=checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "t1"}}
-
-    # First invoke: node fails -> handler runs -> handler crashes -> run fails
-    with pytest.raises(RuntimeError, match="handler crash"):
-        graph.invoke({"foo": ""}, config)
-
-    assert call_count["node"] == 1
-    assert call_count["handler"] == 1
-    assert captured_errors[0].node == "fail"
-    assert isinstance(captured_errors[0].error, RuntimeError)
-    assert str(captured_errors[0].error) == "boom"
-
-    # Resume: handler should run again, NOT the original node
-    handler_should_fail[0] = False
-    result = graph.invoke(None, config)
-
-    assert result["foo"] == "recovered"
-    assert call_count["node"] == 1  # NOT re-executed
-    assert call_count["handler"] == 2  # ran again on resume
-    # on resume the error was round-tripped through the checkpointer, so it
-    # may be deserialized as a string representation rather than the original
-    # exception type — verify the node name and that the error content matches.
-    assert captured_errors[1].node == "fail"
-    assert "boom" in str(captured_errors[1].error)
-
-
-def test_error_handler_resumes_after_crash_multiple_nodes():
-    """When multiple nodes fail in the same superstep and all have error handlers:
-    - error handlers start running while other nodes may still be in-flight
-    - resuming re-schedules each handler (not re-executes the original nodes)
-    """
-
-    class State(TypedDict):
-        results: Annotated[list[str], operator.add]
-
-    call_count = {"a": 0, "b": 0, "handler_a": 0, "handler_b": 0}
-    handler_a_started = threading.Event()
-
-    def node_a(state: State) -> State:
-        call_count["a"] += 1
-        raise RuntimeError("a failed")
-
-    def node_b(state: State) -> State:
-        call_count["b"] += 1
-        # Block until handler_a has started — proves the error handler runs
-        # concurrently with in-flight nodes in the same superstep.
-        assert handler_a_started.wait(timeout=5), "handler_a never started"
-        raise RuntimeError("b failed")
-
-    handler_should_fail = [True]
-
-    def handler_a(state: State, error: NodeError) -> State:
-        call_count["handler_a"] += 1
-        assert error.node == "a"
-        assert "a failed" in str(error.error)
-        handler_a_started.set()
-        if handler_should_fail[0]:
-            raise RuntimeError("handler_a crash")
-        return {"results": [f"recovered_a:{error.node}"]}
-
-    def handler_b(state: State, error: NodeError) -> State:
-        call_count["handler_b"] += 1
-        assert error.node == "b"
-        assert "b failed" in str(error.error)
-        if handler_should_fail[0]:
-            raise RuntimeError("handler_b crash")
-        return {"results": [f"recovered_b:{error.node}"]}
-
-    checkpointer = MemorySaver()
-    graph = (
-        StateGraph(State)
-        .add_node("a", node_a, error_handler=handler_a)
-        .add_node("b", node_b, error_handler=handler_b)
-        .add_edge(START, "a")
-        .add_edge(START, "b")
-        .compile(checkpointer=checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "t1"}}
-
-    # First invoke: node_a fails immediately -> handler_a starts (sets event) ->
-    # node_b unblocks and fails -> handler_b starts -> both handlers crash
-    with pytest.raises(RuntimeError):
-        graph.invoke({"results": []}, config)
-
-    assert call_count["a"] == 1
-    assert call_count["b"] == 1
-    assert call_count["handler_a"] == 1
-    assert call_count["handler_b"] == 1
-
-    # Resume: both handlers should run again, NOT the original nodes
-    handler_should_fail[0] = False
-    handler_a_started.clear()
-    result = graph.invoke(None, config)
-
-    assert call_count["a"] == 1  # NOT re-executed
-    assert call_count["b"] == 1  # NOT re-executed
-    assert call_count["handler_a"] == 2  # ran again on resume
-    assert call_count["handler_b"] == 2  # ran again on resume
-    assert "recovered_a:a" in result["results"]
-    assert "recovered_b:b" in result["results"]
@@ -1350,7 +1350,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.4.0"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
@@ -1363,9 +1363,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/59/de/679a53472c25860837e32c0442c962fa86e95317a36460e2c9d5c91b17c2/langchain_core-1.4.0.tar.gz", hash = "sha256:1dc341eed802ed9c117c0df3923c991e5e9e226571e5725c194eeb5bd93d1a7f", size = 920260, upload-time = "2026-05-11T18:42:35.919Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0f/1a/86c38c27b81913a1c6c12448cab55defb5a1097c7dc9a4cea83f55477a2d/langchain_core-1.4.0-py3-none-any.whl", hash = "sha256:23cbbdb46e38ddd1dd5247e6167e96013eae74bea4c5949c550809970a9e565c", size = 548120, upload-time = "2026-05-11T18:42:33.992Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]

 [[package]]
@@ -1382,7 +1382,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.0"
+version = "1.2.0a7"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -1454,7 +1454,7 @@ test = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
    { name = "langgraph-checkpoint", editable = "../checkpoint" },
    { name = "langgraph-prebuilt", editable = "../prebuilt" },
    { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1563,7 +1563,7 @@ wheels = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -1611,7 +1611,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0"
+version = "3.1.0a4"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
    { name = "langgraph-checkpoint" },
@@ -1658,7 +1658,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.0"
+version = "3.1.0a1"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
    { name = "aiosqlite" },
@@ -1757,7 +1757,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.0a2"
 source = { editable = "../prebuilt" }
 dependencies = [
    { name = "langchain-core" },
@@ -2085,14 +2085,14 @@ wheels = [

 [[package]]
 name = "mistune"
-version = "3.2.1"
+version = "3.2.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/ca/84/620cc3f7e3adf6f5067e10f4dbae71295d8f9e16d5d3f9ef97c40f2f592c/mistune-3.2.1.tar.gz", hash = "sha256:7c8e5501d38bac1582e067e46c8343f17d57ea1aaa735823f3aba1fd59c88a28", size = 98003, upload-time = "2026-05-03T14:33:22.312Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/9d/55/d01f0c4b45ade6536c51170b9043db8b2ec6ddf4a35c7ea3f5f559ac935b/mistune-3.2.0.tar.gz", hash = "sha256:708487c8a8cdd99c9d90eb3ed4c3ed961246ff78ac82f03418f5183ab70e398a", size = 95467, upload-time = "2025-12-23T11:36:34.994Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/2a/7f/a946aa4f8752b37102b41e64dca18a1976ac705c3a0d1dfe74d820a02552/mistune-3.2.1-py3-none-any.whl", hash = "sha256:78cdb0ba5e938053ccf63651b352508d2efa9411dc8810bfb05f2dc5140c0048", size = 53749, upload-time = "2026-05-03T14:33:20.551Z" },
+    { url = "https://files.pythonhosted.org/packages/9b/f7/4a5e785ec9fbd65146a27b6b70b6cdc161a66f2024e4b04ac06a67f5578b/mistune-3.2.0-py3-none-any.whl", hash = "sha256:febdc629a3c78616b94393c6580551e0e34cc289987ec6c35ed3f4be42d0eee1", size = 53598, upload-time = "2025-12-23T11:36:33.211Z" },
 ]

 [[package]]
@@ -3743,11 +3743,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -6,6 +6,7 @@ from typing import (
    Any,
    Literal,
    TypeVar,
+    Union,
    cast,
    get_type_hints,
 )
@@ -125,6 +126,9 @@ Prompt = (
    | Runnable[StateSchema, LanguageModelInput]
 )

+# A single hook or a list of hooks to be composed in order.
+HookLike = Union[RunnableLike, Sequence[RunnableLike]]
+

 def _get_state_value(state: StateSchema, key: str, default: Any = None) -> Any:
    return (
@@ -134,6 +138,131 @@ def _get_state_value(state: StateSchema, key: str, default: Any = None) -> Any:
    )


+def _set_state_value(state: StateSchema, key: str, value: Any) -> None:
+    """Set a value in the state, supporting both dict and Pydantic model states."""
+    if isinstance(state, dict):
+        state[key] = value
+    else:
+        setattr(state, key, value)
+
+
+def _merge_state_update(state: StateSchema, update: dict) -> StateSchema:
+    """Return a shallow copy of *state* with *update* applied.
+
+    This is used when chaining multiple hooks: each hook receives the state
+    as it would look after all previous hooks have run, so that hooks later
+    in the chain can observe updates made by earlier ones.
+
+    Note: only simple key-level merging is performed here (no reducer logic).
+    The full reducer logic is applied by the graph engine when the final
+    combined update dict is written back to the state.
+    """
+    if isinstance(state, dict):
+        return {**state, **update}  # type: ignore[return-value]
+    else:
+        # Pydantic / dataclass – make a shallow copy and patch fields
+        try:
+            merged = state.model_copy()  # pydantic v2
+        except AttributeError:
+            merged = state.copy()  # pydantic v1 / dataclass fallback
+        for k, v in update.items():
+            setattr(merged, k, v)
+        return merged  # type: ignore[return-value]
+
+
+def _coerce_to_runnable(hook: RunnableLike) -> RunnableCallable:
+    """Wrap a plain callable into a RunnableCallable if necessary."""
+    if isinstance(hook, RunnableCallable):
+        return hook
+    if isinstance(hook, Runnable):
+        # Already a Runnable – wrap so we get a uniform interface
+        sync_fn = hook.invoke
+        async_fn = hook.ainvoke
+        return RunnableCallable(sync_fn, async_fn)
+    if inspect.iscoroutinefunction(hook):
+        return RunnableCallable(None, hook)
+    if callable(hook):
+        return RunnableCallable(hook)
+    raise TypeError(f"Expected a callable or Runnable, got {type(hook)!r}")
+
+
+def _chain_hooks(hooks: Sequence[RunnableLike]) -> RunnableCallable:
+    """Compose multiple hook callables into a single hook.
+
+    Each hook is called in order.  After each hook the returned update dict is
+    merged into a running copy of the graph state so that subsequent hooks can
+    observe the changes made by earlier ones.  The accumulated update dict
+    (union of all individual update dicts, with later hooks winning on key
+    conflicts) is returned as the final state update.
+
+    Args:
+        hooks: A sequence of :data:`RunnableLike` objects.  Each must accept
+            the graph state as its first positional argument and return a
+            ``dict`` of state updates.
+
+    Returns:
+        A :class:`~langgraph._internal._runnable.RunnableCallable` that behaves
+        like a single hook but applies all of *hooks* in sequence.
+    """
+    if not hooks:
+        raise ValueError("_chain_hooks requires at least one hook")
+    if len(hooks) == 1:
+        return _coerce_to_runnable(hooks[0])
+
+    runnables = [_coerce_to_runnable(h) for h in hooks]
+
+    def _sync_chained(state: Any, **kwargs: Any) -> dict:
+        accumulated: dict = {}
+        current_state = state
+        for runnable in runnables:
+            # Pass extra kwargs (e.g. config, store) through if the hook
+            # accepts them; RunnableCallable handles introspection.
+            update = runnable.invoke(current_state, **kwargs)
+            if update:
+                accumulated.update(update)
+                current_state = _merge_state_update(current_state, update)
+        return accumulated
+
+    async def _async_chained(state: Any, **kwargs: Any) -> dict:
+        accumulated: dict = {}
+        current_state = state
+        for runnable in runnables:
+            update = await runnable.ainvoke(current_state, **kwargs)
+            if update:
+                accumulated.update(update)
+                current_state = _merge_state_update(current_state, update)
+        return accumulated
+
+    return RunnableCallable(_sync_chained, _async_chained, name="chained_hooks")
+
+
+def _resolve_hook(hook: HookLike | None) -> RunnableLike | None:
+    """Normalise *hook* to a single ``RunnableLike`` (or ``None``).
+
+    * If *hook* is ``None`` → return ``None``.
+    * If *hook* is already a ``RunnableLike`` → return it unchanged.
+    * If *hook* is a non-empty :class:`~collections.abc.Sequence` of
+      ``RunnableLike`` → chain them with :func:`_chain_hooks`.
+    """
+    if hook is None:
+        return None
+    # A Sequence[RunnableLike] but NOT a single Runnable/callable
+    if (
+        isinstance(hook, Sequence)
+        and not isinstance(hook, str)
+        and not isinstance(hook, Runnable)
+        and not callable(hook)
+    ):
+        hooks_list: list[RunnableLike] = list(hook)
+        if not hooks_list:
+            return None
+        if len(hooks_list) == 1:
+            return hooks_list[0]
+        return _chain_hooks(hooks_list)
+    # Single hook – return as-is
+    return hook  # type: ignore[return-value]
+
+
 def _get_prompt_runnable(prompt: Prompt | None) -> Runnable:
    prompt_runnable: Runnable
    if prompt is None:
@@ -293,8 +422,8 @@ def create_react_agent(
    response_format: StructuredResponseSchema
    | tuple[str, StructuredResponseSchema]
    | None = None,
-    pre_model_hook: RunnableLike | None = None,
-    post_model_hook: RunnableLike | None = None,
+    pre_model_hook: HookLike | None = None,
+    post_model_hook: HookLike | None = None,
    state_schema: StateSchemaType | None = None,
    context_schema: type[Any] | None = None,
    checkpointer: Checkpointer | None = None,
@@ -393,10 +522,21 @@ def create_react_agent(
                The graph will make a separate call to the LLM to generate the structured response after the agent loop is finished.
                This is not the only strategy to get structured responses, see more options in [this guide](https://langchain-ai.github.io/langgraph/how-tos/react-agent-structured-output/).

-        pre_model_hook: An optional node to add before the `agent` node (i.e., the node that calls the LLM).
-            Useful for managing long message histories (e.g., message trimming, summarization, etc.).
-            Pre-model hook must be a callable or a runnable that takes in current graph state and returns a state update in the form of
-                ```python
+        pre_model_hook: An optional node (or list of nodes) to add before the
+            ``agent`` node (i.e., the node that calls the LLM). Useful for
+            managing long message histories (e.g., message trimming,
+            summarization, etc.) or for composing multiple pre-processing
+            steps.
+
+            A single hook **or a list of hooks** may be provided. When a list
+            is given the hooks are executed in order: each hook receives the
+            graph state as updated by all preceding hooks, and the union of
+            all their return dicts is applied to the graph state before the
+            agent node runs.
+
+            Each hook must be a callable or a runnable that takes the current
+            graph state and returns a state update::
+
                # At least one of `messages` or `llm_input_messages` MUST be provided
                {
                    # If provided, will UPDATE the `messages` in the state
@@ -407,27 +547,63 @@ def create_react_agent(
                    # Any other state keys that need to be propagated
                    ...
                }
-                ```

            !!! Important
-                At least one of `messages` or `llm_input_messages` MUST be provided and will be used as an input to the `agent` node.
-                The rest of the keys will be added to the graph state.
+                At least one of `messages` or `llm_input_messages` MUST be
+                provided (by at least one hook in the chain) and will be used
+                as an input to the ``agent`` node.  The rest of the keys will
+                be added to the graph state.

            !!! Warning
-                If you are returning `messages` in the pre-model hook, you should OVERWRITE the `messages` key by doing the following:
+                If you are returning `messages` in the pre-model hook, you
+                should OVERWRITE the `messages` key::
+
+                    {
+                        "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]
+                        ...
+                    }
+
+            !!! Example "Composing multiple pre-model hooks"

                ```python
-                {
-                    "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]
-                    ...
-                }
+                from langchain_core.messages import RemoveMessage
+                from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+                def trim_messages(state):
+                    # Keep only the last 10 messages
+                    return {
+                        "messages": [
+                            RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                            *state["messages"][-10:],
+                        ]
+                    }
+
+                def inject_system_prompt(state):
+                    return {
+                        "llm_input_messages": [
+                            SystemMessage("You are a helpful assistant."),
+                            *state["messages"],
+                        ]
+                    }
+
+                agent = create_react_agent(
+                    model,
+                    tools,
+                    pre_model_hook=[trim_messages, inject_system_prompt],
+                )
                ```
-        post_model_hook: An optional node to add after the `agent` node (i.e., the node that calls the LLM).
-            Useful for implementing human-in-the-loop, guardrails, validation, or other post-processing.
-            Post-model hook must be a callable or a runnable that takes in current graph state and returns a state update.
+
+        post_model_hook: An optional node (or list of nodes) to add after the
+            ``agent`` node (i.e., the node that calls the LLM). Useful for
+            implementing human-in-the-loop, guardrails, validation, or other
+            post-processing steps.
+
+            Accepts the same single-hook-or-list-of-hooks form as
+            ``pre_model_hook``.

            !!! Note
-                Only available with `version="v2"`.
+                Only available with ``version="v2"``.
+
        state_schema: An optional state schema that defines graph state.
            Must have `messages` and `remaining_steps` keys.
            Defaults to `AgentState` that defines those two keys.
@@ -551,6 +727,10 @@ def create_react_agent(
            else AgentState
        )

+    # Normalise hook arguments: a list of hooks is composed into a single hook.
+    resolved_pre_model_hook: RunnableLike | None = _resolve_hook(pre_model_hook)
+    resolved_post_model_hook: RunnableLike | None = _resolve_hook(post_model_hook)
+
    llm_builtin_tools: list[dict] = []
    if isinstance(tools, ToolNode):
        tool_classes = list(tools.tools_by_name.values())
@@ -634,7 +814,7 @@ def create_react_agent(
        return False

    def _get_model_input_state(state: StateSchema) -> StateSchema:
-        if pre_model_hook is not None:
+        if resolved_pre_model_hook is not None:
            messages = (
                _get_state_value(state, "llm_input_messages")
            ) or _get_state_value(state, "messages")
@@ -721,7 +901,7 @@ def create_react_agent(
        return {"messages": [response]}

    input_schema: StateSchemaType
-    if pre_model_hook is not None:
+    if resolved_pre_model_hook is not None:
        # Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages'
        if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
            # For Pydantic schemas
@@ -792,8 +972,8 @@ def create_react_agent(
            RunnableCallable(call_model, acall_model),
            input_schema=input_schema,
        )
-        if pre_model_hook is not None:
-            workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+        if resolved_pre_model_hook is not None:
+            workflow.add_node("pre_model_hook", resolved_pre_model_hook)  # type: ignore[arg-type]
            workflow.add_edge("pre_model_hook", "agent")
            entrypoint = "pre_model_hook"
        else:
@@ -801,8 +981,8 @@ def create_react_agent(

        workflow.set_entry_point(entrypoint)

-        if post_model_hook is not None:
-            workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+        if resolved_post_model_hook is not None:
+            workflow.add_node("post_model_hook", resolved_post_model_hook)  # type: ignore[arg-type]
            workflow.add_edge("agent", "post_model_hook")

        if response_format is not None:
@@ -813,7 +993,7 @@ def create_react_agent(
                    agenerate_structured_response,
                ),
            )
-            if post_model_hook is not None:
+            if resolved_post_model_hook is not None:
                workflow.add_edge("post_model_hook", "generate_structured_response")
            else:
                workflow.add_edge("agent", "generate_structured_response")
@@ -833,7 +1013,7 @@ def create_react_agent(
        last_message = messages[-1]
        # If there is no function call, then we finish
        if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
-            if post_model_hook is not None:
+            if resolved_post_model_hook is not None:
                return "post_model_hook"
            elif response_format is not None:
                return "generate_structured_response"
@@ -844,7 +1024,7 @@ def create_react_agent(
            if version == "v1":
                return "tools"
            elif version == "v2":
-                if post_model_hook is not None:
+                if resolved_post_model_hook is not None:
                    return "post_model_hook"
                return [
                    Send(
@@ -873,8 +1053,8 @@ def create_react_agent(

    # Optionally add a pre-model hook node that will be called
    # every time before the "agent" (LLM-calling node)
-    if pre_model_hook is not None:
-        workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+    if resolved_pre_model_hook is not None:
+        workflow.add_node("pre_model_hook", resolved_pre_model_hook)  # type: ignore[arg-type]
        workflow.add_edge("pre_model_hook", "agent")
        entrypoint = "pre_model_hook"
    else:
@@ -888,8 +1068,8 @@ def create_react_agent(
    post_model_hook_paths = [entrypoint, "tools"]

    # Add a post model hook node if post_model_hook is provided
-    if post_model_hook is not None:
-        workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+    if resolved_post_model_hook is not None:
+        workflow.add_node("post_model_hook", resolved_post_model_hook)  # type: ignore[arg-type]
        agent_paths.append("post_model_hook")
        workflow.add_edge("agent", "post_model_hook")
    else:
@@ -904,17 +1084,17 @@ def create_react_agent(
                agenerate_structured_response,
            ),
        )
-        if post_model_hook is not None:
+        if resolved_post_model_hook is not None:
            post_model_hook_paths.append("generate_structured_response")
        else:
            agent_paths.append("generate_structured_response")
    else:
-        if post_model_hook is not None:
+        if resolved_post_model_hook is not None:
            post_model_hook_paths.append(END)
        else:
            agent_paths.append(END)

-    if post_model_hook is not None:
+    if resolved_post_model_hook is not None:

        def post_model_hook_router(state: StateSchema) -> str | list[Send]:
            """Route to the next node after post_model_hook.
@@ -1012,4 +1192,5 @@ __all__ = [
    "AgentStatePydantic",
    "AgentStateWithStructuredResponse",
    "AgentStateWithStructuredResponsePydantic",
+    "HookLike",
 ]
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.0a2"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"
@@ -253,7 +253,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.4.0"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
@@ -266,9 +266,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/59/de/679a53472c25860837e32c0442c962fa86e95317a36460e2c9d5c91b17c2/langchain_core-1.4.0.tar.gz", hash = "sha256:1dc341eed802ed9c117c0df3923c991e5e9e226571e5725c194eeb5bd93d1a7f", size = 920260, upload-time = "2026-05-11T18:42:35.919Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0f/1a/86c38c27b81913a1c6c12448cab55defb5a1097c7dc9a4cea83f55477a2d/langchain_core-1.4.0-py3-none-any.whl", hash = "sha256:23cbbdb46e38ddd1dd5247e6167e96013eae74bea4c5949c550809970a9e565c", size = 548120, upload-time = "2026-05-11T18:42:33.992Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]

 [[package]]
@@ -285,7 +285,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.0"
+version = "1.2.0a7"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
@@ -298,7 +298,7 @@ dependencies = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
    { name = "langgraph-checkpoint", editable = "../checkpoint" },
    { name = "langgraph-prebuilt", editable = "." },
    { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -369,7 +369,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -417,7 +417,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0"
+version = "3.1.0a4"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
    { name = "langgraph-checkpoint" },
@@ -464,7 +464,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.0"
+version = "3.1.0a1"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
    { name = "aiosqlite" },
@@ -507,7 +507,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.0a2"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -1490,11 +1490,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.7.0"
+version = "2.6.3"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
+    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
 ]

 [[package]]
@@ -266,7 +266,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.4.0"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
@@ -279,9 +279,9 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/59/de/679a53472c25860837e32c0442c962fa86e95317a36460e2c9d5c91b17c2/langchain_core-1.4.0.tar.gz", hash = "sha256:1dc341eed802ed9c117c0df3923c991e5e9e226571e5725c194eeb5bd93d1a7f", size = 920260, upload-time = "2026-05-11T18:42:35.919Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0f/1a/86c38c27b81913a1c6c12448cab55defb5a1097c7dc9a4cea83f55477a2d/langchain_core-1.4.0-py3-none-any.whl", hash = "sha256:23cbbdb46e38ddd1dd5247e6167e96013eae74bea4c5949c550809970a9e565c", size = 548120, upload-time = "2026-05-11T18:42:33.992Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]

 [[package]]
@@ -298,7 +298,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.0"
+version = "1.2.0a7"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
@@ -311,7 +311,7 @@ dependencies = [

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
    { name = "langgraph-checkpoint", editable = "../checkpoint" },
    { name = "langgraph-prebuilt", editable = "../prebuilt" },
    { name = "langgraph-sdk", editable = "." },
@@ -382,7 +382,7 @@ test = [

 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0"
+version = "4.1.0a4"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -430,7 +430,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.0a2"
 source = { editable = "../prebuilt" }
 dependencies = [
    { name = "langchain-core" },