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/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/_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,
@@ -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
@@ -1368,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__(
@@ -1389,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,
         )
@@ -1451,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,
@@ -1476,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})
@@ -1487,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)
@@ -1621,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__(
@@ -1642,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,
         )
@@ -1706,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,
@@ -1731,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})

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/tests/test_retry.py

@@ -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
+# ---------------------------------------------------------------------------
+
+