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

simplify: handlers always receive state as first arg, drop takes_input flag
refactor(langgraph): implement error_handler as a policy field, not a hidden node
2026-10-04 23:45:08 +02:00 · 2026-05-11 16:22:10 -07:00 · 2026-05-11 16:04:47 -07:00 · 2026-05-11 15:59:22 -07:00 · 2026-05-11 14:20:18 -07:00 · 2026-05-11 14:19:50 -07:00
23 changed files with 629 additions and 243 deletions
@@ -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]]
@@ -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]]
@@ -60,20 +60,29 @@ class CheckpointMetadata(TypedDict, total=False):
    """
    run_id: str
    """The ID of the run that created this checkpoint."""
-    delta_updates_since_snapshot: dict[str, int]
-    """Per-channel update count since the last `_DeltaSnapshot` was written.
+    counters_since_delta_snapshot: dict[str, tuple[int, int]]
+    """Per-channel counters since the last `_DeltaSnapshot` was written.

    !!! warning "Beta"

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

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


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

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

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

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

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

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


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

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

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

    Parameters:
        reducer: `(state, list[writes]) -> new_state`. Must be deterministic
@@ -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
@@ -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,
            )
@@ -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
    )


@@ -8,11 +8,11 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
    Checkpoint,
-    CheckpointMetadata,
 )
 from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.types import _DeltaSnapshot

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

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

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


 def create_checkpoint(
@@ -69,7 +74,7 @@ def create_checkpoint(
    is written into `channel_values[k]`. Other delta channels are omitted
    from `channel_values` — the ancestor walk reconstructs their state
    from `checkpoint_writes`. Callers compute the set via
-    `delta_channels_to_snapshot(channels, counts)`; defaults to empty
+    `delta_channels_to_snapshot(channels, counters)`; defaults to empty
    (no snapshots) when not provided.
    """
    ts = datetime.now(timezone.utc).isoformat()
@@ -231,17 +236,3 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
        versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
        updated_channels=checkpoint.get("updated_channels", None),
    )
-
-
-def read_delta_updates_since_snapshot(
-    metadata: CheckpointMetadata | None,
-) -> dict[str, int]:
-    """Read the per-channel update counter from checkpoint metadata.
-
-    Returns an empty dict for missing/None metadata; the dict is
-    `total=False` on `CheckpointMetadata`, so absence means "no prior
-    delta-channel activity tracked."
-    """
-    if not metadata:
-        return {}
-    return dict(metadata.get("delta_updates_since_snapshot", {}) or {})
@@ -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,
@@ -279,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
@@ -303,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()
@@ -545,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
@@ -597,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
@@ -978,35 +983,41 @@ class PregelLoop:
        if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
            # checkpoint already saved
            return
-        # Per-delta-channel update bookkeeping.
+        # Per-delta-channel counter bookkeeping.
+        #
+        # Each delta channel tracks a (updates, supersteps) tuple:
+        # - `updates` increments only when the channel is written this step.
+        # - `supersteps` increments every superstep regardless.
        #
        # `_put_checkpoint` is called once per superstep with a fresh
        # metadata dict (source="input"|"loop"|"fork") — those are the
-        # intermediate calls that bump the count by +1 for each delta
-        # channel touched that step. In exit mode,
+        # intermediate calls that bump counters. In exit mode,
        # `_suppress_interrupt`(will rename to _on_loop_exit soon)
        # additionally calls `_put_checkpoint(self.checkpoint_metadata)` AT
        # EXIT to commit the final checkpoint — this runs *after* the last
        # intermediate call already counted the last superstep. So the
        # exit call must NOT bump again or it would double-count the last
-        # superstep. (Sync/async durability does not call `_put_checkpoint`
-        # at exit, so the issue only surfaces in exit mode. force_delta_snapshot
-        # used to mask this latent bug by resetting every count to 0.)
+        # superstep.
        if not exiting:
-            prev_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+            prev_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
            )
-            new_counts = dict(prev_counts)
-            if self.updated_channels:
-                for ch_name in self.updated_channels:
-                    if isinstance(self.channels.get(ch_name), DeltaChannel):
-                        new_counts[ch_name] = new_counts.get(ch_name, 0) + 1
+            new_counters: dict[str, tuple[int, int]] = {}
+            updated = self.updated_channels or set()
+            for ch_name, ch in self.channels.items():
+                if not isinstance(ch, DeltaChannel):
+                    continue
+                u, s = prev_counters.get(ch_name, (0, 0))
+                s += 1
+                if ch_name in updated:
+                    u += 1
+                new_counters[ch_name] = (u, s)
            metadata["step"] = self.step
            metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
            self.checkpoint_metadata = metadata
        else:
-            new_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+            new_counters = dict(
+                self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
            )
        # do checkpoint?
        do_checkpoint = self._checkpointer_put_after_previous is not None and (
@@ -1014,7 +1025,7 @@ class PregelLoop:
        )
        # create new checkpoint
        channels_to_snapshot = (
-            delta_channels_to_snapshot(self.channels, new_counts)
+            delta_channels_to_snapshot(self.channels, new_counters)
            if do_checkpoint
            else set()
        )
@@ -1030,11 +1041,12 @@ class PregelLoop:
            channels_to_snapshot=channels_to_snapshot,
        )
        for k in channels_to_snapshot:
-            new_counts[k] = 0
-        if new_counts:
-            self.checkpoint_metadata["delta_updates_since_snapshot"] = new_counts
-        elif "delta_updates_since_snapshot" in self.checkpoint_metadata:
-            del self.checkpoint_metadata["delta_updates_since_snapshot"]
+            new_counters[k] = (0, 0)
+        non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
+        if non_zero:
+            self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
+        elif "counters_since_delta_snapshot" in self.checkpoint_metadata:
+            del self.checkpoint_metadata["counters_since_delta_snapshot"]
        # sanitize TASK channel in the checkpoint before saving (durability=="exit")
        if TASKS in self.checkpoint["channel_values"] and any(
            isinstance(channel, UntrackedValue) for channel in self.channels.values()
@@ -1109,8 +1121,10 @@ class PregelLoop:
        ):
            return

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

        pending = [
            (step, tid, ch, v)
@@ -1359,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__(
@@ -1380,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,
        )
@@ -1442,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,
@@ -1467,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})
@@ -1478,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)
@@ -1612,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__(
@@ -1633,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,
        )
@@ -1697,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,
@@ -1722,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})
@@ -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:
@@ -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.
@@ -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],
@@ -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
@@ -628,6 +628,7 @@ class PregelExecutableTask:
    writers: Sequence[Runnable] = ()
    subgraphs: Sequence[PregelProtocol] = ()
    timeout: TimeoutPolicy | None = None
+    error_handler: Runnable | None = None


 class StateSnapshot(NamedTuple):
@@ -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",
@@ -160,8 +160,8 @@ async def test_exit_resumed_run_sub_freq() -> None:


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

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


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

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


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

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


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