simplify: drop _should_route_to_error_handler, remove functional API error handler, clean up prepare_node_error_handler_task signature

Replace the hidden __error_handler__<name> node approach with a
callable field on PregelNode/PregelExecutableTask, matching how
retry_policy and cache_policy work:

- error_handler: Runnable | None lives on StateNodeSpec, PregelNode,
  and PregelExecutableTask — no separate node registration at compile time
- ErrorHandlerNode type alias added to _node.py covering the common
  handler signatures (state + NodeError, runtime + NodeError, etc.)
- Handler task is synthesized at runtime in schedule_error_handler using
  the failed task's write pipeline (failed_task.writers), so no separate
  PregelNode is needed
- compile(error_handler=...) adds a graph-level fallback handler;
  per-node handlers take precedence
- @task(error_handler=...) in the functional API uses inline try/except
  wrapping so the caller's future always resolves to a value
- interrupt_before/after ["__error_handler__<node>"] still works via
  updated validate() and _validate.py checks
- RunnableCallable gains takes_input flag so handlers whose only
  parameters are injected kwargs (e.g. def h(error: NodeError) -> T)
  are called without a spurious positional input arg

libs/checkpoint-conformance/README.md

@@ -63,6 +63,7 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `delete_for_runs` | no | `adelete_for_runs` |
 | `copy_thread` | no | `acopy_thread` |
 | `prune` | no | `aprune` |
+| `delta_channel_history` | no | `aget_delta_channel_history` |
 
 Extended capabilities are detected by checking whether the method is overridden from `BaseCheckpointSaver`. If not overridden, those tests are skipped.
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py

@@ -23,6 +23,7 @@ class Capability(str, Enum):
     DELETE_FOR_RUNS = "delete_for_runs"
     COPY_THREAD = "copy_thread"
     PRUNE = "prune"
+    DELTA_CHANNEL_HISTORY = "delta_channel_history"
 
 
 # Capabilities that every checkpointer must support.
@@ -42,6 +43,7 @@ EXTENDED_CAPABILITIES = frozenset(
         Capability.DELETE_FOR_RUNS,
         Capability.COPY_THREAD,
         Capability.PRUNE,
+        Capability.DELTA_CHANNEL_HISTORY,
     }
 )
 
@@ -57,6 +59,7 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
     Capability.DELETE_FOR_RUNS: "adelete_for_runs",
     Capability.COPY_THREAD: "acopy_thread",
     Capability.PRUNE: "aprune",
+    Capability.DELTA_CHANNEL_HISTORY: "aget_delta_channel_history",
 }
 
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py

@@ -9,6 +9,9 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_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
@@ -24,4 +27,5 @@ __all__ = [
     "run_delete_for_runs_tests",
     "run_copy_thread_tests",
     "run_prune_tests",
+    "run_delta_channel_history_tests",
 ]

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/_delta_fixtures.py

@@ -0,0 +1,99 @@
+"""Shared fixtures for delta-channel conformance tests.
+
+Builds a parent chain with `_DeltaSnapshot` blobs at known positions via
+direct `aput` / `aput_writes` calls. No langgraph or Pregel dependency.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+from uuid import uuid4
+
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint
+from langgraph.checkpoint.base.id import uuid6
+
+from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+
+async def build_delta_chain(
+    saver: BaseCheckpointSaver,
+    *,
+    thread_id: str | None = None,
+    checkpoint_ns: str = "",
+    channel: str = "messages",
+    snapshots_at_steps: Sequence[int] = (0,),
+    total_steps: int = 6,
+    write_value_fn: Any | None = None,
+) -> list[RunnableConfig]:
+    """Build a parent chain with `_DeltaSnapshot` at known positions.
+
+    Args:
+        saver: Checkpointer instance.
+        thread_id: Defaults to a random UUID.
+        checkpoint_ns: Namespace (default root).
+        channel: Channel name used for snapshots and writes.
+        snapshots_at_steps: Steps at which a `_DeltaSnapshot` blob is stored
+            in `channel_values[channel]`. Step 0 is the oldest checkpoint.
+        total_steps: Number of checkpoints in the chain.
+        write_value_fn: Callable(step) -> write value. Defaults to step index.
+
+    Returns:
+        List of stored configs (oldest first), one per step.
+    """
+    if write_value_fn is None:
+
+        def write_value_fn(step: int) -> Any:
+            return step
+
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    thread_id = thread_id or str(uuid4())
+    snapshot_set = set(snapshots_at_steps)
+    stored: list[RunnableConfig] = []
+    parent_cfg: RunnableConfig | None = None
+
+    for step in range(total_steps):
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+            }
+        }
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+
+        channel_values: dict[str, Any] = {}
+        channel_versions: dict[str, int] = {}
+        if step in snapshot_set:
+            channel_values[channel] = _DeltaSnapshot(
+                write_value_fn(step),
+            )
+            channel_versions[channel] = step + 1
+
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=channel_values,
+            channel_versions=channel_versions,
+            versions_seen={},
+            updated_channels=None,
+        )
+        new_versions = dict(channel_versions)
+        parent_cfg = await saver.aput(
+            config, cp, generate_metadata(step=step), new_versions
+        )
+        stored.append(parent_cfg)
+
+        # Write a pending write for non-snapshot steps so the walk has
+        # something to collect.
+        if step not in snapshot_set:
+            await saver.aput_writes(
+                parent_cfg, [(channel, write_value_fn(step))], str(uuid4())
+            )
+
+    return stored

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -0,0 +1,247 @@
+"""DELTA_CHANNEL_HISTORY capability tests — aget_delta_channel_history 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_history_returns_writes_oldest_first(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes are returned oldest-to-newest."""
+    tid = str(uuid4())
+    # 5 steps: snapshot at 0, writes at 1,2,3,4.
+    # Head is step 4. Walk starts at step 3 (parent of head).
+    # Collects writes from steps 1,2,3 (between snapshot at 0 and head's parent).
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [1, 2, 3], f"Expected [1,2,3], got {values}"
+
+
+async def test_history_seed_is_nearest_snapshot(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Seed is the value from the nearest ancestor with channel_values populated."""
+    tid = str(uuid4())
+    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
+    # Head is step 5. Walk from step 4 backward stops at step 3 (snapshot).
+    # Collects writes from step 4 only (between step 3 and head's parent step 4).
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[0, 3],
+        total_steps=6,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" in result["ch"], "Expected seed from snapshot at step 3"
+    seed = result["ch"]["seed"]
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    actual_value = seed.value if isinstance(seed, _DeltaSnapshot) else seed
+    assert actual_value == 3, f"Expected seed value 3 (step 3), got {actual_value}"
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [4], f"Expected [4], got {values}"
+
+
+async def test_history_excludes_target_pending_writes(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Target's own pending_writes are NOT included in the history."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    head = configs[-1]
+    # Add writes directly to the head checkpoint
+    await saver.aput_writes(head, [("ch", "extra")], str(uuid4()))
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert "extra" not in values, f"Target's writes should be excluded, got {values}"
+
+
+async def test_history_multi_channel(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Multiple channels have independent walk termination."""
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+    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
+
+    for step in range(5):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        if step == 1:
+            cv["a"] = _DeltaSnapshot("snap_a")
+            cvs["a"] = step + 1
+        if step == 3:
+            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)
+        await saver.aput_writes(parent_cfg, [("a", step), ("b", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["a", "b"])
+    a_writes = [w[2] for w in result["a"]["writes"]]
+    b_writes = [w[2] for w in result["b"]["writes"]]
+    assert a_writes == [1, 2, 3], f"Expected a writes [1,2,3], got {a_writes}"
+    assert b_writes == [3], f"Expected b writes [3], got {b_writes}"
+
+
+async def test_history_empty_channels_returns_empty(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Empty channels list returns empty mapping."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    result = await saver.aget_delta_channel_history(config=configs[-1], channels=[])
+    assert result == {}
+
+
+async def test_history_walk_to_root_no_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Walk reaches root without finding seed — no 'seed' key in result."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[],
+        total_steps=4,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" not in result["ch"], f"Expected no seed, got {result['ch']}"
+
+
+async def test_history_migration_plain_value_as_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Pre-delta plain value in channel_values acts as seed (migration case).
+
+    When a thread was originally using a regular channel (BinaryOperatorAggregate)
+    and later switches to DeltaChannel, the old checkpoint has a plain value in
+    channel_values[ch] (not a _DeltaSnapshot). The walk should treat it as the
+    seed and terminate there.
+    """
+    from langgraph.checkpoint.base import Checkpoint
+    from langgraph.checkpoint.base.id import uuid6
+
+    from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+
+    for step in range(4):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        # Step 1: plain value (migration case — old checkpoint before delta)
+        if step == 1:
+            cv["ch"] = [10, 20, 30]
+            cvs["ch"] = 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)
+        if step != 1:
+            await saver.aput_writes(parent_cfg, [("ch", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    # Seed should be the plain value from step 1
+    assert "seed" in result["ch"], "Expected seed from migration plain value at step 1"
+    seed = result["ch"]["seed"]
+    assert seed == [10, 20, 30], f"Expected plain value [10,20,30], got {seed}"
+    # Writes should be from step 2 only (between seed at step 1 and head's parent step 2)
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [2], f"Expected [2], got {values}"
+
+
+ALL_DELTA_CHANNEL_HISTORY_TESTS = [
+    test_history_returns_writes_oldest_first,
+    test_history_seed_is_nearest_snapshot,
+    test_history_excludes_target_pending_writes,
+    test_history_multi_channel,
+    test_history_empty_channels_returns_empty,
+    test_history_walk_to_root_no_seed,
+    test_history_migration_plain_value_as_seed,
+]
+
+
+async def run_delta_channel_history_tests(
+    saver: BaseCheckpointSaver,
+    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
+) -> tuple[int, int, list[str]]:
+    """Run all delta_channel_history tests. Returns (passed, failed, failure_names)."""
+    passed = 0
+    failed = 0
+    failures: list[str] = []
+    for test_fn in ALL_DELTA_CHANNEL_HISTORY_TESTS:
+        try:
+            await test_fn(saver)
+            passed += 1
+            if on_test_result:
+                on_test_result("delta_channel_history", 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_history",
+                    test_fn.__name__,
+                    False,
+                    traceback.format_exc(),
+                )
+    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py

@@ -19,6 +19,9 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_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
@@ -35,6 +38,7 @@ _RUNNERS = {
     Capability.DELETE_FOR_RUNS: run_delete_for_runs_tests,
     Capability.COPY_THREAD: run_copy_thread_tests,
     Capability.PRUNE: run_prune_tests,
+    Capability.DELTA_CHANNEL_HISTORY: run_delta_channel_history_tests,
 }
 
 

libs/checkpoint-conformance/pyproject.toml

@@ -43,7 +43,11 @@ asyncio_mode = "auto"
 # The extended methods (acopy_thread, adelete_for_runs, aprune) are checked
 # at runtime via capability detection and may not exist on the installed
 # base class. Dict literal inference is also overly strict for RunnableConfig.
+# Delta-channel tests import from `langgraph` (not a declared dep of this
+# package — at test time it is installed alongside); private `_DeltaSnapshot`
+# imports are intentional (beta surface).
 unresolved-attribute = "ignore"
+unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
 
@@ -58,6 +62,9 @@ lint.select = [
 lint.ignore = ["E501", "B008"]
 target-version = "py310"
 
+[tool.uv.sources]
+langgraph-checkpoint = {path = "../checkpoint", editable = true}
+
 [[tool.uv.index]]
 name = "testpypi"
 url = "https://test.pypi.org/simple/"

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -279,8 +279,8 @@ def _build_delta_stage2_sql(
         )
     for _ in channels_with_seed:
         branches.append(
-            "SELECT 'b'::text, NULL, channel, "
-            "type, blob, NULL, NULL, version "
+            "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
+            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"

libs/checkpoint-postgres/uv.lock

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

libs/checkpoint-sqlite/tests/test_conformance_delta.py

@@ -0,0 +1,35 @@
+"""Run delta-channel conformance capabilities against AsyncSqliteSaver."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+pytest.importorskip("aiosqlite", reason="aiosqlite not installed")
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+    @checkpointer_test(name="AsyncSqliteSaver")
+    async def sqlite_saver():
+        async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+            yield saver
+
+    report = await validate(
+        sqlite_saver,
+        capabilities={
+            "delta_channel_history",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")

libs/checkpoint-sqlite/uv.lock

@@ -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,9 +266,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]]
@@ -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]]

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -60,15 +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.
 
-    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.
+    !!! warning "Beta"
+
+        This metadata field backs `DeltaChannel` (beta). The key name and
+        contents may change while the delta-channel design stabilizes.
+
+    Maps channel name -> `(updates, supersteps)`:
+
+    - index 0 (`updates`): number of supersteps that wrote to this channel
+      since its last snapshot blob.
+    - index 1 (`supersteps`): total supersteps elapsed since this channel's
+      last snapshot, regardless of whether the channel was written.
+
+    A snapshot fires when EITHER `updates >= ch.snapshot_frequency` OR
+    `supersteps >= DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT` (system-wide bound,
+    default 5000, env `LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT`).
+    The supersteps bound prevents unbounded ancestor walks on threads where
+    a delta channel exists but is no longer being updated.
+
+    Absent on threads that don't use delta channels. Persisted as a
+    2-element list in JSON (no native tuple).
     """
 
 
@@ -135,6 +149,11 @@ class CheckpointTuple(NamedTuple):
 class DeltaChannelHistory(TypedDict):
     """Per-channel result entry from `BaseCheckpointSaver.get_delta_channel_history`.
 
+    !!! warning "Beta"
+
+        Part of the `DeltaChannel` support surface; in beta. Field names and
+        semantics may change.
+
     Storage-level view of what one channel contributed across the ancestor
     chain of a target checkpoint:
 
@@ -317,6 +336,14 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            Deleting a run that produced ancestor `checkpoint_writes` — or
+            the only `_DeltaSnapshot` blob — for a still-live thread will
+            break reconstruction of any `DeltaChannel` whose history
+            depended on those rows. See the `DeltaChannel` note on `prune`
+            for safe-recovery strategies.
         """
         raise NotImplementedError
 
@@ -330,6 +357,17 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            Implementations must copy the **complete** parent chain (all
+            ancestor checkpoints and their `checkpoint_writes`) — copying
+            only the head checkpoint will leave the target thread with
+            `DeltaChannel` state that cannot be reconstructed (no path back
+            to a `_DeltaSnapshot` ancestor). Equivalently, the copy must
+            include enough ancestors that every `DeltaChannel`-backed key
+            has either a `_DeltaSnapshot` in `channel_values` somewhere in
+            the chain, or a complete write history back to the chain root.
         """
         raise NotImplementedError
 
@@ -345,6 +383,34 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            Custom implementations must be `DeltaChannel`-aware. `DeltaChannel`
+            stores only a sentinel in `channel_values` for non-snapshot steps;
+            reconstruction walks the parent chain via
+            `get_delta_channel_history`, accumulating rows from
+            `checkpoint_writes` until it reaches an ancestor whose
+            `channel_values` contains a `_DeltaSnapshot` blob (written every
+            `snapshot_frequency` updates).
+
+            A naive `"keep_latest"` that drops intermediate checkpoints and
+            their writes can sever that chain: the surviving "latest"
+            checkpoint is rarely a snapshot point itself, so its delta
+            channels would silently reconstruct as empty (no error raised —
+            `get_delta_channel_history` simply returns no `seed`). Safe
+            options when the graph uses `DeltaChannel`:
+
+            * Walk back from each kept checkpoint and preserve every
+              ancestor (plus its `checkpoint_writes`) up to the nearest one
+              whose `channel_values` already contains a `_DeltaSnapshot` for
+              every `DeltaChannel`-backed key.
+            * Force a fresh snapshot on the kept checkpoint before deleting
+              ancestors — rewrite `channel_values[k] = _DeltaSnapshot(value)`
+              for each delta channel `k` (resolving `value` via the existing
+              ancestor walk first), then prune.
+            * Skip pruning threads whose graph uses `DeltaChannel` until one
+              of the above is implemented.
         """
         raise NotImplementedError
 
@@ -461,6 +527,13 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            See `delete_for_runs` — deleting rows a still-live thread's
+            `DeltaChannel` reconstruction depends on (writes between the
+            head and its nearest `_DeltaSnapshot` ancestor) will silently
+            corrupt that channel's state.
         """
         raise NotImplementedError
 
@@ -474,6 +547,13 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            See `copy_thread` — the copy must carry the complete parent
+            chain (or at least back to a `_DeltaSnapshot` ancestor for every
+            `DeltaChannel`) so the target thread can reconstruct delta
+            state.
         """
         raise NotImplementedError
 
@@ -489,6 +569,13 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            See `prune` for the full `DeltaChannel` caveat. In short:
+            `"keep_latest"` must not drop ancestor checkpoints / writes that
+            sit between the kept checkpoint and the nearest `_DeltaSnapshot`
+            ancestor, or delta channels will silently reconstruct as empty.
         """
         raise NotImplementedError
 
@@ -497,6 +584,14 @@ class BaseCheckpointSaver(Generic[V]):
     ) -> Mapping[str, DeltaChannelHistory]:
         """Walk the parent chain returning per-channel writes + seed.
 
+        !!! warning "Beta"
+
+            This method is part of the `DeltaChannel` support surface and is
+            in beta. The signature, return shape (`DeltaChannelHistory`), and
+            interaction with `_DeltaSnapshot` blobs may change. Override at
+            your own risk; the default implementation will continue to work
+            against the public `BaseCheckpointSaver` contract.
+
         For each requested channel, walks ancestors of the checkpoint
         identified by `config` (following `parent_config`) and accumulates
         `pending_writes` for that channel. The walk terminates per-channel
@@ -556,7 +651,13 @@ class BaseCheckpointSaver(Generic[V]):
     async def aget_delta_channel_history(
         self, *, config: RunnableConfig, channels: Sequence[str]
     ) -> Mapping[str, DeltaChannelHistory]:
-        """Async version of `get_delta_channel_history`."""
+        """Async version of `get_delta_channel_history`.
+
+        !!! warning "Beta"
+
+            This method is part of the `DeltaChannel` support surface and is
+            in beta. See `get_delta_channel_history` for caveats.
+        """
         if not channels:
             return {}
         collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}

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

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
         AllowedMsgpackModules,
     )
 
-LC_REVIVER = Reviver()
+LC_REVIVER = Reviver(allowed_objects="core")
 EMPTY_BYTES = b""
 logger = logging.getLogger(__name__)
 

libs/checkpoint/tests/test_conformance_delta.py

@@ -0,0 +1,33 @@
+"""Run delta-channel conformance capabilities against InMemorySaver."""
+
+from __future__ import annotations
+
+import pytest
+
+conformance = pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    @checkpointer_test(name="InMemorySaver")
+    async def mem_saver():
+        yield InMemorySaver()
+
+    report = await validate(
+        mem_saver,
+        capabilities={
+            "delta_channel_history",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")

libs/cli/langgraph_cli/__init__.py

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

libs/cli/uv.lock

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

libs/langgraph/langgraph/_internal/_config.py

@@ -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:

libs/langgraph/langgraph/channels/delta.py

@@ -26,6 +26,15 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
     """Reducer channel that stores only a sentinel in checkpoint blobs and
     reconstructs state by replaying ancestor writes through the reducer.
 
+    !!! warning "Beta"
+
+        `DeltaChannel` is in beta. The API and on-disk representation may
+        change in future releases. Threads written with `DeltaChannel` today
+        are expected to remain readable, but the surrounding contract
+        (`BaseCheckpointSaver.get_delta_channel_history`, the
+        `_DeltaSnapshot` blob shape, the `counters_since_delta_snapshot`
+        metadata field) is not yet stable.
+
     The reducer receives the current accumulated value and a batch of writes
     in one call: `reducer(state, [write1, write2, ...]) -> new_state`.
 
@@ -38,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

libs/langgraph/langgraph/graph/_node.py

@@ -8,6 +8,7 @@ from langchain_core.runnables import Runnable, RunnableConfig
 from langgraph.store.base import BaseStore
 
 from langgraph._internal._typing import EMPTY_SEQ
+from langgraph.errors import NodeError
 from langgraph.runtime import Runtime
 from langgraph.types import CachePolicy, RetryPolicy, StreamWriter, TimeoutPolicy
 from langgraph.typing import ContextT, NodeInputT, NodeInputT_contra
@@ -64,6 +65,22 @@ class _NodeWithRuntime(Protocol[NodeInputT_contra, ContextT]):
     ) -> Any: ...
 
 
+class _NodeWithNodeError(Protocol[NodeInputT_contra]):
+    def __call__(self, state: NodeInputT_contra, *, error: NodeError) -> Any: ...
+
+
+class _NodeWithConfigNodeError(Protocol[NodeInputT_contra]):
+    def __call__(
+        self, state: NodeInputT_contra, *, config: RunnableConfig, error: NodeError
+    ) -> Any: ...
+
+
+class _NodeWithRuntimeNodeError(Protocol[NodeInputT_contra, ContextT]):
+    def __call__(
+        self, state: NodeInputT_contra, *, runtime: Runtime[ContextT], error: NodeError
+    ) -> Any: ...
+
+
 # TODO: we probably don't want to explicitly support the config / store signatures once
 # we move to adding a context arg. Maybe what we do is we add support for kwargs with param spec
 # this is purely for typing purposes though, so can easily change in the coming weeks.
@@ -80,6 +97,13 @@ StateNode: TypeAlias = (
     | Runnable[NodeInputT, Any]
 )
 
+ErrorHandlerNode: TypeAlias = (
+    StateNode[NodeInputT, ContextT]
+    | _NodeWithNodeError[NodeInputT]
+    | _NodeWithConfigNodeError[NodeInputT]
+    | _NodeWithRuntimeNodeError[NodeInputT, ContextT]
+)
+
 
 @dataclass(slots=True)
 class StateNodeSpec(Generic[NodeInputT, ContextT]):
@@ -88,8 +112,7 @@ class StateNodeSpec(Generic[NodeInputT, ContextT]):
     input_schema: type[NodeInputT]
     retry_policy: RetryPolicy | Sequence[RetryPolicy] | None
     cache_policy: CachePolicy | None
-    is_error_handler: bool = False
-    error_handler_node: str | None = None
+    error_handler: Runnable[Any, Any] | None = None
     ends: tuple[str, ...] | dict[str, str] | None = EMPTY_SEQ
     defer: bool = False
     timeout: TimeoutPolicy | None = None

libs/langgraph/langgraph/graph/state.py

@@ -65,7 +65,7 @@ from langgraph.errors import (
     create_error_message,
 )
 from langgraph.graph._branch import BranchSpec
-from langgraph.graph._node import StateNode, StateNodeSpec
+from langgraph.graph._node import ErrorHandlerNode, StateNode, StateNodeSpec
 from langgraph.managed.base import (
     ManagedValueSpec,
     is_managed_value,
@@ -772,24 +772,15 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         if destinations is not None:
             ends = destinations
 
-        resolved_input_schema: type[Any] = (
-            input_schema or inferred_input_schema or self.state_schema
-        )
-        handler_node_name: str | None = None
-        if error_handler is not None:
-            handler_node_name = f"__error_handler__{node}"
-            if handler_node_name in self.nodes:
-                raise ValueError(
-                    f"Auto-generated error handler node `{handler_node_name}` already exists."
-                )
-            self.nodes[handler_node_name] = StateNodeSpec[Any, ContextT](
-                coerce_to_runnable(error_handler, name=handler_node_name, trace=False),  # type: ignore[arg-type]
-                metadata=None,
-                input_schema=resolved_input_schema,
-                retry_policy=None,
-                cache_policy=None,
-                is_error_handler=True,
+        coerced_error_handler: Runnable[Any, Any] | None = (
+            coerce_to_runnable(  # type: ignore[arg-type]
+                error_handler,
+                name=f"__error_handler__{node}",
+                trace=False,
             )
+            if error_handler is not None
+            else None
+        )
 
         if input_schema is not None:
             self.nodes[node] = StateNodeSpec[NodeInputT, ContextT](
@@ -798,7 +789,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 input_schema=input_schema,
                 retry_policy=retry_policy,
                 cache_policy=cache_policy,
-                error_handler_node=handler_node_name,
+                error_handler=coerced_error_handler,
                 ends=ends,
                 defer=defer,
                 timeout=timeout,
@@ -810,7 +801,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 input_schema=inferred_input_schema,
                 retry_policy=retry_policy,
                 cache_policy=cache_policy,
-                error_handler_node=handler_node_name,
+                error_handler=coerced_error_handler,
                 ends=ends,
                 defer=defer,
                 timeout=timeout,
@@ -822,7 +813,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 input_schema=self.state_schema,
                 retry_policy=retry_policy,
                 cache_policy=cache_policy,
-                error_handler_node=handler_node_name,
+                error_handler=coerced_error_handler,
                 ends=ends,
                 defer=defer,
                 timeout=timeout,
@@ -1079,7 +1070,17 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         if interrupt:
             for node in interrupt:
                 if node not in self.nodes:
-                    raise ValueError(f"Interrupt node `{node}` not found")
+                    # __error_handler__<name> is a valid virtual task name when the
+                    # base node has an error_handler configured.
+                    if node.startswith("__error_handler__"):
+                        base = node[len("__error_handler__"):]
+                        if (
+                            base not in self.nodes
+                            or self.nodes[base].error_handler is None
+                        ):
+                            raise ValueError(f"Interrupt node `{node}` not found")
+                    else:
+                        raise ValueError(f"Interrupt node `{node}` not found")
         self.compiled = True
         return self
 
@@ -1094,6 +1095,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         debug: bool = False,
         name: str | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+        error_handler: ErrorHandlerNode[Any, ContextT] | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 
@@ -1193,11 +1195,15 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 key for key, val in self.channels.items() if not is_managed_value(val)
             ]
         )
-        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
-        }
+        error_handler: Runnable[Any, Any] | None = (
+            coerce_to_runnable(  # type: ignore[arg-type]
+                error_handler,
+                name="__graph_error_handler__",
+                trace=False,
+            )
+            if error_handler is not None
+            else None
+        )
 
         compiled = CompiledStateGraph[StateT, ContextT, InputT, OutputT](
             builder=self,
@@ -1220,7 +1226,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             debug=debug,
             store=store,
             cache=cache,
-            node_error_handler_map=node_error_handler_map,
+            error_handler=error_handler,
             name=name or "LangGraph",
             stream_transformers=transformers,
         )
@@ -1395,8 +1401,7 @@ class CompiledStateGraph(
                 metadata=node.metadata,
                 retry_policy=node.retry_policy,
                 cache_policy=node.cache_policy,
-                is_error_handler=node.is_error_handler,
-                error_handler_node=node.error_handler_node,
+                error_handler=node.error_handler,
                 bound=node.runnable,  # type: ignore[arg-type]
                 timeout=node.timeout,
             )

libs/langgraph/langgraph/pregel/_algo.py

@@ -20,6 +20,7 @@ from typing import (
 
 from langchain_core.callbacks import Callbacks
 from langchain_core.callbacks.manager import AsyncParentRunManager, ParentRunManager
+from langchain_core.runnables import Runnable
 from langchain_core.runnables.config import RunnableConfig
 from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
@@ -32,6 +33,7 @@ from langgraph.store.base import BaseStore
 from xxhash import xxh3_128_hexdigest
 
 from langgraph._internal._config import merge_configs, patch_config
+from langgraph._internal._runnable import RunnableSeq
 from langgraph._internal._constants import (
     CACHE_NS_WRITES,
     CONF,
@@ -71,6 +73,7 @@ from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import NodeError
 from langgraph.managed.base import ManagedValueMapping
 from langgraph.pregel._call import get_runnable_for_task, identifier
+from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.pregel._io import read_channels
 from langgraph.pregel._log import logger
 from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode
@@ -407,6 +410,7 @@ def prepare_next_tasks(
     updated_channels: set[str] | None = None,
     retry_policy: Sequence[RetryPolicy] = (),
     cache_policy: CachePolicy | None = None,
+    error_handler: Runnable[Any, Any] | None = None,
 ) -> dict[str, PregelTask] | dict[str, PregelExecutableTask]:
     """Prepare the set of tasks that will make up the next Pregel step.
 
@@ -462,6 +466,7 @@ def prepare_next_tasks(
                 input_cache=input_cache,
                 cache_policy=cache_policy,
                 retry_policy=retry_policy,
+                error_handler=error_handler,
             ):
                 tasks.append(task)
 
@@ -508,6 +513,7 @@ def prepare_next_tasks(
             input_cache=input_cache,
             cache_policy=cache_policy,
             retry_policy=retry_policy,
+            error_handler=error_handler,
         ):
             tasks.append(task)
     return {t.id: t for t in tasks}
@@ -542,6 +548,7 @@ def prepare_single_task(
     input_cache: dict[INPUT_CACHE_KEY_TYPE, Any] | None = None,
     cache_policy: CachePolicy | None = None,
     retry_policy: Sequence[RetryPolicy] = (),
+    error_handler: Runnable[Any, Any] | None = None,
 ) -> None | PregelTask | PregelExecutableTask:
     """Prepares a single task for the next Pregel step, given a task path, which
     uniquely identifies a PUSH or PULL task within the graph."""
@@ -756,6 +763,7 @@ def prepare_single_task(
                         writers=proc.flat_writers,
                         subgraphs=proc.subgraphs,
                         timeout=proc.timeout,
+                        error_handler=proc.error_handler or error_handler,
                     )
             else:
                 return PregelTask(task_id, name, task_path[:3])
@@ -1110,11 +1118,10 @@ def prepare_push_task_send(
 def prepare_node_error_handler_task(
     failed_task: PregelExecutableTask,
     *,
-    handler_node_name: str,
+    handler: Runnable,
     failed_error: BaseException,
     checkpoint: Checkpoint,
     pending_writes: list[PendingWrite],
-    processes: Mapping[str, PregelNode],
     channels: Mapping[str, BaseChannel],
     managed: ManagedValueMapping,
     config: RunnableConfig,
@@ -1123,17 +1130,14 @@ def prepare_node_error_handler_task(
     store: BaseStore | None = None,
     checkpointer: BaseCheckpointSaver | None = None,
     manager: None | ParentRunManager | AsyncParentRunManager = None,
-    cache_policy: CachePolicy | None = None,
     retry_policy: Sequence[RetryPolicy] = (),
-) -> PregelExecutableTask | None:
-    """Prepare an immediate node-level error handler task for a failed task."""
-    if handler_node_name not in processes:
-        return None
-    proc = processes[handler_node_name]
-    proc_node = proc.node
-    if proc_node is None:
-        return None
+) -> PregelExecutableTask:
+    """Prepare an error handler task for a failed task.
 
+    The handler borrows the failed task's write pipeline (same state channels),
+    so no separate node registration is needed.
+    """
+    handler_node_name = f"__error_handler__{failed_task.name}"
     checkpoint_id_bytes = binascii.unhexlify(checkpoint["id"].replace("-", ""))
     task_id_func = _xxhash_str if checkpoint["v"] > 1 else _uuid5_str
     configurable = config.get(CONF, {})
@@ -1159,27 +1163,17 @@ def prepare_node_error_handler_task(
         "langgraph_path": translated_task_path,
         "langgraph_checkpoint_ns": task_checkpoint_ns,
     }
-    if proc.metadata:
-        metadata.update(proc.metadata)
     writes: deque[tuple[str, Any]] = deque()
-
-    effective_retry_policy = proc.retry_policy or retry_policy
-    effective_cache_policy = proc.cache_policy or cache_policy
-    if effective_cache_policy:
-        args_key = effective_cache_policy.key_func(failed_task.input)
-        cache_key = CacheKey(
-            (
-                CACHE_NS_WRITES,
-                (identifier(proc) or "__dynamic__"),
-                handler_node_name,
-            ),
-            xxh3_128_hexdigest(
-                args_key.encode() if isinstance(args_key, str) else args_key
-            ),
-            effective_cache_policy.ttl,
-        )
+    # Mirror how regular node procs are built: combine handler with a write pipeline
+    # so run_with_retry invokes the full pipeline in one shot.
+    # - PULL node tasks: writers are in failed_task.writers → reuse them
+    # - PUSH functional tasks: writers are embedded in proc (empty writers list) →
+    #   add a RETURN write so the handler's result becomes the future's value.
+    handler_writers = failed_task.writers
+    if handler_writers:
+        handler_proc: Runnable = RunnableSeq(handler, *handler_writers)
     else:
-        cache_key = None
+        handler_proc = RunnableSeq(handler, ChannelWrite([ChannelWriteEntry(RETURN)]))
 
     scratchpad = _scratchpad(
         config[CONF].get(CONFIG_KEY_SCRATCHPAD),
@@ -1194,14 +1188,11 @@ def prepare_node_error_handler_task(
     runtime = runtime.override(
         store=store, previous=checkpoint["channel_values"].get(PREVIOUS, None)
     )
-    additional_config: RunnableConfig = {
-        "metadata": metadata,
-        "tags": proc.tags,
-    }
+    additional_config: RunnableConfig = {"metadata": metadata}
     return PregelExecutableTask(
         handler_node_name,
         failed_task.input,
-        proc_node,
+        handler_proc,
         writes,
         patch_config(
             merge_configs(config, additional_config),
@@ -1239,12 +1230,11 @@ def prepare_node_error_handler_task(
             },
         ),
         PUSH_TRIGGER,
-        effective_retry_policy,
-        cache_key,
+        retry_policy,
+        None,  # handlers don't cache
         task_id,
         translated_task_path,
-        writers=proc.flat_writers,
-        subgraphs=proc.subgraphs,
+        writers=handler_writers,  # for ParentCommand / subgraph routing
     )
 
 

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -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
@@ -34,28 +34,28 @@ def empty_checkpoint() -> Checkpoint:
     )
 
 
-def _should_snapshot_delta(
-    name: str,
-    ch: DeltaChannel,
-    updates_since_snapshot: Mapping[str, int],
-    *,
-    force: bool,
-) -> bool:
-    """Decide whether `ch` should write a `_DeltaSnapshot` this step.
+def delta_channels_to_snapshot(
+    channels: Mapping[str, BaseChannel],
+    counters_since_delta_snapshot: Mapping[str, tuple[int, int]],
+) -> set[str]:
+    """Return the set of DeltaChannel names that should snapshot now.
 
-    Triggers:
-      * `force` — always snapshot (used by `durability="exit"`).
-      * Update-count: this channel has accumulated at least
-        `snapshot_frequency` updates since its last snapshot. The count
-        is supplied by the caller via `updates_since_snapshot[name]` and
-        is reset to `0` whenever a snapshot fires.
-
-    Version-format-independent: works for `int`, `float`, and `str`
-    versioning schemes alike.
+    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.
     """
-    if force:
-        return True
-    return updates_since_snapshot.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(
@@ -66,34 +66,19 @@ def create_checkpoint(
     id: str | None = None,
     updated_channels: set[str] | None = None,
     get_next_version: GetNextVersion | None = None,
-    force_delta_snapshot: bool = False,
-    updates_since_snapshot: Mapping[str, int] | None = None,
-    new_updates_since_snapshot: dict[str, int] | None = None,
+    channels_to_snapshot: set[str] | None = None,
 ) -> Checkpoint:
-    """Create a checkpoint for the given channels.
+    """Build a new Checkpoint from the previous one and live channel state.
 
-    For each `DeltaChannel`, a `_DeltaSnapshot(value)` blob is written into
-    `channel_values[k]` when this channel has accumulated at least
-    `snapshot_frequency` updates since its last snapshot (counter supplied
-    via `updates_since_snapshot`). Otherwise the channel is omitted from
-    `channel_values`; its `channel_versions` entry still bumps so that the
-    saver tracks the channel and the ancestor walk can replay writes.
-
-    Snapshots are eager: even if the channel had no write this step, a
-    version bump is forced (via `get_next_version`) so `put()` includes
-    the channel in `new_versions` and stores the blob.
-
-    `force_delta_snapshot` ignores the cadence and always snapshots —
-    used by `durability="exit"` where intermediate writes are not stored
-    as ancestor `checkpoint_writes`.
-
-    If `new_updates_since_snapshot` is provided, the function resets the
-    counter to `0` for any channel that snapshotted this step. Counters
-    for channels that did not snapshot are left untouched (the caller is
-    responsible for incrementing them based on `updated_channels`).
+    For each name in `channels_to_snapshot`, a `_DeltaSnapshot(value)` blob
+    is written into `channel_values[k]`. Other delta channels are omitted
+    from `channel_values` — the ancestor walk reconstructs their state
+    from `checkpoint_writes`. Callers compute the set via
+    `delta_channels_to_snapshot(channels, counters)`; defaults to empty
+    (no snapshots) when not provided.
     """
     ts = datetime.now(timezone.utc).isoformat()
-    counts = updates_since_snapshot or {}
+    channels_to_snapshot = channels_to_snapshot or set()
     if channels is None:
         values = checkpoint["channel_values"]
         channel_versions = checkpoint["channel_versions"]
@@ -104,25 +89,23 @@ def create_checkpoint(
             if k not in channel_versions:
                 continue
             ch = channels[k]
-            if (
-                isinstance(ch, DeltaChannel)
-                and ch.is_available()
-                and _should_snapshot_delta(
-                    k,
-                    ch,
-                    counts,
-                    force=force_delta_snapshot,
-                )
-            ):
-                # Eager snapshot: bump version if not already written this step
-                # so put() includes this channel in new_versions and stores blob.
+            if k in channels_to_snapshot:
+                # In exit mode, the snapshot decision is deferred to exit
+                # time (intermediate steps have do_checkpoint=False). The
+                # channel's count may have reached snapshot_frequency over
+                # several supersteps, but the LAST superstep may not have
+                # written to this channel. In that case apply_writes()
+                # (in _algo.py) didn't bump this channel's version, so
+                # saver.put() wouldn't include it in new_versions and
+                # the snapshot blob would be silently dropped. The manual
+                # bump below closes the gap. In sync/async durability this
+                # branch is effectively dead code (the step that pushes
+                # the count to freq always writes the channel).
                 if get_next_version is not None and (
                     updated_channels is None or k not in updated_channels
                 ):
                     channel_versions[k] = get_next_version(channel_versions[k], None)
                 values[k] = _DeltaSnapshot(ch.get())
-                if new_updates_since_snapshot is not None:
-                    new_updates_since_snapshot[k] = 0
             else:
                 v = ch.checkpoint()
                 if v is not MISSING:
@@ -253,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 {})

libs/langgraph/langgraph/pregel/_loop.py

@@ -22,7 +22,8 @@ from typing import (
 )
 
 from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
-from langchain_core.runnables import RunnableConfig
+from langchain_core.runnables import Runnable, RunnableConfig
+
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
     WRITES_IDX_MAP,
@@ -100,6 +101,7 @@ from langgraph.pregel._checkpoint import (
     channels_from_checkpoint,
     copy_checkpoint,
     create_checkpoint,
+    delta_channels_to_snapshot,
     empty_checkpoint,
 )
 from langgraph.pregel._executor import (
@@ -194,8 +196,40 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
-    # Only set on AsyncPregelLoop; sync loops keep this as None.
+    # Futures from `checkpointer.put_writes` calls that produced delta-channel
+    # writes. `_checkpointer_put_after_previous` drains this list (swap to a
+    # local `futs` then reset to `[]` and wait/gather) before putting the
+    # next checkpoint, so a checkpoint never becomes durable before the
+    # writes that produced it. Initialised to `[]` in both sync and async
+    # `__enter__`; stays `None` only when no checkpointer.
     _delta_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
+    # that will snapshot, then persists the rest under an anchor parent.
+    # `None` when not in exit mode (so the capture sites are no-ops).
+    # Each tuple is `(step, task_id, channel, value)` — `step` drives the
+    # synthetic step-prefixed task_id used to preserve chronological order
+    # under the saver's `ORDER BY task_id, idx` sorting.
+    _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None
+
+    # The checkpoint_config that points at the parent loaded at `__enter__`
+    # (or the synthetic-empty checkpoint, on first run). We capture it
+    # eagerly because every `_put_checkpoint` advances `self.checkpoint_config`
+    # to the newly-saved checkpoint's id — by exit time the original parent
+    # config would otherwise be lost. `_put_exit_delta_writes` uses this:
+    # on resumed runs as the anchor for exit delta writes; on first runs
+    # to derive the lazy stub's config (its `checkpoint_id` is the
+    # synthetic-empty id we want the stub persisted under).
+    _initial_checkpoint_config: RunnableConfig
+
+    # True iff the saver actually returned a tuple at `__enter__`. False
+    # on the first-ever run for a thread (no parent persisted yet).
+    # `_put_exit_delta_writes` uses this to decide between anchoring on
+    # the existing parent (True) or creating a lazy stub (False).
+    _has_persisted_parent: bool = False
+
     managed: ManagedValueMapping
     checkpoint: Checkpoint
     checkpoint_id_saved: str
@@ -246,6 +280,7 @@ class PregelLoop:
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        error_handler: Runnable[Any, Any] | None = None,
         has_graph_lifecycle_callbacks: bool = False,
     ) -> None:
         self.stream = stream
@@ -270,6 +305,7 @@ class PregelLoop:
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
+        self.error_handler = error_handler
         self.durability = durability
         self._has_graph_lifecycle_callbacks = has_graph_lifecycle_callbacks
         self._graph_lifecycle_events = deque()
@@ -512,6 +548,7 @@ class PregelLoop:
                 manager=self.manager,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                error_handler=self.error_handler,
             ),
         ):
             # produce debug output
@@ -564,6 +601,7 @@ class PregelLoop:
             updated_channels=self.updated_channels,
             retry_policy=self.retry_policy,
             cache_policy=self.cache_policy,
+            error_handler=self.error_handler,
         )
 
         # produce debug output
@@ -637,6 +675,11 @@ class PregelLoop:
             self._emit(
                 "values", map_output_values, self.output_keys, writes, self.channels
             )
+        # capture delta-channel writes for exit-mode accumulator before clearing
+        if self._exit_delta_writes is not None:
+            for tid, ch, v in self.checkpoint_pending_writes:
+                if isinstance(self.specs.get(ch), DeltaChannel):
+                    self._exit_delta_writes.append((self.step, tid, ch, v))
         # clear pending writes
         self.checkpoint_pending_writes.clear()
         # only replay (re-execute) done tasks on the first tick
@@ -854,6 +897,27 @@ class PregelLoop:
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
             )
+            # Input writes go through `apply_writes` directly (above) — they
+            # never enter `checkpoint_pending_writes`, so the after_tick
+            # capture site does not see them. In exit mode, capture them
+            # here so `_exit_delta_writes` includes the input's delta writes
+            # alongside per-superstep writes; otherwise the input would be
+            # lost on read (it's not in final_checkpoint.channel_values for
+            # sub-freq channels, and walks ignore target.pending_writes).
+            if self._exit_delta_writes is not None:
+                for c, v in input_writes:
+                    if isinstance(self.specs.get(c), DeltaChannel):
+                        self._exit_delta_writes.append((self.step, NULL_TASK_ID, c, v))
+            # Persist delta-channel input writes so sub-freq inputs are
+            # recoverable via ancestor walk (mirrors the Command input path).
+            if self.durability != "exit":
+                delta_input = [
+                    (c, v)
+                    for c, v in input_writes
+                    if isinstance(self.specs.get(c), DeltaChannel)
+                ]
+                if delta_input:
+                    self.put_writes(NULL_TASK_ID, delta_input)
             # save input checkpoint
             self.updated_channels = updated_channels
             self._put_checkpoint({"source": "input"})
@@ -905,36 +969,66 @@ class PregelLoop:
         return updated_channels
 
     def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
-        # assign step and parents
+        # `is` (object identity) — not `==`. Three of four call sites pass a
+        # fresh dict ({"source":"input"|"loop"|"fork"}); only
+        # `_suppress_interrupt`(will rename to _on_loop_exit soon)
+        # at exit reuses the existing `self.checkpoint_metadata` instance. So
+        # `metadata is self.checkpoint_metadata` is True only on the exit call,
+        # which is what we use to gate exit-only behaviour (skip count-bump,
+        # don't replace metadata). Could be replaced by an explicit
+        # `exiting: bool = False` parameter; left as-is to match the existing
+        # idiom in this file.
+        # TODO: replace with an explicit `exiting: bool = False` parameter.
         exiting = metadata is self.checkpoint_metadata
         if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
             # checkpoint already saved
             return
-        # Carry per-delta-channel update bookkeeping forward across
-        # supersteps. Capture from the OLD metadata before potentially
-        # replacing it with a fresh dict that wouldn't contain it. Then
-        # increment for any delta channel updated this step (so the count
-        # reflects "supersteps that wrote to this channel since last
-        # snapshot"). create_checkpoint will reset entries to 0 for any
-        # channel that fires a snapshot this step.
-        prev_counts = dict(
-            self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
-        )
-        new_counts = dict(prev_counts)
-        if self.updated_channels:
-            for ch_name in self.updated_channels:
-                ch_obj = self.channels.get(ch_name)
-                if isinstance(ch_obj, DeltaChannel):
-                    new_counts[ch_name] = new_counts.get(ch_name, 0) + 1
+        # 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 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.
         if not exiting:
+            prev_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
+            )
+            new_counters: dict[str, tuple[int, int]] = {}
+            updated = self.updated_channels or set()
+            for ch_name, ch in self.channels.items():
+                if not isinstance(ch, DeltaChannel):
+                    continue
+                u, s = prev_counters.get(ch_name, (0, 0))
+                s += 1
+                if ch_name in updated:
+                    u += 1
+                new_counters[ch_name] = (u, s)
             metadata["step"] = self.step
             metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
             self.checkpoint_metadata = metadata
+        else:
+            new_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
+            )
         # do checkpoint?
         do_checkpoint = self._checkpointer_put_after_previous is not None and (
             exiting or self.durability != "exit"
         )
         # create new checkpoint
+        channels_to_snapshot = (
+            delta_channels_to_snapshot(self.channels, new_counters)
+            if do_checkpoint
+            else set()
+        )
         self.checkpoint = create_checkpoint(
             self.checkpoint,
             self.channels if do_checkpoint else None,
@@ -944,14 +1038,15 @@ class PregelLoop:
             get_next_version=self.checkpointer_get_next_version
             if do_checkpoint
             else None,
-            force_delta_snapshot=exiting and self.durability == "exit",
-            updates_since_snapshot=new_counts,
-            new_updates_since_snapshot=new_counts,
+            channels_to_snapshot=channels_to_snapshot,
         )
-        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"]
+        for k in channels_to_snapshot:
+            new_counters[k] = (0, 0)
+        non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
+        if non_zero:
+            self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
+        elif "counters_since_delta_snapshot" in self.checkpoint_metadata:
+            del self.checkpoint_metadata["counters_since_delta_snapshot"]
         # 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()
@@ -1010,6 +1105,99 @@ class PregelLoop:
             # increment step
             self.step += 1
 
+    def _put_exit_delta_writes(self) -> None:
+        """Stage stub + accumulated delta writes so final_checkpoint's put
+        waits on them (visibility invariant: both must be durable before
+        final_checkpoint becomes visible to readers).
+
+        Stub is created lazily — only when no persisted parent exists AND at
+        least one delta channel has writes that won't be snapshotted.
+        """
+        if (
+            not self._exit_delta_writes
+            or self.checkpointer is None
+            or self._checkpointer_put_after_previous is None
+            or self.checkpointer_put_writes is None
+        ):
+            return
+
+        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)
+            for (step, tid, ch, v) in self._exit_delta_writes
+            if ch not in channels_to_snapshot
+        ]
+        if not pending:
+            return
+
+        if self._has_persisted_parent:
+            # _initial_checkpoint_config's checkpoint_id is the saved parent's
+            # id (saver returned a real tuple at __enter__).
+            anchor_config = self._initial_checkpoint_config
+        else:
+            stub_cp = empty_checkpoint()
+            stub_cp["id"] = self.checkpoint_id_saved
+            stub_cp["ts"] = datetime.now(timezone.utc).isoformat()
+            # Stub has no parent (checkpoint_id=None in config).
+            stub_put_config = patch_configurable(
+                self._initial_checkpoint_config,
+                {CONFIG_KEY_CHECKPOINT_ID: None},
+            )
+            # Anchor config for put_writes: checkpoint_id = stub's id.
+            anchor_config = patch_configurable(
+                self._initial_checkpoint_config,
+                {CONFIG_KEY_CHECKPOINT_ID: stub_cp["id"]},
+            )
+            self._put_checkpoint_fut = self.submit(
+                self._checkpointer_put_after_previous,
+                getattr(self, "_put_checkpoint_fut", None),
+                stub_put_config,
+                stub_cp,
+                {"step": -2},
+                {},
+            )
+            # Set checkpoint_config so final_checkpoint's _put_checkpoint
+            # sees the stub as its parent.
+            self.checkpoint_config = anchor_config
+
+        # Step-prefixed synthetic task_id preserves chronological superstep
+        # order under the saver's ORDER BY task_id, idx sorting.
+        grouped: dict[tuple[int, str], list[tuple[str, Any]]] = {}
+        for step, tid, ch, v in pending:
+            grouped.setdefault((step, tid), []).append((ch, v))
+        anchor_write_config = patch_configurable(
+            anchor_config,
+            {
+                CONFIG_KEY_CHECKPOINT_NS: self.config[CONF].get(
+                    CONFIG_KEY_CHECKPOINT_NS, ""
+                ),
+                CONFIG_KEY_CHECKPOINT_ID: anchor_config[CONF][CONFIG_KEY_CHECKPOINT_ID],
+            },
+        )
+        for (step, tid), entries in grouped.items():
+            synth_tid = f"{step:08d}-{tid}"
+            if self.checkpointer_put_writes_accepts_task_path:
+                fut = self.submit(
+                    self.checkpointer_put_writes,
+                    anchor_write_config,
+                    entries,
+                    synth_tid,
+                    "",
+                )
+            else:
+                fut = self.submit(
+                    self.checkpointer_put_writes,
+                    anchor_write_config,
+                    entries,
+                    synth_tid,
+                )
+            if self._delta_write_futs is not None:
+                self._delta_write_futs.append(fut)
+
     def _suppress_interrupt(
         self,
         exc_type: type[BaseException] | None,
@@ -1025,6 +1213,7 @@ class PregelLoop:
             # or a nested graph with checkpointer=True
             or all(NS_END not in part for part in self.checkpoint_ns)
         ):
+            self._put_exit_delta_writes()
             self._put_checkpoint(self.checkpoint_metadata)
             self._put_pending_writes()
         # suppress interrupt
@@ -1184,6 +1373,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        error_handler: Runnable[Any, Any] | None = None,
         has_graph_lifecycle_callbacks: bool = False,
     ) -> None:
         super().__init__(
@@ -1205,6 +1395,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             trigger_to_nodes=trigger_to_nodes,
             retry_policy=retry_policy,
             cache_policy=cache_policy,
+            error_handler=error_handler,
             durability=durability,
             has_graph_lifecycle_callbacks=has_graph_lifecycle_callbacks,
         )
@@ -1230,6 +1421,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         metadata: CheckpointMetadata,
         new_versions: ChannelVersions,
     ) -> RunnableConfig:
+        if self._delta_write_futs:
+            futs, self._delta_write_futs = self._delta_write_futs, []
+            concurrent.futures.wait(futs)
         try:
             if prev is not None:
                 prev.result()
@@ -1264,22 +1458,18 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     def schedule_error_handler(
         self, failed_task: PregelExecutableTask, error: BaseException
     ) -> PregelExecutableTask | None:
-        handler_node = self.nodes[failed_task.name].error_handler_node
-        if not handler_node:
+        handler = failed_task.error_handler or self.error_handler
+        if handler is None:
             return None
         writes = list(failed_task.writes)
         writes.append((ERROR_SOURCE_NODE, failed_task.name))
-        self.put_writes(
-            failed_task.id,
-            writes,
-        )
+        self.put_writes(failed_task.id, writes)
         handler_task = prepare_node_error_handler_task(
             failed_task,
-            handler_node_name=handler_node,
+            handler=handler,
             failed_error=error,
             checkpoint=self.checkpoint,
             pending_writes=self.checkpoint_pending_writes,
-            processes=self.nodes,
             channels=self.channels,
             managed=self.managed,
             config=failed_task.config,
@@ -1289,10 +1479,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             checkpointer=self.checkpointer,
             manager=self.manager,
             retry_policy=self.retry_policy,
-            cache_policy=self.cache_policy,
         )
-        if handler_task is None:
-            return None
         self.tasks[handler_task.id] = handler_task
         if not self.is_replaying:
             self._match_writes({handler_task.id: handler_task})
@@ -1300,6 +1487,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             self.output_writes(task.id, task.writes, cached=True)
         return handler_task
 
+
+
     def put_writes(self, task_id: str, writes: WritesT) -> None:
         """Put writes for a task, to be read by the next tick."""
         super().put_writes(task_id, writes)
@@ -1347,6 +1536,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # graph/thread. Returns None on first invocation.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
 
+        # Capture before the synthetic-empty fallback below overwrites `saved`.
+        # `_put_exit_delta_writes` uses this on first run (no persisted parent)
+        # to lazy-create a stub instead of anchoring delta writes on a parent.
+        self._has_persisted_parent = saved is not None
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1362,6 +1555,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
                 **saved.config.get(CONF, {}),
             },
         }
+        self._initial_checkpoint_config = self.checkpoint_config
         self.prev_checkpoint_config = saved.parent_config
         self.checkpoint_id_saved = saved.checkpoint["id"]
         self.checkpoint = saved.checkpoint
@@ -1371,6 +1565,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
+        self._delta_write_futs = []
+        self._exit_delta_writes = (
+            [] if self.durability == "exit" and self.checkpointer is not None else None
+        )
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs,
@@ -1425,6 +1623,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        error_handler: Runnable[Any, Any] | None = None,
         has_graph_lifecycle_callbacks: bool = False,
     ) -> None:
         super().__init__(
@@ -1446,6 +1645,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             trigger_to_nodes=trigger_to_nodes,
             retry_policy=retry_policy,
             cache_policy=cache_policy,
+            error_handler=error_handler,
             durability=durability,
             has_graph_lifecycle_callbacks=has_graph_lifecycle_callbacks,
         )
@@ -1510,22 +1710,18 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     async def aschedule_error_handler(
         self, failed_task: PregelExecutableTask, error: BaseException
     ) -> PregelExecutableTask | None:
-        handler_node = self.nodes[failed_task.name].error_handler_node
-        if not handler_node:
+        handler = failed_task.error_handler or self.error_handler
+        if handler is None:
             return None
         writes = list(failed_task.writes)
         writes.append((ERROR_SOURCE_NODE, failed_task.name))
-        self.put_writes(
-            failed_task.id,
-            writes,
-        )
+        self.put_writes(failed_task.id, writes)
         handler_task = prepare_node_error_handler_task(
             failed_task,
-            handler_node_name=handler_node,
+            handler=handler,
             failed_error=error,
             checkpoint=self.checkpoint,
             pending_writes=self.checkpoint_pending_writes,
-            processes=self.nodes,
             channels=self.channels,
             managed=self.managed,
             config=failed_task.config,
@@ -1535,10 +1731,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             checkpointer=self.checkpointer,
             manager=self.manager,
             retry_policy=self.retry_policy,
-            cache_policy=self.cache_policy,
         )
-        if handler_task is None:
-            return None
         self.tasks[handler_task.id] = handler_task
         if not self.is_replaying:
             self._match_writes({handler_task.id: handler_task})
@@ -1596,6 +1789,10 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # graph/thread. Returns None on first invocation.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
 
+        # Capture before the synthetic-empty fallback below overwrites `saved`.
+        # `_put_exit_delta_writes` uses this on first run (no persisted parent)
+        # to lazy-create a stub instead of anchoring delta writes on a parent.
+        self._has_persisted_parent = saved is not None
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1611,6 +1808,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
                 **saved.config.get(CONF, {}),
             },
         }
+        self._initial_checkpoint_config = self.checkpoint_config
         self.prev_checkpoint_config = saved.parent_config
         self.checkpoint_id_saved = saved.checkpoint["id"]
         self.checkpoint = saved.checkpoint
@@ -1621,6 +1819,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             else []
         )
         self._delta_write_futs = []
+        self._exit_delta_writes = (
+            [] if self.durability == "exit" and self.checkpointer is not None else None
+        )
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )

libs/langgraph/langgraph/pregel/_read.py

@@ -138,11 +138,8 @@ class PregelNode:
     metadata: Mapping[str, Any] | None
     """Metadata to attach to the node for tracing."""
 
-    is_error_handler: bool
-    """Whether this node is registered as an error handler node."""
-
-    error_handler_node: str | None
-    """Optional handler node name for failures from this node."""
+    error_handler: Runnable[Any, Any] | None
+    """Callable invoked after retries are exhausted; receives same input as the node."""
 
     subgraphs: Sequence[PregelProtocol]
     """Subgraphs used by the node."""
@@ -159,8 +156,7 @@ class PregelNode:
         bound: Runnable[Any, Any] | None = None,
         retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
         cache_policy: CachePolicy | None = None,
-        is_error_handler: bool = False,
-        error_handler_node: str | None = None,
+        error_handler: Runnable[Any, Any] | None = None,
         subgraphs: Sequence[PregelProtocol] | None = None,
         timeout: float | timedelta | TimeoutPolicy | None = None,
     ) -> None:
@@ -177,8 +173,7 @@ class PregelNode:
         self.timeout = coerce_timeout_policy(timeout)
         self.tags = tags
         self.metadata = metadata
-        self.is_error_handler = is_error_handler
-        self.error_handler_node = error_handler_node
+        self.error_handler = error_handler
         if subgraphs is not None:
             self.subgraphs = subgraphs
         elif self.bound is not DEFAULT_BOUND:

libs/langgraph/langgraph/pregel/_runner.py

@@ -13,7 +13,6 @@ from collections.abc import (
     Collection,
     Iterable,
     Iterator,
-    Mapping,
     Sequence,
 )
 from functools import partial
@@ -143,7 +142,6 @@ class PregelRunner:
         put_writes: weakref.ref[Callable[[str, Sequence[tuple[str, Any]]], None]],
         use_astream: bool = False,
         node_finished: Callable[[str], None] | None = None,
-        node_error_handler_map: Mapping[str, str] | None = None,
         schedule_error_handler: Callable[
             [PregelExecutableTask, BaseException], PregelExecutableTask | None
         ]
@@ -158,20 +156,10 @@ class PregelRunner:
         self.put_writes = put_writes
         self.use_astream = use_astream
         self.node_finished = node_finished
-        self.node_error_handler_map = dict(node_error_handler_map or {})
-        self.error_handler_nodes = set(self.node_error_handler_map.values())
         self.schedule_error_handler = schedule_error_handler
         self.aschedule_error_handler = aschedule_error_handler
-        # Exception object ids that are already routed to graph-level error handler.
-        # These ids are consulted by stop/panic checks to avoid re-raising handled
-        # exceptions via the normal fatal path in the same run.
         self._handled_exception_ids: set[int] = set()
 
-    def _should_route_to_error_handler(self, task: PregelExecutableTask) -> bool:
-        if task.name in self.error_handler_nodes:
-            return False
-        return task.name in self.node_error_handler_map
-
     def tick(
         self,
         tasks: Iterable[PregelExecutableTask],
@@ -222,7 +210,7 @@ class PregelRunner:
                 self.commit(t, exc)
                 if (
                     not isinstance(exc, GraphBubbleUp)
-                    and self._should_route_to_error_handler(t)
+                    and t.error_handler is not None
                     and self.schedule_error_handler is not None
                 ):
                     self._handled_exception_ids.add(id(exc))
@@ -295,7 +283,7 @@ class PregelRunner:
                         futures[get_waiter()] = None
                 elif (
                     (task_exc := _exception(fut))
-                    and self._should_route_to_error_handler(task)
+                    and task.error_handler is not None
                     and not isinstance(task_exc, GraphBubbleUp)
                 ):
                     self._handled_exception_ids.add(id(task_exc))
@@ -414,7 +402,7 @@ class PregelRunner:
                 self.commit(t, exc)
                 if (
                     not isinstance(exc, GraphBubbleUp)
-                    and self._should_route_to_error_handler(t)
+                    and t.error_handler is not None
                     and self.aschedule_error_handler is not None
                 ):
                     self._handled_exception_ids.add(id(exc))
@@ -494,7 +482,7 @@ class PregelRunner:
                         futures[get_waiter()] = None
                 elif (
                     (task_exc := _exception(fut))
-                    and self._should_route_to_error_handler(task)
+                    and task.error_handler is not None
                     and not isinstance(task_exc, GraphBubbleUp)
                 ):
                     self._handled_exception_ids.add(id(task_exc))
@@ -594,7 +582,7 @@ class PregelRunner:
             else:
                 # save error to checkpointer
                 task.writes.append((ERROR, exception))
-                if self._should_route_to_error_handler(task) and not isinstance(
+                if task.error_handler is not None and not isinstance(
                     exception, GraphBubbleUp
                 ):
                     # Mark early in commit path; loop-side routing may happen later.

libs/langgraph/langgraph/pregel/_validate.py

@@ -99,14 +99,21 @@ def validate_graph(
 
     if interrupt_after_nodes != "*":
         for n in interrupt_after_nodes:
-            if n not in nodes:
+            if n not in nodes and not _is_valid_error_handler_interrupt(n, nodes):
                 raise ValueError(f"Node {n} not in nodes")
     if interrupt_before_nodes != "*":
         for n in interrupt_before_nodes:
-            if n not in nodes:
+            if n not in nodes and not _is_valid_error_handler_interrupt(n, nodes):
                 raise ValueError(f"Node {n} not in nodes")
 
 
+def _is_valid_error_handler_interrupt(name: str, nodes: Mapping[str, PregelNode]) -> bool:
+    if not name.startswith("__error_handler__"):
+        return False
+    base = name[len("__error_handler__"):]
+    return base in nodes and nodes[base].error_handler is not None
+
+
 def validate_keys(
     keys: str | Sequence[str] | None,
     channels: Mapping[str, Any],

libs/langgraph/langgraph/pregel/main.py

@@ -33,6 +33,7 @@ from uuid import UUID, uuid5
 from langchain_core._api import beta
 from langchain_core.globals import get_debug
 from langchain_core.runnables import (
+    Runnable,
     RunnableSequence,
 )
 from langchain_core.runnables.base import Input, Output
@@ -751,7 +752,7 @@ class Pregel(
     name: str = "LangGraph"
 
     trigger_to_nodes: Mapping[str, Sequence[str]]
-    node_error_handler_map: Mapping[str, str]
+    error_handler: Runnable[Any, Any] | None
 
     def __init__(
         self,
@@ -776,7 +777,7 @@ class Pregel(
         context_schema: type[ContextT] | None = None,
         config: RunnableConfig | None = None,
         trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
-        node_error_handler_map: Mapping[str, str] | None = None,
+        error_handler: Runnable[Any, Any] | None = None,
         name: str = "LangGraph",
         stream_transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
@@ -824,7 +825,7 @@ class Pregel(
         self.context_schema = context_schema
         self.config = config
         self.trigger_to_nodes = trigger_to_nodes or {}
-        self.node_error_handler_map = node_error_handler_map or {}
+        self.error_handler = error_handler
         self.name = name
         self.stream_transformers: tuple[Callable[[tuple[str, ...]], Any], ...] = tuple(
             stream_transformers or ()
@@ -2885,6 +2886,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                error_handler=self.error_handler,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
                 emit_graph_lifecycle_events(loop)
@@ -2895,7 +2897,6 @@ class Pregel(
                     ),
                     put_writes=weakref.WeakMethod(loop.put_writes),
                     node_finished=config[CONF].get(CONFIG_KEY_NODE_FINISHED),
-                    node_error_handler_map=self.node_error_handler_map,
                     schedule_error_handler=loop.schedule_error_handler,
                 )
                 # enable subgraph streaming
@@ -3337,6 +3338,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                error_handler=self.error_handler,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
                 await aemit_graph_lifecycle_events(loop)
@@ -3348,7 +3350,6 @@ class Pregel(
                     put_writes=weakref.WeakMethod(loop.put_writes),
                     use_astream=do_stream,
                     node_finished=config[CONF].get(CONFIG_KEY_NODE_FINISHED),
-                    node_error_handler_map=self.node_error_handler_map,
                     aschedule_error_handler=loop.aschedule_error_handler,
                 )
                 # enable subgraph streaming

libs/langgraph/langgraph/types.py

@@ -628,6 +628,7 @@ class PregelExecutableTask:
     writers: Sequence[Runnable] = ()
     subgraphs: Sequence[PregelProtocol] = ()
     timeout: TimeoutPolicy | None = None
+    error_handler: Runnable | None = None
 
 
 class StateSnapshot(NamedTuple):

libs/langgraph/pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
     'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.4.0a2,<2",
+    "langchain-core>=1.4.0,<2",
     "langgraph-checkpoint>=4.1.0a4,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
     "langgraph-prebuilt>=1.1.0a2,<1.2.0",

libs/langgraph/tests/test_delta_channel_exit_mode.py

@@ -0,0 +1,372 @@
+"""Tests for exit-mode delta channel persistence redesign.
+
+Validates that `durability="exit"` correctly persists delta-channel writes
+using count-based snapshot decisions (rather than force-snapshotting every
+channel), lazy stub creation when no parent exists, and proper read-path
+reconstruction via ancestor walks.
+"""
+
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+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 START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer
+
+pytestmark = pytest.mark.anyio
+
+
+def _build_graph(
+    checkpointer: InMemorySaver,
+    *,
+    freq: int = 1000,
+) -> Any:
+    channel = DeltaChannel(_messages_delta_reducer, snapshot_frequency=freq)
+    # Functional TypedDict form: class form can't reference `channel` (a
+    # local variable) inside Annotated due to forward-ref evaluation rules.
+    State = TypedDict("State", {"messages": Annotated[list, channel]})  # type: ignore[call-overload]  # noqa: UP013
+
+    def respond(state: dict) -> dict:
+        i = len(state["messages"])
+        return {"messages": [AIMessage(content=f"reply-{i}", id=f"ai{i}")]}
+
+    builder = StateGraph(State)
+    builder.add_node("respond", respond)
+    builder.add_edge(START, "respond")
+    return builder.compile(checkpointer=checkpointer)
+
+
+# ---------------------------------------------------------------------------
+# 8a. Write-path / structural tests
+# ---------------------------------------------------------------------------
+
+
+async def test_exit_first_run_no_delta_writes() -> None:
+    """Graph with delta channel invoked with input that doesn't touch it.
+    Only one checkpoint row, no stub."""
+    State = TypedDict(  # noqa: UP013
+        "State",
+        {
+            "messages": Annotated[list, DeltaChannel(_messages_delta_reducer)],
+            "value": str,
+        },
+    )  # type: ignore[call-overload]
+
+    def noop(state: dict) -> dict:
+        return {"value": "done"}
+
+    saver = InMemorySaver()
+    builder = StateGraph(State)
+    builder.add_node("noop", noop)
+    builder.add_edge(START, "noop")
+    graph = builder.compile(checkpointer=saver)
+    config = {"configurable": {"thread_id": "no-delta-writes"}}
+
+    graph.invoke({"value": "start"}, config, durability="exit")
+
+    checkpoints = list(saver.list(config))
+    assert len(checkpoints) == 1
+    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
+    assert len(stubs) == 0
+
+
+async def test_exit_first_run_all_snapshot() -> None:
+    """snapshot_frequency=1 forces every channel to snapshot.
+    No stub needed; final_checkpoint has _DeltaSnapshot."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver, freq=1)
+    config = {"configurable": {"thread_id": "all-snapshot"}}
+
+    result = graph.invoke(
+        {"messages": [HumanMessage(content="hi", id="h1")]},
+        config,
+        durability="exit",
+    )
+    assert len(result["messages"]) == 2
+
+    checkpoints = list(saver.list(config))
+    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
+    assert len(stubs) == 0
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+
+    state = graph.get_state(config)
+    assert [m.content for m in state.values["messages"]] == ["hi", "reply-1"]
+
+
+async def test_exit_first_run_sub_freq_with_writes() -> None:
+    """First run with default snapshot_frequency (1000), writes below threshold.
+    A stub is created; writes are anchored under it; get_state reconstructs."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver)
+    config = {"configurable": {"thread_id": "sub-freq-first"}}
+
+    result = graph.invoke(
+        {"messages": [HumanMessage(content="hello", id="h1")]},
+        config,
+        durability="exit",
+    )
+    assert [m.content for m in result["messages"]] == ["hello", "reply-1"]
+
+    checkpoints = list(saver.list(config))
+    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
+    assert len(stubs) == 1, f"Expected 1 stub, got {len(stubs)}"
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert "messages" not in head.checkpoint["channel_values"]
+    assert "messages" in head.checkpoint["channel_versions"]
+
+    state = graph.get_state(config)
+    assert [m.content for m in state.values["messages"]] == ["hello", "reply-1"]
+
+
+async def test_exit_resumed_run_sub_freq() -> None:
+    """Two consecutive exit runs. Second run anchors on the first's
+    final_checkpoint (no new stub). Ordering preserved."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver)
+    config = {"configurable": {"thread_id": "resumed-sub-freq"}}
+
+    graph.invoke(
+        {"messages": [HumanMessage(content="msg1", id="h1")]},
+        config,
+        durability="exit",
+    )
+
+    graph.invoke(
+        {"messages": [HumanMessage(content="msg2", id="h2")]},
+        config,
+        durability="exit",
+    )
+
+    checkpoints = list(saver.list(config))
+    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
+    assert len(stubs) == 1
+
+    state = graph.get_state(config)
+    contents = [m.content for m in state.values["messages"]]
+    assert len(contents) == 4
+    assert contents[0] == "msg1"
+    assert contents[2] == "msg2"
+    assert contents[0:4:2] == ["msg1", "msg2"]
+
+
+async def test_exit_count_parity_sync_vs_exit() -> None:
+    """Sync and exit durability produce the same update count in
+    counters_since_delta_snapshot after an equivalent run."""
+    for durability in ("sync", "exit"):
+        saver = InMemorySaver()
+        graph = _build_graph(saver)
+        config = {"configurable": {"thread_id": f"parity-{durability}"}}
+
+        graph.invoke(
+            {"messages": [HumanMessage(content="hi", id="h1")]},
+            config,
+            durability=durability,
+        )
+
+        head = saver.get_tuple(config)
+        assert head is not None
+        counters = head.metadata.get("counters_since_delta_snapshot", {})
+        updates, supersteps = counters.get("messages", (0, 0))
+        assert updates == 2, (
+            f"durability={durability}: expected updates=2, got {updates}"
+        )
+        assert supersteps >= 2, (
+            f"durability={durability}: expected supersteps>=2, got {supersteps}"
+        )
+
+
+async def test_exit_snapshot_fires_at_frequency() -> None:
+    """With snapshot_frequency=3, after 3 exit runs (each incrementing count
+    by 2: input + superstep), the 2nd run hits count=4>=3, triggering snapshot.
+    After that run, count resets to 0 and channel_values has _DeltaSnapshot."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver, freq=3)
+    config = {"configurable": {"thread_id": "snapshot-at-freq"}}
+
+    graph.invoke(
+        {"messages": [HumanMessage(content="m1", id="h1")]},
+        config,
+        durability="exit",
+    )
+    head = saver.get_tuple(config)
+    assert head is not None
+    counters1 = head.metadata.get("counters_since_delta_snapshot", {})
+    updates1 = counters1.get("messages", (0, 0))[0]
+    assert updates1 == 2
+
+    graph.invoke(
+        {"messages": [HumanMessage(content="m2", id="h2")]},
+        config,
+        durability="exit",
+    )
+    head = saver.get_tuple(config)
+    assert head is not None
+    counters2 = head.metadata.get("counters_since_delta_snapshot", {})
+    updates2 = counters2.get("messages", (0, 0))[0]
+    assert updates2 == 0, f"Expected reset to 0 after snapshot, got {updates2}"
+    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+
+
+async def test_exit_mixed_snapshot_and_non_snapshot() -> None:
+    """One delta channel at freq=1 (always snapshot) and one at freq=1000
+    (never snapshot within this test). Verify correct behavior for both."""
+
+    fast_ch = DeltaChannel(_messages_delta_reducer, snapshot_frequency=1)
+    slow_ch = DeltaChannel(_messages_delta_reducer, snapshot_frequency=1000)
+    State = TypedDict(  # noqa: UP013
+        "State",
+        {"fast": Annotated[list, fast_ch], "slow": Annotated[list, slow_ch]},
+    )  # type: ignore[call-overload]
+
+    def respond(state: dict) -> dict:
+        return {
+            "fast": [AIMessage(content="fast-reply", id="f1")],
+            "slow": [AIMessage(content="slow-reply", id="s1")],
+        }
+
+    saver = InMemorySaver()
+    builder = StateGraph(State)
+    builder.add_node("respond", respond)
+    builder.add_edge(START, "respond")
+    graph = builder.compile(checkpointer=saver)
+    config = {"configurable": {"thread_id": "mixed-freq"}}
+
+    graph.invoke(
+        {
+            "fast": [HumanMessage(content="fast-in", id="fi")],
+            "slow": [HumanMessage(content="slow-in", id="si")],
+        },
+        config,
+        durability="exit",
+    )
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("fast"), _DeltaSnapshot)
+    assert "slow" not in head.checkpoint["channel_values"]
+
+    state = graph.get_state(config)
+    assert [m.content for m in state.values["fast"]] == ["fast-in", "fast-reply"]
+    assert [m.content for m in state.values["slow"]] == ["slow-in", "slow-reply"]
+
+
+# ---------------------------------------------------------------------------
+# 8b. Read-path tests
+# ---------------------------------------------------------------------------
+
+
+async def test_exit_multi_run_replay_chain() -> None:
+    """K=4 consecutive exit runs, each adding a message. After each run,
+    get_state returns all messages in chronological order."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver)
+    config = {"configurable": {"thread_id": "replay-chain"}}
+
+    for i in range(4):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"user-{i}", id=f"h{i}")]},
+            config,
+            durability="exit",
+        )
+
+        state = graph.get_state(config)
+        contents = [m.content for m in state.values["messages"]]
+        user_msgs = [c for c in contents if c.startswith("user-")]
+        assert user_msgs == [f"user-{j}" for j in range(i + 1)], (
+            f"After run {i}: user messages out of order: {user_msgs}"
+        )
+        assert len(contents) == (i + 1) * 2
+
+
+async def test_exit_metadata_round_trip() -> None:
+    """K=5 consecutive exit runs with snapshot_frequency=5. Verify metadata
+    counters_since_delta_snapshot increments correctly across runs."""
+    freq = 5
+    saver = InMemorySaver()
+    graph = _build_graph(saver, freq=freq)
+    config = {"configurable": {"thread_id": "metadata-rt"}}
+
+    for i in range(1, 6):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"m{i}", id=f"h{i}")]},
+            config,
+            durability="exit",
+        )
+        head = saver.get_tuple(config)
+        assert head is not None
+        counters = head.metadata.get("counters_since_delta_snapshot", {})
+        updates = counters.get("messages", (0, 0))[0]
+        cumulative = i * 2
+        if cumulative >= freq:
+            assert updates == 0 or updates == cumulative % freq or updates < freq, (
+                f"After run {i}: updates={updates} should have reset or be partial"
+            )
+        else:
+            assert updates == cumulative, (
+                f"After run {i}: expected {cumulative}, got {updates}"
+            )
+
+
+async def test_exit_mixed_durability_round_trip() -> None:
+    """Alternate sync and exit durability; verify counts stay monotonic
+    and state accumulates correctly."""
+    saver = InMemorySaver()
+    graph = _build_graph(saver)
+    config = {"configurable": {"thread_id": "mixed-durability"}}
+
+    for i, dur in enumerate(["sync", "exit", "sync", "exit"]):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"msg-{i}", id=f"h{i}")]},
+            config,
+            durability=dur,
+        )
+
+        state = graph.get_state(config)
+        contents = [m.content for m in state.values["messages"]]
+        user_msgs = [c for c in contents if c.startswith("msg-")]
+        assert user_msgs == [f"msg-{j}" for j in range(i + 1)], (
+            f"After run {i} (durability={dur}): {user_msgs}"
+        )
+        assert len(contents) == (i + 1) * 2
+
+
+async def test_exit_snapshot_then_tail_deltas() -> None:
+    """Run 1 forces snapshot (freq=1). Run 2 at freq=1000 adds more writes
+    that don't snapshot. Reading after run 2 must combine the snapshot seed
+    with the tail deltas."""
+    saver = InMemorySaver()
+
+    graph1 = _build_graph(saver, freq=1)
+    config = {"configurable": {"thread_id": "snapshot-then-tail"}}
+    graph1.invoke(
+        {"messages": [HumanMessage(content="seed-msg", id="h1")]},
+        config,
+        durability="exit",
+    )
+
+    head = saver.get_tuple(config)
+    assert head is not None
+    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+
+    graph2 = _build_graph(saver, freq=1000)
+    graph2.invoke(
+        {"messages": [HumanMessage(content="tail-msg", id="h2")]},
+        config,
+        durability="exit",
+    )
+
+    state = graph2.get_state(config)
+    contents = [m.content for m in state.values["messages"]]
+    assert "seed-msg" in contents
+    assert "tail-msg" in contents
+    assert contents.index("seed-msg") < contents.index("tail-msg")

libs/langgraph/tests/test_delta_channel_supersteps_bound.py

@@ -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"]

libs/langgraph/tests/test_retry.py

@@ -1674,15 +1674,28 @@ async def test_arun_with_retry_timeout_observer_tracks_attempts():
 async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     events: list = []
 
+    # `_TimedAttemptScope.__init__` sets `_last_progress` to `time.monotonic()`,
+    # but the watchdog itself doesn't start running until after `wrap_config`
+    # and task scheduling — under CI load that gap can be large enough to eat
+    # the entire idle window before the task body's first await even runs. We
+    # defend against that by:
+    #   1. Using a generous idle_timeout so scheduling slack stays well within it.
+    #   2. Calling `runtime.heartbeat()` BEFORE the first sleep, which resets
+    #      `_last_progress` to "now" the moment the task body actually starts.
+    idle_timeout_s = 1.0
+
     class HeartbeatProc:
         async def ainvoke(self, input, config):
             runtime = config[CONF][CONFIG_KEY_RUNTIME]
+            runtime.heartbeat()  # reset the idle clock at task-body entry
             for _ in range(8):
                 await asyncio.sleep(0.05)
                 runtime.heartbeat()
             return "ok"
 
-    task = _make_task(HeartbeatProc(), timeout=_idle_timeout(0.2), name="heartbeat")
+    task = _make_task(
+        HeartbeatProc(), timeout=_idle_timeout(idle_timeout_s), name="heartbeat"
+    )
     task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
     assert await arun_with_retry(task, retry_policy=None) == "ok"
 
@@ -1691,13 +1704,13 @@ async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     assert by_event[-1] == "finish"
     progress = [ev for ev in events if ev.event == "progress"]
     assert progress, "expected at least one progress event from heartbeat"
-    # Rate limit is `idle_timeout / 4` = 0.05s; with 8 heartbeats spaced ~0.05s
-    # we should see at most ~one progress event per heartbeat (well below 8).
+    # Rate limit is `idle_timeout / 4` = 0.25s; with the task running for
+    # ~400ms we expect 1–2 progress events (well below the 9 heartbeats).
     assert len(progress) <= len(by_event)
     for ev in progress:
         assert ev.context.task_name == "heartbeat"
         assert ev.context.attempt == 1
-        assert ev.context.idle_timeout_secs == 0.2
+        assert ev.context.idle_timeout_secs == idle_timeout_s
         assert isinstance(ev.progress_at, datetime)
 
 
@@ -2267,3 +2280,139 @@ def test_node_without_error_handler_still_fails_run():
 
     with pytest.raises(ValueError, match="no handler"):
         graph.invoke({"foo": ""})
+
+
+# ---------------------------------------------------------------------------
+# Structural invariants from the policy-style refactor
+# ---------------------------------------------------------------------------
+
+
+def test_error_handler_not_registered_as_node():
+    """After compile, no hidden __error_handler__* nodes should exist in the graph."""
+
+    class State(TypedDict):
+        foo: str
+
+    def failing_node(state: State) -> State:
+        raise ValueError("boom")
+
+    def handler(state: State, error: NodeError) -> State:
+        return {"foo": "handled"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("failing_node", failing_node, error_handler=handler)
+        .add_edge(START, "failing_node")
+        .compile()
+    )
+
+    hidden = [k for k in graph.nodes if k.startswith("__error_handler__")]
+    assert hidden == [], f"unexpected hidden nodes: {hidden}"
+
+
+def test_error_handler_stored_on_pregel_node():
+    """The error_handler callable should be a Runnable field on PregelNode, not a name pointer."""
+
+    class State(TypedDict):
+        foo: str
+
+    def failing_node(state: State) -> State:
+        raise ValueError("boom")
+
+    def handler(state: State) -> State:
+        return {"foo": "handled"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("failing_node", failing_node, error_handler=handler)
+        .add_edge(START, "failing_node")
+        .compile()
+    )
+
+    pregel_node = graph.nodes["failing_node"]
+    assert pregel_node.error_handler is not None, "error_handler should be set on PregelNode"
+    assert not hasattr(pregel_node, "error_handler_node"), "old string-pointer field should be gone"
+    assert not hasattr(pregel_node, "is_error_handler"), "is_error_handler flag should be gone"
+
+
+def test_error_handler_dispatched_from_task_field():
+    """error_handler on PregelExecutableTask drives dispatch — no node-map lookup needed."""
+
+    class State(TypedDict):
+        foo: str
+
+    def failing_node(state: State) -> State:
+        raise ValueError("boom")
+
+    def handler(state: State) -> State:
+        return {"foo": "handled"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("failing_node", failing_node, error_handler=handler)
+        .add_edge(START, "failing_node")
+        .compile()
+    )
+    result = graph.invoke({"foo": ""})
+    assert result["foo"] == "handled"
+
+
+# ---------------------------------------------------------------------------
+# Graph-level error handler
+# ---------------------------------------------------------------------------
+
+
+def test_graph_level_error_handler_used_when_no_per_node_handler():
+    """compile(error_handler=fallback) should catch failures from nodes without their own handler."""
+
+    class State(TypedDict):
+        foo: str
+
+    def failing_node(state: State) -> State:
+        raise RuntimeError("node failed")
+
+    def graph_handler(state: State, error: NodeError) -> State:
+        return {"foo": f"graph_handler_caught:{error.node}"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("failing_node", failing_node)
+        .add_edge(START, "failing_node")
+        .compile(error_handler=graph_handler)
+    )
+
+    result = graph.invoke({"foo": ""})
+    assert result["foo"] == "graph_handler_caught:failing_node"
+
+
+def test_per_node_handler_takes_precedence_over_graph_level():
+    """When a node has its own error_handler, it should win over the graph-level fallback."""
+
+    class State(TypedDict):
+        foo: str
+
+    def failing_node(state: State) -> State:
+        raise RuntimeError("node failed")
+
+    def node_handler(state: State, error: NodeError) -> State:
+        return {"foo": "node_handler"}
+
+    def graph_handler(state: State, error: NodeError) -> State:
+        return {"foo": "graph_handler"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("failing_node", failing_node, error_handler=node_handler)
+        .add_edge(START, "failing_node")
+        .compile(error_handler=graph_handler)
+    )
+
+    result = graph.invoke({"foo": ""})
+    assert result["foo"] == "node_handler"
+
+
+# ---------------------------------------------------------------------------
+# Functional API
+# ---------------------------------------------------------------------------
+
+

libs/langgraph/uv.lock

@@ -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]]
@@ -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" },
@@ -1849,14 +1849,14 @@ dev = [
     { name = "pytest-watch" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.20.2" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 test = [
     { name = "pytest" },
@@ -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]]

libs/prebuilt/uv.lock

@@ -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]]
@@ -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" },
@@ -618,14 +618,14 @@ dev = [
     { name = "pytest-watch" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.20.2" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 test = [
     { name = "pytest" },
@@ -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]]

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

@@ -110,7 +110,7 @@ def get_client(
     if url is None:
         url = "http://api"
         if os.environ.get("__LANGGRAPH_DEFER_LOOPBACK_TRANSPORT") == "true":
-            transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]
+            transport = get_asgi_transport()(app=None, root_path="/noauth")  # ty: ignore[invalid-argument-type]
             _registered_transports.append(transport)
         else:
             try:
@@ -122,7 +122,7 @@ def get_client(
                     "Failed to connect to in-process LangGraph server. Deferring configuration.",
                     exc_info=True,
                 )
-                transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]
+                transport = get_asgi_transport()(app=None, root_path="/noauth")  # ty: ignore[invalid-argument-type]
                 _registered_transports.append(transport)
 
     if transport is None:
@@ -131,7 +131,7 @@ def get_client(
         base_url=url,
         transport=transport,
         timeout=(
-            httpx.Timeout(timeout)  # type: ignore[arg-type]
+            httpx.Timeout(timeout)  # ty: ignore[invalid-argument-type]
             if timeout is not None
             else httpx.Timeout(connect=5, read=300, write=300, pool=5)
         ),

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

@@ -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
         )

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

@@ -49,7 +49,7 @@ async def _wrap_stream_v2(
     async for part in raw:
         v2 = _sse_to_v2_dict(part.event, part.data)
         if v2 is not None:
-            yield v2
+            yield v2  # ty: ignore[invalid-yield]
 
 
 class RunsClient:

libs/sdk-py/langgraph_sdk/_shared/utilities.py

@@ -144,8 +144,9 @@ def _resolve_timezone(tz: str | tzinfo | ZoneInfo | None) -> str | None:
         return tz
     if isinstance(tz, tzinfo):
         # ZoneInfo objects have a .key attribute with the IANA name
-        if hasattr(tz, "key"):
-            return tz.key  # type: ignore[union-attr]
+        key = getattr(tz, "key", None)
+        if isinstance(key, str):
+            return key
         # Fall back to tzname for fixed-offset timezones like datetime.timezone.utc
         name = tz.tzname(None)
         if name is not None:
@@ -209,7 +210,7 @@ def configure_loopback_transports(app: Any) -> None:
 @functools.lru_cache(maxsize=1)
 def get_asgi_transport() -> type[httpx.ASGITransport]:
     try:
-        from langgraph_api import asgi_transport  # type: ignore[unresolved-import]
+        from langgraph_api import asgi_transport  # ty: ignore[unresolved-import]
 
         return asgi_transport.ASGITransport
     except ImportError:

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

@@ -77,7 +77,7 @@ def get_sync_client(
         base_url=url,
         transport=transport,
         timeout=(
-            httpx.Timeout(timeout)  # type: ignore[arg-type]
+            httpx.Timeout(timeout)  # ty: ignore[invalid-argument-type]
             if timeout is not None
             else httpx.Timeout(connect=5, read=300, write=300, pool=5)
         ),

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

@@ -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
         )

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

@@ -49,7 +49,7 @@ def _wrap_stream_v2_sync(
     for part in raw:
         v2 = _sse_to_v2_dict(part.event, part.data)
         if v2 is not None:
-            yield v2
+            yield v2  # ty: ignore[invalid-yield]
 
 
 class SyncRunsClient:

libs/sdk-py/langgraph_sdk/cache.py

@@ -16,10 +16,10 @@ T = TypeVar("T")
 CacheStatus = Literal["miss", "fresh", "stale", "expired"]
 
 try:
-    from langgraph_api.cache import (  # type: ignore[unresolved-import]
+    from langgraph_api.cache import (  # ty: ignore[unresolved-import]
         cache_get as _cache_get,
     )
-    from langgraph_api.cache import (  # type: ignore[unresolved-import]
+    from langgraph_api.cache import (  # ty: ignore[unresolved-import]
         cache_set as _cache_set,
     )
 except ImportError:
@@ -28,8 +28,8 @@ except ImportError:
 
 
 try:
-    from langgraph_api.cache import SWRResult  # type: ignore[unresolved-import]
-    from langgraph_api.cache import swr as _api_swr  # type: ignore[unresolved-import]
+    from langgraph_api.cache import SWRResult  # ty: ignore[unresolved-import]
+    from langgraph_api.cache import swr as _api_swr  # ty: ignore[unresolved-import]
 
 except ImportError:
     _api_swr = None
@@ -40,7 +40,10 @@ except ImportError:
         value: T
         status: CacheStatus
 
-        async def mutate(self, value: T = ...) -> T:  # type: ignore[assignment]
+        async def mutate(
+            self,
+            value: T = ...,  # ty: ignore[invalid-parameter-default]
+        ) -> T:  # ty: ignore[empty-body]
             """Update or revalidate the cached value."""
             ...
 

libs/sdk-py/langgraph_sdk/errors.py

@@ -37,7 +37,7 @@ class APIError(httpx.HTTPStatusError, LangGraphError):
             req = response_or_request
             response = None
 
-        httpx.HTTPStatusError.__init__(self, message, request=req, response=response)  # type: ignore[arg-type]
+        httpx.HTTPStatusError.__init__(self, message, request=req, response=response)  # ty: ignore[invalid-argument-type]
         LangGraphError.__init__(self, message)
 
         self.request = req

libs/sdk-py/langgraph_sdk/runtime.py

@@ -156,7 +156,7 @@ class _ExecutionRuntime(_ServerRuntimeBase[ContextT], Generic[ContextT]):
         This API is in beta and may change in future releases.
     """
 
-    context: ContextT = field(default=None)  # type: ignore[assignment]
+    context: ContextT = field(default=None)  # ty: ignore[invalid-assignment]
     """The graph run context, typed by the graph's `context_schema`.
 
     Only available during `threads.create_run`.

libs/sdk-py/langgraph_sdk/sse.py

@@ -55,7 +55,7 @@ class BytesLineDecoder:
             # Include any existing buffer in the first portion of the
             # splitlines result.
             self.buffer.extend(lines[0])
-            lines = cast(list[BytesLike], [self.buffer, *lines[1:]])
+            lines = [self.buffer, *lines[1:]]
             self.buffer = bytearray()
 
         if not trailing_newline:
@@ -69,7 +69,7 @@ class BytesLineDecoder:
         if not self.buffer and not self.trailing_cr:
             return []
 
-        lines = [self.buffer]
+        lines: list[BytesLike] = [self.buffer]
         self.buffer = bytearray()
         self.trailing_cr = False
         return lines
@@ -102,7 +102,7 @@ class SSEDecoder:
 
             sse = StreamPart(
                 event=self._event,
-                data=orjson.loads(self._data) if self._data else None,  # type: ignore[invalid-argument-type]
+                data=orjson.loads(self._data) if self._data else None,  # ty: ignore[invalid-argument-type]
                 id=self.last_event_id,
             )
 

libs/sdk-py/pyproject.toml

@@ -33,7 +33,7 @@ lint = [
     "ruff==0.15.12",
     "codespell",
     "mypy==1.20.2",
-    "ty==0.0.23",
+    "ty==0.0.33",
     "starlette",
 ]
 dev = [

libs/sdk-py/tests/test_client_stream.py

@@ -388,7 +388,7 @@ async def test_async_stream_v2_client_side_conversion() -> None:
             event="values", data={"messages": [{"role": "user", "content": "hi"}]}
         )
         yield StreamPart(event="updates|sub:abc", data={"node": {"out": 1}})
-        yield StreamPart(event="end", data=None)  # type: ignore[arg-type]
+        yield StreamPart(event="end", data=None)  # ty: ignore[invalid-argument-type]
 
     parts: list[StreamPartV2] = [part async for part in _wrap_stream_v2(mock_stream())]
     assert len(parts) == 3
@@ -420,7 +420,7 @@ def test_sync_stream_v2_client_side_conversion() -> None:
     def mock_stream() -> Any:
         yield StreamPart(event="metadata", data={"run_id": "r1"})
         yield StreamPart(event="values", data={"state": "full"})
-        yield StreamPart(event="end", data=None)  # type: ignore[arg-type]
+        yield StreamPart(event="end", data=None)  # ty: ignore[invalid-argument-type]
 
     parts: list[StreamPartV2] = list(_wrap_stream_v2_sync(mock_stream()))
     assert len(parts) == 2

libs/sdk-py/tests/test_crons_client.py

@@ -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()

libs/sdk-py/tests/test_encryption.py

@@ -67,6 +67,6 @@ class TestHandlerValidation:
 
         with pytest.raises(TypeError, match="must accept exactly 2 parameters"):
 
-            @encryption.encrypt.blob  # type: ignore[arg-type]
+            @encryption.encrypt.blob  # ty: ignore[invalid-argument-type]
             async def wrong_params(ctx):
                 return ctx

libs/sdk-py/uv.lock

@@ -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]]
@@ -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 = "." },
@@ -533,14 +533,14 @@ dev = [
     { name = "pytest-watch" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.20.2" },
     { name = "ruff", specifier = "==0.15.12" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.23" },
+    { name = "ty", specifier = "==0.0.33" },
 ]
 test = [
     { name = "pytest" },
@@ -1275,26 +1275,26 @@ wheels = [
 
 [[package]]
 name = "ty"
-version = "0.0.23"
+version = "0.0.33"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/75/ba/d3c998ff4cf6b5d75b39356db55fe1b7caceecc522b9586174e6a5dee6f7/ty-0.0.23.tar.gz", hash = "sha256:5fb05db58f202af366f80ef70f806e48f5237807fe424ec787c9f289e3f3a4ef", size = 5341461, upload-time = "2026-03-13T12:34:23.125Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/84/44/9478c50c266826c1bf30d1692e589755bffa8f1c0a3eb7af8a346c255991/ty-0.0.33.tar.gz", hash = "sha256:46d63bda07403322cb6c28ccfdd5536be916e13df725c29f7ccd0a21f06bd9e8", size = 5559373, upload-time = "2026-04-28T10:45:13.18Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f4/21/aab32603dfdfacd4819e52fa8c6074e7bd578218a5142729452fc6a62db6/ty-0.0.23-py3-none-linux_armv6l.whl", hash = "sha256:e810eef1a5f1cfc0731a58af8d2f334906a96835829767aed00026f1334a8dd7", size = 10329096, upload-time = "2026-03-13T12:34:09.432Z" },
-    { url = "https://files.pythonhosted.org/packages/9f/a9/dd3287a82dce3df546ec560296208d4905dcf06346b6e18c2f3c63523bd1/ty-0.0.23-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e43d36bd89a151ddcad01acaeff7dcc507cb73ff164c1878d2d11549d39a061c", size = 10156631, upload-time = "2026-03-13T12:34:53.122Z" },
-    { url = "https://files.pythonhosted.org/packages/0f/01/3f25909b02fac29bb0a62b2251f8d62e65d697781ffa4cf6b47a4c075c85/ty-0.0.23-py3-none-macosx_11_0_arm64.whl", hash = "sha256:bd6a340969577b4645f231572c4e46012acba2d10d4c0c6570fe1ab74e76ae00", size = 9653211, upload-time = "2026-03-13T12:34:15.049Z" },
-    { url = "https://files.pythonhosted.org/packages/d5/60/bfc0479572a6f4b90501c869635faf8d84c8c68ffc5dd87d04f049affabc/ty-0.0.23-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:341441783e626eeb7b1ec2160432956aed5734932ab2d1c26f94d0c98b229937", size = 10156143, upload-time = "2026-03-13T12:34:34.468Z" },
-    { url = "https://files.pythonhosted.org/packages/3a/81/8a93e923535a340f54bea20ff196f6b2787782b2f2f399bd191c4bc132d6/ty-0.0.23-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:8ce1dc66c26d4167e2c78d12fa870ef5a7ec9cc344d2baaa6243297cfa88bd52", size = 10136632, upload-time = "2026-03-13T12:34:28.832Z" },
-    { url = "https://files.pythonhosted.org/packages/da/cb/2ac81c850c58acc9f976814404d28389c9c1c939676e32287b9cff61381e/ty-0.0.23-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:bae1e7a294bf8528836f7617dc5c360ea2dddb63789fc9471ae6753534adca05", size = 10655025, upload-time = "2026-03-13T12:34:37.105Z" },
-    { url = "https://files.pythonhosted.org/packages/b5/9b/bac771774c198c318ae699fc013d8cd99ed9caf993f661fba11238759244/ty-0.0.23-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d2b162768764d9dc177c83fb497a51532bb67cbebe57b8fa0f2668436bf53f3c", size = 11230107, upload-time = "2026-03-13T12:34:20.751Z" },
-    { url = "https://files.pythonhosted.org/packages/14/09/7644fb0e297265e18243f878aca343593323b9bb19ed5278dcbc63781be0/ty-0.0.23-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:d28384e48ca03b34e4e2beee0e230c39bbfb68994bb44927fec61ef3642900da", size = 10934177, upload-time = "2026-03-13T12:34:17.904Z" },
-    { url = "https://files.pythonhosted.org/packages/18/14/69a25a0cad493fb6a947302471b579a03516a3b00e7bece77fdc6b4afb9b/ty-0.0.23-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:559d9a299df793cb7a7902caed5eda8a720ff69164c31c979673e928f02251ee", size = 10752487, upload-time = "2026-03-13T12:34:31.785Z" },
-    { url = "https://files.pythonhosted.org/packages/9d/2a/42fc3cbccf95af0a62308ebed67e084798ab7a85ef073c9986ef18032743/ty-0.0.23-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:32a7b8a14a98e1d20a9d8d2af23637ed7efdb297ac1fa2450b8e465d05b94482", size = 10133007, upload-time = "2026-03-13T12:34:42.838Z" },
-    { url = "https://files.pythonhosted.org/packages/e1/69/307833f1b52fa3670e0a1d496e43ef7df556ecde838192d3fcb9b35e360d/ty-0.0.23-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:6f803b9b9cca87af793467973b9abdd4b83e6b96d9b5e749d662cff7ead70b6d", size = 10169698, upload-time = "2026-03-13T12:34:12.351Z" },
-    { url = "https://files.pythonhosted.org/packages/89/ae/5dd379ec22d0b1cba410d7af31c366fcedff191d5b867145913a64889f66/ty-0.0.23-py3-none-musllinux_1_2_i686.whl", hash = "sha256:4a0bf086ec8e2197b7ea7ebfcf4be36cb6a52b235f8be61647ef1b2d99d6ffd3", size = 10346080, upload-time = "2026-03-13T12:34:40.012Z" },
-    { url = "https://files.pythonhosted.org/packages/98/c7/dfc83203d37998620bba9c4873a080c8850a784a8a46f56f8163c5b4e320/ty-0.0.23-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:252539c3fcd7aeb9b8d5c14e2040682c3e1d7ff640906d63fd2c4ce35865a4ba", size = 10848162, upload-time = "2026-03-13T12:34:45.421Z" },
-    { url = "https://files.pythonhosted.org/packages/89/08/05481511cfbcc1fd834b6c67aaae090cb609a079189ddf2032139ccfc490/ty-0.0.23-py3-none-win32.whl", hash = "sha256:51b591d19eef23bbc3807aef77d38fa1f003c354e1da908aa80ea2dca0993f77", size = 9748283, upload-time = "2026-03-13T12:34:50.607Z" },
-    { url = "https://files.pythonhosted.org/packages/31/2e/eaed4ff5c85e857a02415084c394e02c30476b65e158eec1938fdaa9a205/ty-0.0.23-py3-none-win_amd64.whl", hash = "sha256:1e137e955f05c501cfbb81dd2190c8fb7d01ec037c7e287024129c722a83c9ad", size = 10698355, upload-time = "2026-03-13T12:34:26.134Z" },
-    { url = "https://files.pythonhosted.org/packages/91/29/b32cb7b4c7d56b9ed50117f8ad6e45834aec293e4cb14749daab4e9236d5/ty-0.0.23-py3-none-win_arm64.whl", hash = "sha256:a0399bd13fd2cd6683fd0a2d59b9355155d46546d8203e152c556ddbdeb20842", size = 10155890, upload-time = "2026-03-13T12:34:48.082Z" },
+    { url = "https://files.pythonhosted.org/packages/e9/24/e287388c63a19191be26b32ff4dbd06029834068150ebe2532939bc4c851/ty-0.0.33-py3-none-linux_armv6l.whl", hash = "sha256:94d0a9d2234261a8911396d59e506b5923fe0971dbda43b9dcea287936887fcc", size = 11021308, upload-time = "2026-04-28T10:45:43.34Z" },
+    { url = "https://files.pythonhosted.org/packages/00/ca/ba1eed819895bd239fba8ee35dfcd5fcb266c203b0914a17a59579096bb5/ty-0.0.33-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e4a2b5ba078f90de342f56b5f7979bb77c9b9b1d8625a041352ffc6ee93c4073", size = 10777272, upload-time = "2026-04-28T10:45:32.905Z" },
+    { url = "https://files.pythonhosted.org/packages/25/a8/c3131d37b44b3fea1d6654a1c929a0cd0873822f77a90482b8ec28f6fbbd/ty-0.0.33-py3-none-macosx_11_0_arm64.whl", hash = "sha256:84ff5707825e9af9668d2bcf66975f93e520a63b524ab494e3a8265735be2563", size = 10201078, upload-time = "2026-04-28T10:45:23.374Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/db/d8e37ff0045810cc65e1ff36aa0da0a2253c05659787ac987df8a16c7897/ty-0.0.33-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e375285736f57886868e7af0b11c7b0ec5b6543fa15e7ad2a714fed9f077d4e0", size = 10732347, upload-time = "2026-04-28T10:45:21.444Z" },
+    { url = "https://files.pythonhosted.org/packages/e0/1a/20e83a412506a918e4684fc67b567cf7cc13b105470b3428cb23c3d5aa13/ty-0.0.33-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:5680f6350c3b4e46b8bff6d7bb132366ea239463d6cad4892725d06046e65464", size = 10808238, upload-time = "2026-04-28T10:45:38.565Z" },
+    { url = "https://files.pythonhosted.org/packages/5d/4b/d0a39f4464dc6cb4cc2c159473ce216bd1846bfb684c0323a3cb36dce5c6/ty-0.0.33-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c5535538bad8d0f7e62bcdff02197cdb30e41451d80b35d27e17d128f2e1dc5d", size = 11288348, upload-time = "2026-04-28T10:45:08.419Z" },
+    { url = "https://files.pythonhosted.org/packages/35/7e/f1745e0f9583363d7a83d9a4990fc244f76ecc30840ddad83dc16a33c52d/ty-0.0.33-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:da196c42bbbc069e1e21e3e52107c061aa9660352dae57a41930690b56e2c02d", size = 11789907, upload-time = "2026-04-28T10:45:19.064Z" },
+    { url = "https://files.pythonhosted.org/packages/a5/71/25f39f46a12d662859d45bc648555d0661044eb43db6b5648c9947487da9/ty-0.0.33-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:9281672921ef6d4460e03146b5e6c18cb1a3e3a3b8a1a88f6f33226d05a469b7", size = 11500774, upload-time = "2026-04-28T10:45:48.012Z" },
+    { url = "https://files.pythonhosted.org/packages/94/ec/136959ecbb7c71cb90537f5aea441c73f4ab24612868a6ecdc9d7444d32d/ty-0.0.33-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:82c1b8f303f82da64e878108e764be3ecbcd7c9903ac0a7f7031614ed00b97ab", size = 11360314, upload-time = "2026-04-28T10:45:05.402Z" },
+    { url = "https://files.pythonhosted.org/packages/cf/95/32809575c222f00beed498cb728e9290a0f5009f930025381bb7253b2206/ty-0.0.33-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:efe3af412c9ff67bce5fa37d0a2b0d8555c24072b145a5bac6c79637f1c83abe", size = 10707785, upload-time = "2026-04-28T10:45:10.836Z" },
+    { url = "https://files.pythonhosted.org/packages/13/89/c8e9531f7aa4a093359e15fa32c8e1277fbbe90d16894d7c6032d29f4b34/ty-0.0.33-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:aeec29c91ea768601747da546c3efc20b72c2fb1bd52bcc786a5c6eeff51d27b", size = 10834987, upload-time = "2026-04-28T10:45:40.738Z" },
+    { url = "https://files.pythonhosted.org/packages/31/16/9835fbcf5338af1a1917bd28fdb8a7193c210b83f243aa286fa9f79cb3ad/ty-0.0.33-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a535977c52bbb5f7e96b8b70a6ad375ad077f4a9ff2492508ea3816a2b403819", size = 10968968, upload-time = "2026-04-28T10:45:30.26Z" },
+    { url = "https://files.pythonhosted.org/packages/36/69/64c76aabc1bc70c7f24b686cd93c3407f8ea430905e395f59bf9603ef571/ty-0.0.33-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:1d732facf39fcb221ba279d469c5040d37883e964f123b1563888efd34818180", size = 11458077, upload-time = "2026-04-28T10:45:45.971Z" },
+    { url = "https://files.pythonhosted.org/packages/91/84/fae27b0c4718776a298690d31ca4cc1995f2e3e1c63a7b59e84c41498e9a/ty-0.0.33-py3-none-win32.whl", hash = "sha256:d90960b574428dc252f85e8598ec5fcb7f619794196b2fc95a90da075ed4681c", size = 10345364, upload-time = "2026-04-28T10:45:16.836Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/a0/a2938b23ae3e1a09a2d7c189e2ac5f7113676bae4e0e23948b568e18e5f8/ty-0.0.33-py3-none-win_amd64.whl", hash = "sha256:c1c3aec62c44de610c6e95f0a4e97ac3dbc07934bfdbf1fd90d758c9ff72f48e", size = 11342470, upload-time = "2026-04-28T10:45:26.455Z" },
+    { url = "https://files.pythonhosted.org/packages/ab/62/7fb948aace38d2f6329261bb33c035a8484549c74f1db28649c7a4c6fed9/ty-0.0.33-py3-none-win_arm64.whl", hash = "sha256:0d44f99ba1b441e55e2aa301b2ac0a21112784931b46a5f66f4ea9efe5620d97", size = 10742673, upload-time = "2026-04-28T10:45:35.555Z" },
 ]
 
 [[package]]