release: bump alpha packages to official versions (#7775 )

## Summary Bumps all alpha-versioned packages to their official releases: - `langgraph`: `1.2.0a7` → `1.2.0` - `langgraph-checkpoint`: `4.1.0a4` → `4.1.0` - `langgraph-checkpoint-postgres`: `3.1.0a4` → `3.1.0` - `langgraph-checkpoint-sqlite`: `3.1.0a1` → `3.1.0` - `langgraph-prebuilt`: `1.1.0a2` → `1.1.0` Also removes the `a*` alpha specifiers from cross-dependency pins in `langgraph`, `checkpoint-postgres`, and `checkpoint-sqlite`, and regenerates all `uv.lock` files.
feat(langgraph): durable error-handler resume across host crashes (#7773 )
2026-08-30 03:39:38 +02:00 · 2026-05-11 23:27:38 -04:00 · 2026-05-12 01:39:03 +00:00 · 2026-05-11 20:24:21 -04:00 · 2026-05-11 20:20:17 -04:00 · 2026-05-11 19:37:16 -04:00
44 changed files with 1415 additions and 796 deletions
@@ -64,8 +64,6 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `copy_thread` | no | `acopy_thread` |
 | `prune` | no | `aprune` |
 | `delta_channel_history` | no | `aget_delta_channel_history` |
-| `delta_channel_keepset` | no | `aget_delta_channel_keepset` |
-| `delta_channel_reconstruction` | no | `aput` |

 Extended capabilities are detected by checking whether the method is overridden from `BaseCheckpointSaver`. If not overridden, those tests are skipped.

@@ -24,8 +24,6 @@ class Capability(str, Enum):
    COPY_THREAD = "copy_thread"
    PRUNE = "prune"
    DELTA_CHANNEL_HISTORY = "delta_channel_history"
-    DELTA_CHANNEL_KEEPSET = "delta_channel_keepset"
-    DELTA_CHANNEL_RECONSTRUCTION = "delta_channel_reconstruction"


 # Capabilities that every checkpointer must support.
@@ -46,8 +44,6 @@ EXTENDED_CAPABILITIES = frozenset(
        Capability.COPY_THREAD,
        Capability.PRUNE,
        Capability.DELTA_CHANNEL_HISTORY,
-        Capability.DELTA_CHANNEL_KEEPSET,
-        Capability.DELTA_CHANNEL_RECONSTRUCTION,
    }
 )

@@ -64,8 +60,6 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
    Capability.COPY_THREAD: "acopy_thread",
    Capability.PRUNE: "aprune",
    Capability.DELTA_CHANNEL_HISTORY: "aget_delta_channel_history",
-    Capability.DELTA_CHANNEL_KEEPSET: "aget_tuple",
-    Capability.DELTA_CHANNEL_RECONSTRUCTION: "aput",
 }


@@ -12,12 +12,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
    run_delta_channel_history_tests,
 )
-from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
-    run_delta_channel_keepset_tests,
-)
-from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
-    run_delta_channel_reconstruction_tests,
-)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -34,6 +28,4 @@ __all__ = [
    "run_copy_thread_tests",
    "run_prune_tests",
    "run_delta_channel_history_tests",
-    "run_delta_channel_keepset_tests",
-    "run_delta_channel_reconstruction_tests",
 ]
@@ -1,172 +0,0 @@
-"""DELTA_CHANNEL_KEEPSET capability tests — aget_delta_channel_keepset contract."""
-
-from __future__ import annotations
-
-import traceback
-from collections.abc import Callable
-from uuid import uuid4
-
-from langgraph.checkpoint.base import BaseCheckpointSaver
-
-from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
-
-
-async def test_keepset_empty_channels_returns_target_only(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Empty channels → keep-set is just {target_id}."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=4
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=[])
-    head_id = head["configurable"]["checkpoint_id"]
-    assert keep == {head_id}, f"Expected only target, got {keep}"
-
-
-async def test_keepset_snapshot_at_target(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """When target itself has a snapshot, keep-set is just {target_id}."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="ch",
-        snapshots_at_steps=[0, 3],
-        total_steps=4,
-    )
-    head = configs[3]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    head_id = head["configurable"]["checkpoint_id"]
-    assert keep == {head_id}, f"Snapshot at target should yield only target, got {keep}"
-
-
-async def test_keepset_snapshot_n_back(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Snapshot N steps back → target + intermediates + snapshot ancestor."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs}
-    assert keep == expected_ids, f"Expected all ancestors, got {keep}"
-
-
-async def test_keepset_multi_channel_union(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Multi-channel keep-set is the union (max chain per channel)."""
-    tid = str(uuid4())
-    from langgraph.checkpoint.base import Checkpoint
-    from langgraph.checkpoint.base.id import uuid6
-    from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-    from langgraph.checkpoint.conformance.test_utils import generate_metadata
-
-    configs: list = []
-    parent_cfg = None
-    for step in range(6):
-        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
-        if parent_cfg:
-            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
-                "checkpoint_id"
-            ]
-        cv: dict = {}
-        cvs: dict = {}
-        # Channel "a" has snapshot at step 3 (recent)
-        if step == 3:
-            cv["a"] = _DeltaSnapshot("snap_a")
-            cvs["a"] = step + 1
-        # Channel "b" has snapshot at step 1 (further back)
-        if step == 1:
-            cv["b"] = _DeltaSnapshot("snap_b")
-            cvs["b"] = step + 1
-        cp = Checkpoint(
-            v=1,
-            id=str(uuid6(clock_seq=-1)),
-            ts="",
-            channel_values=cv,
-            channel_versions=cvs,
-            versions_seen={},
-            updated_channels=None,
-        )
-        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
-        configs.append(parent_cfg)
-
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["a", "b"])
-    # Union: b needs back to step 1, so steps 1..5 (all except step 0) plus head
-    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs[1:]}
-    expected_ids.add(head["configurable"]["checkpoint_id"])
-    assert keep == expected_ids, f"Expected union, got {keep} vs {expected_ids}"
-
-
-async def test_keepset_walk_to_root(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """No snapshot anywhere → entire chain to root is in keep-set."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[], total_steps=4
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    all_ids = {c["configurable"]["checkpoint_id"] for c in configs}
-    assert keep == all_ids, f"Expected full chain, got {keep}"
-
-
-async def test_keepset_deterministic(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Same inputs return identical sets."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
-    )
-    head = configs[-1]
-    keep1 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    keep2 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    assert keep1 == keep2
-
-
-ALL_DELTA_CHANNEL_KEEPSET_TESTS = [
-    test_keepset_empty_channels_returns_target_only,
-    test_keepset_snapshot_at_target,
-    test_keepset_snapshot_n_back,
-    test_keepset_multi_channel_union,
-    test_keepset_walk_to_root,
-    test_keepset_deterministic,
-]
-
-
-async def run_delta_channel_keepset_tests(
-    saver: BaseCheckpointSaver,
-    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
-) -> tuple[int, int, list[str]]:
-    """Run all delta_channel_keepset tests. Returns (passed, failed, failure_names)."""
-    passed = 0
-    failed = 0
-    failures: list[str] = []
-    for test_fn in ALL_DELTA_CHANNEL_KEEPSET_TESTS:
-        try:
-            await test_fn(saver)
-            passed += 1
-            if on_test_result:
-                on_test_result("delta_channel_keepset", test_fn.__name__, True, None)
-        except Exception:
-            failed += 1
-            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
-            failures.append(msg)
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_keepset",
-                    test_fn.__name__,
-                    False,
-                    traceback.format_exc(),
-                )
-    return passed, failed, failures
@@ -1,164 +0,0 @@
-"""DELTA_CHANNEL_RECONSTRUCTION capability tests — end-to-end round-trip.
-
-Exercises: aput + aput_writes + aget_delta_channel_history + reconstruction.
-This catches the most common silent-corruption mode: failing to round-trip
-`_DeltaSnapshot` blobs through serialization.
-
-NOTE: This test does NOT import from `langgraph` (which is not a dependency
-of checkpoint-conformance). Instead it inlines a minimal reconstruction
-equivalent: seed + fold writes through a simple list-append reducer.
-"""
-
-from __future__ import annotations
-
-import traceback
-from collections.abc import Callable
-from typing import Any
-from uuid import uuid4
-
-from langgraph.checkpoint.base import BaseCheckpointSaver
-
-from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
-
-
-def _reconstruct(seed: Any, writes: list) -> list:
-    """Minimal DeltaChannel reconstruction: list-append reducer.
-
-    Mirrors DeltaChannel.from_checkpoint(seed) + replay_writes(writes).
-    """
-    from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-    if seed is None:
-        base: list = []
-    elif isinstance(seed, _DeltaSnapshot):
-        base = list(seed.value)
-    else:
-        base = list(seed)
-    for _task_id, _ch, value in writes:
-        base = base + value
-    return base
-
-
-async def test_reconstruction_basic(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruct delta channel value from history matches expected."""
-    tid = str(uuid4())
-    # 5 steps: snapshot at 0 (value=[0]), writes at 1,2,3,4.
-    # Head = step 4. Walk from parent (step 3) collects writes 1,2,3.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[0],
-        total_steps=5,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # seed=[0] from step 0, writes from steps 1,2,3
-    expected = [0] + [1] + [2] + [3]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-async def test_reconstruction_mid_chain_snapshot(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruction works when snapshot is mid-chain."""
-    tid = str(uuid4())
-    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
-    # Head = step 5. Walk from step 4 stops at step 3 (snapshot).
-    # Collects writes from step 4.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[0, 3],
-        total_steps=6,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # Snapshot at step 3 = [3], write from step 4
-    expected = [3] + [4]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-async def test_reconstruction_no_snapshot(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruction from root (no snapshot) gives all writes accumulated."""
-    tid = str(uuid4())
-    # 4 steps: no snapshot, writes at 0,1,2,3.
-    # Head = step 3. Walk from step 2 collects writes 0,1,2.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[],
-        total_steps=4,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # No seed → start empty, writes from steps 0,1,2
-    expected = [0] + [1] + [2]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS = [
-    test_reconstruction_basic,
-    test_reconstruction_mid_chain_snapshot,
-    test_reconstruction_no_snapshot,
-]
-
-
-async def run_delta_channel_reconstruction_tests(
-    saver: BaseCheckpointSaver,
-    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
-) -> tuple[int, int, list[str]]:
-    """Run all reconstruction tests. Returns (passed, failed, failure_names)."""
-    passed = 0
-    failed = 0
-    failures: list[str] = []
-    for test_fn in ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS:
-        try:
-            await test_fn(saver)
-            passed += 1
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_reconstruction", test_fn.__name__, True, None
-                )
-        except Exception:
-            failed += 1
-            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
-            failures.append(msg)
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_reconstruction",
-                    test_fn.__name__,
-                    False,
-                    traceback.format_exc(),
-                )
-    return passed, failed, failures
@@ -22,12 +22,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
    run_delta_channel_history_tests,
 )
-from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
-    run_delta_channel_keepset_tests,
-)
-from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
-    run_delta_channel_reconstruction_tests,
-)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -45,8 +39,6 @@ _RUNNERS = {
    Capability.COPY_THREAD: run_copy_thread_tests,
    Capability.PRUNE: run_prune_tests,
    Capability.DELTA_CHANNEL_HISTORY: run_delta_channel_history_tests,
-    Capability.DELTA_CHANNEL_KEEPSET: run_delta_channel_keepset_tests,
-    Capability.DELTA_CHANNEL_RECONSTRUCTION: run_delta_channel_reconstruction_tests,
 }


@@ -50,7 +50,6 @@ unresolved-attribute = "ignore"
 unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
-missing-typed-dict-key = "ignore"

 [tool.ruff]
 lint.select = [
@@ -279,7 +279,7 @@ wheels = [

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

 [project]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a4"
+version = "3.1.0"
 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.0a4,<5.0.0",
+  "langgraph-checkpoint>=4.1.0,<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.2"
+version = "1.3.3"
 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/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
+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" }
 wheels = [
-    { 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" },
+    { 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]]
@@ -276,7 +276,7 @@ wheels = [

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

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

 [project]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.0a1"
+version = "3.1.0"
 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.0a4,<5.0.0",
+    "langgraph-checkpoint>=4.1.0,<5.0.0",
    "aiosqlite>=0.20",
    "sqlite-vec>=0.1.6",
 ]
@@ -27,8 +27,6 @@ async def test_delta_channel_conformance():
        sqlite_saver,
        capabilities={
            "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
        },
    )
    for cap, result in report.results.items():
@@ -253,10 +253,11 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.2.28"
+version = "1.3.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
+    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -265,14 +266,26 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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


@@ -680,123 +689,6 @@ class BaseCheckpointSaver(Generic[V]):
            result[ch] = entry
        return result

-    def get_delta_channel_keepset(
-        self,
-        *,
-        config: RunnableConfig,
-        channels: Sequence[str],
-    ) -> set[str]:
-        """Return ancestor checkpoint_ids that must survive deletion.
-
-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. The signature may change while the delta-channel design
-            stabilizes.
-
-        Walks the parent chain from `config` backward, collecting visited
-        checkpoint_ids (inclusive of the target), and terminates per-channel
-        when that channel has a populated `channel_values[ch]` (a
-        `_DeltaSnapshot` blob or a pre-migration plain value). The returned
-        set is the minimum keep-set: every checkpoint_id whose removal would
-        break reconstruction of the listed channels at `config`.
-
-        Pass `channels=[]` to return just `{config.checkpoint_id}` — useful
-        for graphs that don't use `DeltaChannel`.
-
-        Compose this into custom `prune` / `delete_for_runs` / `copy_thread`::
-
-            keep = saver.get_delta_channel_keepset(
-                config=head_config, channels=delta_channels,
-            )
-            delete_rows_not_in(keep)
-
-        Note:
-            The default implementation here uses repeated `get_tuple` calls
-            to walk the parent chain. This is a basic reference implementation
-            suitable for low-frequency maintenance operations (prune, etc.).
-            Custom checkpointer backends may override with a more efficient
-            version tailored to their data model (e.g. a single SQL query
-            with a recursive CTE), but it is not required — the default works
-            correctly for any saver that implements `get_tuple`.
-
-        Args:
-            config: Configuration identifying the target checkpoint.
-            channels: Channel names whose delta history must be preserved.
-                Empty sequence means only the target checkpoint_id is kept.
-
-        Returns:
-            Set of checkpoint_ids that must not be deleted.
-        """
-        target_tuple = self.get_tuple(config)
-        if target_tuple is None:
-            return set()
-        target_id = target_tuple.config["configurable"]["checkpoint_id"]
-        keep: set[str] = {target_id}
-        if not channels:
-            return keep
-        remaining: set[str] = set(channels)
-        for ch in list(remaining):
-            if ch in target_tuple.checkpoint["channel_values"]:
-                remaining.discard(ch)
-        if not remaining:
-            return keep
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and remaining:
-            tup = self.get_tuple(cursor_config)
-            if tup is None:
-                break
-            cid = tup.config["configurable"]["checkpoint_id"]
-            keep.add(cid)
-            for ch in list(remaining):
-                if ch in tup.checkpoint["channel_values"]:
-                    remaining.discard(ch)
-            if not remaining:
-                break
-            cursor_config = tup.parent_config
-        return keep
-
-    async def aget_delta_channel_keepset(
-        self,
-        *,
-        config: RunnableConfig,
-        channels: Sequence[str],
-    ) -> set[str]:
-        """Async version of `get_delta_channel_keepset`.
-
-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. See `get_delta_channel_keepset` for full documentation.
-        """
-        target_tuple = await self.aget_tuple(config)
-        if target_tuple is None:
-            return set()
-        target_id = target_tuple.config["configurable"]["checkpoint_id"]
-        keep: set[str] = {target_id}
-        if not channels:
-            return keep
-        remaining: set[str] = set(channels)
-        for ch in list(remaining):
-            if ch in target_tuple.checkpoint["channel_values"]:
-                remaining.discard(ch)
-        if not remaining:
-            return keep
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and remaining:
-            tup = await self.aget_tuple(cursor_config)
-            if tup is None:
-                break
-            cid = tup.config["configurable"]["checkpoint_id"]
-            keep.add(cid)
-            for ch in list(remaining):
-                if ch in tup.checkpoint["channel_values"]:
-                    remaining.discard(ch)
-            if not remaining:
-                break
-            cursor_config = tup.parent_config
-        return keep
-
    def get_next_version(self, current: V | None, channel: None) -> V:
        """Generate the next version ID for a channel.

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
        AllowedMsgpackModules,
    )

-LC_REVIVER = Reviver()
+LC_REVIVER = Reviver(allowed_objects="core")
 EMPTY_BYTES = b""
 logger = logging.getLogger(__name__)

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

 [project]
 name = "langgraph-checkpoint"
-version = "4.1.0a4"
+version = "4.1.0"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.10"
@@ -25,8 +25,6 @@ async def test_delta_channel_conformance():
        mem_saver,
        capabilities={
            "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
        },
    )
    for cap, result in report.results.items():
@@ -268,7 +268,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.3.3"
 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/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
+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" }
 wheels = [
-    { 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" },
+    { 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]]
@@ -300,7 +300,7 @@ wheels = [

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

 [[package]]
@@ -1 +1 @@
-__version__ = "0.4.25"
+__version__ = "0.4.26"
@@ -35,6 +35,12 @@ 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:
@@ -123,6 +129,18 @@ 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):
@@ -176,12 +194,12 @@ def validate_config(config: Config) -> Config:
            )
    if api_version:
        try:
-            parts = tuple(map(int, api_version.split("-")[0].split(".")))
+            parts = _parse_api_version_parts(api_version)
            if len(parts) > 3:
                raise ValueError(
                    "Version must be major or major.minor or major.minor.patch."
                )
-        except TypeError:
+        except (TypeError, ValueError):
            raise click.UsageError(
                f"Invalid version format: {api_version}.\n\n"
                "Pin to a minor version, e.g.:\n"
@@ -2944,6 +2944,23 @@ 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>=0.2.14",
+    "langchain-core>=1.3.3",
 ]

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

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

 [[package]]
 name = "langchain-core"
-version = "1.2.28"
+version = "1.3.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
+    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -227,9 +228,21 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=0.2.14" },
+    { name = "langchain-core", specifier = ">=1.3.3" },
    { 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>=0.2.14",
+    "langchain-core>=1.3.3",
 ]

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

 [[package]]
 name = "langchain-core"
-version = "1.2.28"
+version = "1.3.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "jsonpatch" },
+    { name = "langchain-protocol" },
    { name = "langsmith" },
    { name = "packaging" },
    { name = "pydantic" },
@@ -203,9 +204,21 @@ dependencies = [
    { name = "typing-extensions" },
    { name = "uuid-utils" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

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

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

 [[package]]
 name = "langchain-core"
-version = "1.3.0"
+version = "1.3.3"
 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'" },
@@ -958,9 +959,21 @@ 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/92/fe/20190232d9b513242899dbb0c2bb77e31b4d61e343743adbe90ebc2603d2/langchain_core-1.3.0.tar.gz", hash = "sha256:14a39f528bf459aa3aa40d0a7f7f1bae7520d435ef991ae14a4ceb74d8c49046", size = 860755, upload-time = "2026-04-17T14:51:38.298Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d3/ae/8b74458fc3850ec3d150eb9f45e857db129dafa801fb5cf173dfc9f8bbf3/langchain_core-1.3.3.tar.gz", hash = "sha256:fa510a5db8efdc0c6ff41c0939fb5c00a0183c11f6b84233e892e3227ff69182", size = 915041, upload-time = "2026-05-05T19:02:36.612Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f8/e2/dbfa347aa072a6dc4cd38d6f9ebfc730b4c14c258c47f480f4c5c546f177/langchain_core-1.3.0-py3-none-any.whl", hash = "sha256:baf16ee028475df177b9ab8869a751c79406d64a6f12125b93802991b566cced", size = 515140, upload-time = "2026-04-17T14:51:36.274Z" },
+    { url = "https://files.pythonhosted.org/packages/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" },
 ]

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

 [[package]]
@@ -29,6 +29,9 @@ 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 `delta_updates_since_snapshot`
+        `_DeltaSnapshot` blob shape, the `counters_since_delta_snapshot`
        metadata field) is not yet stable.

    The reducer receives the current accumulated value and a batch of writes
@@ -47,9 +47,12 @@ 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 per-channel update count. `create_checkpoint`
-    writes a full `_DeltaSnapshot` blob every `snapshot_frequency` updates to
-    this channel, bounding replay depth.
+    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.

    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 is_dataclass
+from dataclasses import dataclass, is_dataclass
 from datetime import timedelta
 from functools import partial
 from inspect import isclass, isfunction, ismethod, signature
@@ -95,6 +95,17 @@ __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:
@@ -251,10 +262,77 @@ 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 | {
@@ -1193,10 +1271,63 @@ 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 spec.error_handler_node is not None
+            if not spec.is_error_handler and 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,21 +36,26 @@ def empty_checkpoint() -> Checkpoint:

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

-    A channel snapshots when its accumulated update count (since the last
-    snapshot) reaches or exceeds `snapshot_frequency`. This is a pure
+    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
    predicate — no mutation.
    """
-    return {
-        name
-        for name, ch in channels.items()
-        if isinstance(ch, DeltaChannel)
-        and ch.is_available()
-        and counts.get(name, 0) >= ch.snapshot_frequency
-    }
+    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


 def create_checkpoint(
@@ -69,7 +74,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, counts)`; defaults to empty
+    `delta_channels_to_snapshot(channels, counters)`; defaults to empty
    (no snapshots) when not provided.
    """
    ts = datetime.now(timezone.utc).isoformat()
@@ -231,17 +236,3 @@ 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,6 +203,12 @@ 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
@@ -474,6 +480,13 @@ 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)
@@ -553,7 +566,7 @@ class PregelLoop:
            self.tasks[pushed.id] = pushed
            # match any pending writes to the new task
            if not self.is_replaying:
-                self._match_writes({pushed.id: pushed})
+                self._reapply_writes_to_succeeded_nodes({pushed.id: pushed})
            # return the new task, to be started if not run before
            return pushed

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

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

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

    # private

-    def _match_writes(self, tasks: Mapping[str, PregelExecutableTask]) -> None:
+    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.
+        """
        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
@@ -776,17 +865,6 @@ class PregelLoop:
                and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
                in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {})
            )
-            # Outer graph: time-travel-resume — explicit non-head
-            # checkpoint paired with a Command(resume=...). Without this,
-            # `Command(resume=...) + checkpoint=<non-head>` would match the
-            # plain-resume exclusion below and skip the cleanup needed to
-            # treat this as a fork.
-            or (
-                not self.is_nested
-                and getattr(self, "_loaded_explicit_non_head", False)
-                and input_is_command
-                and cast(Command, self.input).resume is not None
-            )
            or not (
                # Outer graph: resume arrives as Command(resume=...)
                (input_is_command and cast(Command, self.input).resume is not None)
@@ -989,35 +1067,41 @@ class PregelLoop:
        if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
            # checkpoint already saved
            return
-        # Per-delta-channel update bookkeeping.
+        # 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.
        #
        # `_put_checkpoint` is called once per superstep with a fresh
        # metadata dict (source="input"|"loop"|"fork") — those are the
-        # intermediate calls that bump the count by +1 for each delta
-        # channel touched that step. In exit mode,
+        # intermediate calls that bump counters. 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. (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.)
+        # superstep.
        if not exiting:
-            prev_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+            prev_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
            )
-            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
+            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)
            metadata["step"] = self.step
            metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
            self.checkpoint_metadata = metadata
        else:
-            new_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+            new_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
            )
        # do checkpoint?
        do_checkpoint = self._checkpointer_put_after_previous is not None and (
@@ -1025,7 +1109,7 @@ class PregelLoop:
        )
        # create new checkpoint
        channels_to_snapshot = (
-            delta_channels_to_snapshot(self.channels, new_counts)
+            delta_channels_to_snapshot(self.channels, new_counters)
            if do_checkpoint
            else set()
        )
@@ -1041,11 +1125,12 @@ class PregelLoop:
            channels_to_snapshot=channels_to_snapshot,
        )
        for k in channels_to_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"]
+            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"]
        # 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()
@@ -1120,8 +1205,10 @@ class PregelLoop:
        ):
            return

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

        pending = [
            (step, tid, ch, v)
@@ -1456,12 +1543,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
        handler_node = self.nodes[failed_task.name].error_handler_node
        if not handler_node:
            return None
-        writes = list(failed_task.writes)
-        writes.append((ERROR_SOURCE_NODE, failed_task.name))
-        self.put_writes(
-            failed_task.id,
-            writes,
-        )
+        # 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)
        handler_task = prepare_node_error_handler_task(
            failed_task,
            handler_node_name=handler_node,
@@ -1484,7 +1569,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            return None
        self.tasks[handler_task.id] = handler_task
        if not self.is_replaying:
-            self._match_writes({handler_task.id: handler_task})
+            self._reapply_writes_to_succeeded_nodes({handler_task.id: handler_task})
        for task in self.match_cached_writes():
            self.output_writes(task.id, task.writes, cached=True)
        return handler_task
@@ -1511,13 +1596,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):

    def __enter__(self) -> Self:
        self._graph_lifecycle_events = deque()
-        # Set by the explicit-id branch when the loaded checkpoint is not
-        # the latest for the (thread, ns). Lets _first distinguish a
-        # plain resume (`Command(resume=...)` against head, which carries
-        # an explicit checkpoint_id only because clients like LangGraph
-        # Studio echo it back) from time-travel-resume (`Command(resume=
-        # ...)` against an explicit non-head checkpoint).
-        self._loaded_explicit_non_head = False
        if not self.checkpointer:
            saved = None
        elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1525,14 +1603,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            # This covers both normal replay and subgraphs resolved via
            # checkpoint_map during time-travel.
            saved = self.checkpointer.get_tuple(self.checkpoint_config)
-            if saved is not None and not self.is_nested:
-                head = self.checkpointer.get_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
        elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
            # Subgraph replay: the parent is replaying and passed us a
            # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1581,6 +1651,7 @@ 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
        )
@@ -1726,12 +1797,10 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
        handler_node = self.nodes[failed_task.name].error_handler_node
        if not handler_node:
            return None
-        writes = list(failed_task.writes)
-        writes.append((ERROR_SOURCE_NODE, failed_task.name))
-        self.put_writes(
-            failed_task.id,
-            writes,
-        )
+        # 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)
        handler_task = prepare_node_error_handler_task(
            failed_task,
            handler_node_name=handler_node,
@@ -1754,7 +1823,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            return None
        self.tasks[handler_task.id] = handler_task
        if not self.is_replaying:
-            self._match_writes({handler_task.id: handler_task})
+            self._reapply_writes_to_succeeded_nodes({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
@@ -1784,7 +1853,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):

    async def __aenter__(self) -> Self:
        self._graph_lifecycle_events = deque()
-        self._loaded_explicit_non_head = False
        if not self.checkpointer:
            saved = None
        elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1792,14 +1860,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            # This covers both normal replay and subgraphs resolved via
            # checkpoint_map during time-travel.
            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-            if saved is not None and not self.is_nested:
-                head = await self.checkpointer.aget_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
        elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
            # Subgraph replay: the parent is replaying and passed us a
            # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1848,6 +1908,7 @@ 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,6 +31,7 @@ from langgraph._internal._constants import (
    CONFIG_KEY_CALL,
    CONFIG_KEY_SCRATCHPAD,
    ERROR,
+    ERROR_SOURCE_NODE,
    INTERRUPT,
    NO_WRITES,
    RESUME,
@@ -597,7 +598,7 @@ class PregelRunner:
                if self._should_route_to_error_handler(task) and not isinstance(
                    exception, GraphBubbleUp
                ):
-                    # Mark early in commit path; loop-side routing may happen later.
+                    task.writes.append((ERROR_SOURCE_NODE, task.name))
                    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.0a7"
+version = "1.2.0"
 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.0a2,<2",
-    "langgraph-checkpoint>=4.1.0a4,<5.0.0",
+    "langchain-core>=1.4.0,<2",
+    "langgraph-checkpoint>=4.1.0,<5.0.0",
    "langgraph-sdk>=0.3.0,<0.4.0",
-    "langgraph-prebuilt>=1.1.0a2,<1.2.0",
+    "langgraph-prebuilt>=1.1.0,<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 delta_updates_since_snapshot
-    after an equivalent run."""
+    """Sync and exit durability produce the same update count in
+    counters_since_delta_snapshot after an equivalent run."""
    for durability in ("sync", "exit"):
        saver = InMemorySaver()
        graph = _build_graph(saver)
@@ -175,9 +175,13 @@ async def test_exit_count_parity_sync_vs_exit() -> None:

        head = saver.get_tuple(config)
        assert head is not None
-        counts = head.metadata.get("delta_updates_since_snapshot", {})
-        assert counts.get("messages") == 2, (
-            f"durability={durability}: expected count=2, got {counts}"
+        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}"
        )


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

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


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


@@ -0,0 +1,195 @@
+"""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 RunnableLambda, RunnableParallel
+from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnableParallel
 from langgraph.checkpoint.memory import InMemorySaver, MemorySaver
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from typing_extensions import TypedDict
@@ -2280,3 +2280,525 @@ 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"]
@@ -3964,96 +3964,3 @@ def test_subgraph_called_in_loop_loads_state_on_replay(

    assert len(observed) == 1
    assert observed[0] == ("sub_step", {"sub_trail": ["s", "s", "s"]})
-
-
-def test_subgraph_time_travel_resume_with_command_in_one_call(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Repro for time-travel-resume issued as a SINGLE call:
-    `invoke(Command(resume=...), config={..., checkpoint_id: <non-head>})`.
-
-    Equivalent to the existing two-call pattern
-    (`test_subgraph_time_travel_resume_from_first_interrupt`) which does
-    `invoke(None, sub_config)` then `invoke(Command(resume=...))`. Clients
-    like LangGraph API server / Studio can submit this as one request.
-    """
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run to completion through both interrupts.
-    graph.invoke({"value": []}, config)
-    parent_state_at_first = graph.get_state(config, subgraphs=True)
-    sub_config_at_first = parent_state_at_first.tasks[0].state.config
-    graph.invoke(Command(resume="answer_1"), config)
-    graph.invoke(Command(resume="answer_2"), config)
-
-    final_after_original = graph.get_state(config).values
-    assert final_after_original["value"] == [
-        "step_a_done",
-        "ask_1:answer_1",
-        "ask_2:answer_2",
-    ]
-
-    # ONE-CALL time-travel-resume: pair Command(resume=...) with a
-    # non-head parent checkpoint_id in the same invocation.
-    called.clear()
-    result = graph.invoke(Command(resume="new_answer_1"), sub_config_at_first)
-
-    # `step_a` ran before the time-travel target — must not re-execute.
-    assert "step_a" not in called, f"step_a re-executed; called={called}"
-    # The new answer must be applied (not the cached `answer_1`).
-    assert result.get("__interrupt__"), (
-        f"expected to pause at the next interrupt; got result={result}"
-    )
-    assert result["__interrupt__"][0].value == "Question 2?"
-    sub_state_now = graph.get_state(config, subgraphs=True).tasks[0].state
-    assert sub_state_now.values["value"] == ["step_a_done", "ask_1:new_answer_1"], (
-        f"subgraph state at int2 should reflect new_answer_1; "
-        f"got {sub_state_now.values}"
-    )
-
-    # Resume the second interrupt with another new answer; ensure the
-    # whole branch concludes consistently.
-    called.clear()
-    final = graph.invoke(Command(resume="new_answer_2"), config)
-    assert "step_a" not in called
-    assert "ask_1" not in called
-    assert final["value"] == [
-        "step_a_done",
-        "ask_1:new_answer_1",
-        "ask_2:new_answer_2",
-    ]
@@ -1350,7 +1350,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.4.0a2"
+version = "1.4.0"
 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

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

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0,<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.0a4"
+version = "4.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -1611,7 +1611,7 @@ test = [

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

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

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

 [[package]]
 name = "mistune"
-version = "3.2.0"
+version = "3.2.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

 [project]
 name = "langgraph-prebuilt"
-version = "1.1.0a2"
+version = "1.1.0"
 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.0a2"
+version = "1.4.0"
 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

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

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0,<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.0a4"
+version = "4.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -417,7 +417,7 @@ test = [

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

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

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

 [[package]]
 name = "urllib3"
-version = "2.6.3"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

 [[package]]
@@ -18,6 +18,7 @@ from langgraph_sdk.schema import (
    CronSortBy,
    Durability,
    Input,
+    Json,
    OnCompletionBehavior,
    QueryParamTypes,
    Run,
@@ -413,6 +414,7 @@ class CronClient:
        assistant_id: str | None = None,
        thread_id: str | None = None,
        enabled: bool | None = None,
+        metadata: Json = None,
        limit: int = 10,
        offset: int = 0,
        sort_by: CronSortBy | None = None,
@@ -427,6 +429,8 @@ class CronClient:
            assistant_id: The assistant ID or graph name to search for.
            thread_id: the thread ID to search for.
            enabled: The enabled status to search for.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
            limit: The maximum number of results to return.
            offset: The number of results to skip.
            headers: Optional custom headers to include with the request.
@@ -481,6 +485,8 @@ class CronClient:
            "limit": limit,
            "offset": offset,
        }
+        if metadata:
+            payload["metadata"] = metadata
        if sort_by:
            payload["sort_by"] = sort_by
        if sort_order:
@@ -497,6 +503,7 @@ class CronClient:
        *,
        assistant_id: str | None = None,
        thread_id: str | None = None,
+        metadata: Json = None,
        headers: Mapping[str, str] | None = None,
        params: QueryParamTypes | None = None,
    ) -> int:
@@ -505,6 +512,8 @@ class CronClient:
        Args:
            assistant_id: Assistant ID to filter by.
            thread_id: Thread ID to filter by.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
            headers: Optional custom headers to include with the request.
            params: Optional query parameters to include with the request.

@@ -516,6 +525,8 @@ class CronClient:
            payload["assistant_id"] = assistant_id
        if thread_id:
            payload["thread_id"] = thread_id
+        if metadata:
+            payload["metadata"] = metadata
        return await self.http.post(
            "/runs/crons/count", json=payload, headers=headers, params=params
        )
@@ -18,6 +18,7 @@ from langgraph_sdk.schema import (
    CronSortBy,
    Durability,
    Input,
+    Json,
    OnCompletionBehavior,
    QueryParamTypes,
    Run,
@@ -402,6 +403,7 @@ class SyncCronClient:
        assistant_id: str | None = None,
        thread_id: str | None = None,
        enabled: bool | None = None,
+        metadata: Json = None,
        limit: int = 10,
        offset: int = 0,
        sort_by: CronSortBy | None = None,
@@ -416,6 +418,8 @@ class SyncCronClient:
            assistant_id: The assistant ID or graph name to search for.
            thread_id: the thread ID to search for.
            enabled: Whether the cron job is enabled.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
            limit: The maximum number of results to return.
            offset: The number of results to skip.
            headers: Optional custom headers to include with the request.
@@ -468,6 +472,8 @@ class SyncCronClient:
            "limit": limit,
            "offset": offset,
        }
+        if metadata:
+            payload["metadata"] = metadata
        if sort_by:
            payload["sort_by"] = sort_by
        if sort_order:
@@ -484,6 +490,7 @@ class SyncCronClient:
        *,
        assistant_id: str | None = None,
        thread_id: str | None = None,
+        metadata: Json = None,
        headers: Mapping[str, str] | None = None,
        params: QueryParamTypes | None = None,
    ) -> int:
@@ -492,6 +499,8 @@ class SyncCronClient:
        Args:
            assistant_id: Assistant ID to filter by.
            thread_id: Thread ID to filter by.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
            headers: Optional custom headers to include with the request.
            params: Optional query parameters to include with the request.

@@ -503,6 +512,8 @@ class SyncCronClient:
            payload["assistant_id"] = assistant_id
        if thread_id:
            payload["thread_id"] = thread_id
+        if metadata:
+            payload["metadata"] = metadata
        return self.http.post(
            "/runs/crons/count", json=payload, headers=headers, params=params
        )
@@ -485,3 +485,165 @@ def test_sync_update_with_enabled_parameter(enabled_value):
        )

    assert result == cron
+
+
+@pytest.mark.asyncio
+async def test_async_search_with_metadata():
+    """Test that CronClient.search forwards metadata in the request body."""
+    cron = _cron_response()
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/search"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"owner": "alice"}
+        assert body["limit"] == 10
+        assert body["offset"] == 0
+
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.search(metadata={"owner": "alice"})
+
+    assert result == [cron]
+
+
+@pytest.mark.asyncio
+async def test_async_search_omits_empty_metadata():
+    """Test that CronClient.search does not send metadata when not provided."""
+    cron = _cron_response()
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        await cron_client.search()
+
+
+@pytest.mark.asyncio
+async def test_async_count_with_metadata():
+    """Test that CronClient.count forwards metadata in the request body."""
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/count"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"team": "infra"}
+
+        return httpx.Response(200, json=2)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.count(metadata={"team": "infra"})
+
+    assert result == 2
+
+
+@pytest.mark.asyncio
+async def test_async_count_omits_empty_metadata():
+    """Test that CronClient.count does not send metadata when not provided."""
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=0)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        await cron_client.count()
+
+
+def test_sync_search_with_metadata():
+    """Test that SyncCronClient.search forwards metadata in the request body."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/search"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"owner": "alice"}
+
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.search(metadata={"owner": "alice"})
+
+    assert result == [cron]
+
+
+def test_sync_search_omits_empty_metadata():
+    """Test that SyncCronClient.search does not send metadata when not provided."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        cron_client.search()
+
+
+def test_sync_count_with_metadata():
+    """Test that SyncCronClient.count forwards metadata in the request body."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/count"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"team": "infra"}
+
+        return httpx.Response(200, json=2)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.count(metadata={"team": "infra"})
+
+    assert result == 2
+
+
+def test_sync_count_omits_empty_metadata():
+    """Test that SyncCronClient.count does not send metadata when not provided."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=0)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        cron_client.count()
@@ -266,7 +266,7 @@ wheels = [

 [[package]]
 name = "langchain-core"
-version = "1.4.0a2"
+version = "1.4.0"
 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

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

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

 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0,<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.0a4"
+version = "4.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
    { name = "langchain-core" },
@@ -430,7 +430,7 @@ test = [

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