feat(langgraph)!: remove Pregel construction and NodeBuilder from the public API

Building a graph by constructing `Pregel(...)` with `NodeBuilder` nodes
was never meant to be public. Graphs are built with `StateGraph` or
`@entrypoint`.

- Delete `NodeBuilder`; `langgraph.pregel` now exports only `Pregel`.
- `Pregel(...)` raises `TypeError` pointing to `StateGraph` and
  `@entrypoint`. `Pregel` stays public as the type that `compile()` and
  `@entrypoint` return, for type annotations and `isinstance` checks.
- `@entrypoint` builds its graph through a private `_EntrypointPregel`
  subclass, the same way `StateGraph` uses `CompiledStateGraph`.
- Rewrite the `Pregel` docstring to describe the compiled graph type and
  drop the hand-built graph examples.
- Engine tests that build graphs by hand now use a test-only copy of
  `NodeBuilder` and a `RawPregel` subclass in `tests/raw_pregel.py`.

BREAKING CHANGE: `from langgraph.pregel import NodeBuilder` and calling
`Pregel(...)` directly no longer work. Build graphs with `StateGraph` or
`@entrypoint`.

libs/checkpoint-conformance/uv.lock

@@ -279,7 +279,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/checkpoint-postgres/uv.lock

@@ -278,7 +278,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/checkpoint-sqlite/uv.lock

@@ -285,7 +285,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/checkpoint/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.10"

libs/checkpoint/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/langgraph/langgraph/channels/delta.py

@@ -60,7 +60,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         typ: The value type (e.g. `list`, `dict`). Inferred automatically
             from the outer type when used inside `Annotated[T, DeltaChannel(...)]`.
         snapshot_frequency: Every Nth update to this channel writes a snapshot
-            blob (default `1000`). Must be a positive int.
+            blob (default `100`). Must be a positive int.
     """
 
     __slots__ = ("value", "reducer", "snapshot_frequency")
@@ -71,7 +71,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         reducer: Callable[[Any, Sequence[Any]], Any],
         typ: type[Value] | None = None,
         *,
-        snapshot_frequency: int = 1000,
+        snapshot_frequency: int = 100,
     ) -> None:
         if snapshot_frequency <= 0:
             raise ValueError(

libs/langgraph/langgraph/func/__init__.py

@@ -255,6 +255,11 @@ R = TypeVar("R")
 S = TypeVar("S")
 
 
+class _EntrypointPregel(Pregel):
+    """The graph that `@entrypoint` builds. `Pregel` itself cannot be constructed
+    directly."""
+
+
 # The decorator was wrapped in a class to support the `final` attribute.
 # In this form, the `final` attribute should play nicely with IDE autocompletion,
 # and type checking tools.
@@ -573,7 +578,7 @@ class entrypoint(Generic[ContextT]):
                 else:
                     output_type = save_type = sig.return_annotation
 
-        graph: Pregel[Any, ContextT, Any, Any] = Pregel(
+        graph: Pregel[Any, ContextT, Any, Any] = _EntrypointPregel(
             nodes={
                 func.__name__: PregelNode(
                     bound=bound,

libs/langgraph/langgraph/pregel/__init__.py

@@ -1,3 +1,3 @@
-from langgraph.pregel.main import NodeBuilder, Pregel
+from langgraph.pregel.main import Pregel
 
-__all__ = ("Pregel", "NodeBuilder")
+__all__ = ("Pregel",)

libs/langgraph/langgraph/pregel/_draw.py

@@ -228,6 +228,10 @@ def draw_graph(
             metadata["__interrupt"] = "before"
         elif name in interrupt_after_nodes:
             metadata["__interrupt"] = "after"
+        # Duck-typed so core doesn't import ToolNode from langgraph-prebuilt.
+        tools_by_name = getattr(node.bound, "tools_by_name", None)
+        if isinstance(tools_by_name, Mapping) and tools_by_name:
+            metadata["__tools"] = list(tools_by_name)
         graph.add_node(node.bound, name, metadata=metadata or None)
     # add start node
     if START not in nodes:

libs/langgraph/langgraph/pregel/_executor.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import concurrent.futures
 import time
+import weakref
 from collections.abc import Awaitable, Callable, Coroutine
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from contextvars import copy_context
@@ -37,6 +38,13 @@ class Submit(Protocol[P, T]):
     ) -> concurrent.futures.Future[T]: ...
 
 
+# Futures whose exception a node's error handler took over; the executors
+# don't re-raise them on exit.
+SKIP_RERAISE_SET: weakref.WeakSet[concurrent.futures.Future | asyncio.Future] = (
+    weakref.WeakSet()
+)
+
+
 class BackgroundExecutor(AbstractContextManager):
     """A context manager that runs sync tasks in the background.
     Uses a thread pool executor to delegate tasks to separate threads.
@@ -111,7 +119,7 @@ class BackgroundExecutor(AbstractContextManager):
         if exc_type is None:
             # re-raise the first exception that occurred in a task
             for task, (_, reraise) in tasks.items():
-                if not reraise:
+                if not reraise or task in SKIP_RERAISE_SET:
                     continue
                 try:
                     task.result()
@@ -202,7 +210,7 @@ class AsyncBackgroundExecutor(AbstractAsyncContextManager):
         if exc_type is None:
             # re-raise the first exception that occurred in a task
             for task, (_, reraise) in tasks.items():
-                if not reraise:
+                if not reraise or task in SKIP_RERAISE_SET:
                     continue
                 try:
                     if exc := task.exception():

libs/langgraph/langgraph/pregel/_retry.py

@@ -34,6 +34,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_THREAD_ID,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     NS_SEP,
+    RETURN,
 )
 from langgraph._internal._runnable import create_task_in_config_context
 from langgraph._internal._timeout import sync_timeout_unsupported
@@ -733,15 +734,14 @@ async def arun_with_retry(
                 if stream:
                     async for _ in task.proc.astream(task.input, config):
                         pass
-                    break
+                    return _return_value(task)
                 return await task.proc.ainvoke(task.input, config)
             result = await _arun_with_timeout(
                 task, config, resolved_timeout, attempt_ctx, stream=stream
             )
             _finish_timed_attempt(config, attempt_ctx)
             if stream:
-                # if successful, end
-                break
+                return _return_value(task)
             return result
         except ParentCommand as exc:
             ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
@@ -831,6 +831,12 @@ async def arun_with_retry(
             config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})
 
 
+def _return_value(task: PregelExecutableTask) -> Any:
+    # Streaming drains the task instead of returning its result, so a caller
+    # awaiting a `@task` needs the value the task wrote.
+    return next((value for channel, value in task.writes if channel == RETURN), None)
+
+
 def _should_retry_on(retry_policy: RetryPolicy, exc: Exception) -> bool:
     """Check if the given exception should be retried based on the retry policy."""
     if isinstance(retry_policy.retry_on, Sequence):

libs/langgraph/langgraph/pregel/_runner.py

@@ -43,7 +43,7 @@ from langgraph._internal._typing import MISSING
 from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume
 from langgraph.pregel._algo import Call
-from langgraph.pregel._executor import Submit
+from langgraph.pregel._executor import SKIP_RERAISE_SET, Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
 from langgraph.pregel._task_status import CONTROL_WRITES
 from langgraph.types import (
@@ -68,10 +68,6 @@ EXCLUDED_FRAME_FNAMES = (
     "concurrent/futures/_base.py",
 )
 
-SKIP_RERAISE_SET: weakref.WeakSet[concurrent.futures.Future | asyncio.Future] = (
-    weakref.WeakSet()
-)
-
 
 class FuturesDict(Generic[F, E], dict[F, PregelExecutableTask | None]):
     event: E
@@ -239,6 +235,8 @@ class PregelRunner:
                         fut.set_exception(exc)
                         futures.done.add(fut)
                 elif reraise and id(exc) not in self._handled_exception_ids:
+                    # let the caller stream this task's result before it raises
+                    yield
                     if tb := exc.__traceback__:
                         while tb.tb_next is not None and any(
                             tb.tb_frame.f_code.co_filename.endswith(name)
@@ -430,6 +428,8 @@ class PregelRunner:
                         fut.set_exception(exc)
                         futures.done.add(fut)
                 elif reraise and id(exc) not in self._handled_exception_ids:
+                    # let the caller stream this task's result before it raises
+                    yield
                     if tb := exc.__traceback__:
                         while tb.tb_next is not None and any(
                             tb.tb_frame.f_code.co_filename.endswith(name)

libs/langgraph/langgraph/pregel/_tools.py

@@ -9,6 +9,7 @@ from langchain_core.callbacks import BaseCallbackHandler
 
 from langgraph._internal._constants import NS_SEP
 from langgraph.constants import TAG_NOSTREAM
+from langgraph.errors import GraphDrained, GraphInterrupt
 from langgraph.pregel.protocol import StreamChunk
 
 try:
@@ -188,6 +189,10 @@ class StreamToolCallHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             return
         ns, tool_call_id, token = info
         self._reset_writer(token)
+        # An interrupted or drained tool is paused, not failed: it runs again
+        # on resume and reports its start and end then.
+        if isinstance(error, (GraphInterrupt, GraphDrained)):
+            return
         self.stream(
             (
                 ns,

libs/langgraph/langgraph/pregel/_validate.py

@@ -47,9 +47,7 @@ def validate_graph(
                             f"not in known channels: '{repr(sorted(channels))[:100]}'"
                         )
         else:
-            raise TypeError(
-                f"Invalid node type {type(node)}, expected PregelNode or NodeBuilder"
-            )
+            raise TypeError(f"Invalid node type {type(node)}, expected PregelNode")
 
     for chan in subscribed_channels:
         if chan not in channels:

libs/langgraph/langgraph/pregel/main.py

@@ -17,7 +17,6 @@ from collections.abc import (
     Sequence,
 )
 from dataclasses import is_dataclass, replace
-from datetime import timedelta
 from functools import partial
 from inspect import isclass
 from typing import (
@@ -35,7 +34,6 @@ from langchain_core.globals import get_debug
 from langchain_core.runnables import (
     RunnableSequence,
 )
-from langchain_core.runnables.base import Input, Output
 from langchain_core.runnables.config import (
     RunnableConfig,
     get_async_callback_manager_for_config,
@@ -93,13 +91,6 @@ from langgraph._internal._queue import (  # type: ignore[attr-defined]
     AsyncQueue,
     SyncQueue,
 )
-from langgraph._internal._runnable import (
-    Runnable,
-    RunnableLike,
-    RunnableSeq,
-    coerce_to_runnable,
-)
-from langgraph._internal._timeout import coerce_timeout_policy
 from langgraph._internal._typing import MISSING, DeprecatedKwargs
 from langgraph.callbacks import (
     GraphInterruptEvent,
@@ -155,7 +146,7 @@ from langgraph.pregel._messages import (
     StreamMessagesHandlerV2,
     ensure_message_ids,
 )
-from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
+from langgraph.pregel._read import PregelNode
 from langgraph.pregel._retry import RetryPolicy
 from langgraph.pregel._runner import PregelRunner
 from langgraph.pregel._task_status import read_task_statuses
@@ -165,7 +156,6 @@ from langgraph.pregel._utils import (
     validate_timeout_supported,
 )
 from langgraph.pregel._validate import validate_graph, validate_keys
-from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.pregel.debug import get_bolded_text, get_colored_text, tasks_w_writes
 from langgraph.pregel.protocol import PregelProtocol, StreamChunk, StreamProtocol
 from langgraph.runtime import (
@@ -197,7 +187,6 @@ from langgraph.types import (
     StateUpdate,
     StreamMode,
     StreamPart,
-    TimeoutPolicy,
     ensure_valid_checkpointer,
 )
 from langgraph.typing import ContextT, InputT, OutputT, StateT
@@ -208,182 +197,7 @@ try:
 except ImportError:
     _StreamingCallbackHandler = None  # type: ignore
 
-__all__ = ("NodeBuilder", "Pregel")
-
-_WriteValue = Callable[[Input], Output] | Any
-
-
-class NodeBuilder:
-    __slots__ = (
-        "_channels",
-        "_triggers",
-        "_tags",
-        "_metadata",
-        "_writes",
-        "_bound",
-        "_retry_policy",
-        "_cache_policy",
-        "_timeout",
-    )
-
-    _channels: str | list[str]
-    _triggers: list[str]
-    _tags: list[str]
-    _metadata: dict[str, Any]
-    _writes: list[ChannelWriteEntry]
-    _bound: Runnable
-    _retry_policy: list[RetryPolicy]
-    _cache_policy: CachePolicy | None
-    _timeout: TimeoutPolicy | None
-
-    def __init__(
-        self,
-    ) -> None:
-        self._channels = []
-        self._triggers = []
-        self._tags = []
-        self._metadata = {}
-        self._writes = []
-        self._bound = DEFAULT_BOUND
-        self._retry_policy = []
-        self._cache_policy = None
-        self._timeout = None
-
-    def subscribe_only(
-        self,
-        channel: str,
-    ) -> Self:
-        """Subscribe to a single channel."""
-        if not self._channels:
-            self._channels = channel
-        else:
-            raise ValueError(
-                "Cannot subscribe to single channels when other channels are already subscribed to"
-            )
-
-        self._triggers.append(channel)
-
-        return self
-
-    def subscribe_to(
-        self,
-        *channels: str,
-        read: bool = True,
-    ) -> Self:
-        """Add channels to subscribe to.
-
-        Node will be invoked when any of these channels are updated, with a dict of the
-        channel values as input.
-
-        Args:
-            channels: Channel name(s) to subscribe to
-            read: If `True`, the channels will be included in the input to the node.
-                Otherwise, they will trigger the node without being sent in input.
-
-        Returns:
-            Self for chaining
-        """
-        if isinstance(self._channels, str):
-            raise ValueError(
-                "Cannot subscribe to channels when subscribed to a single channel"
-            )
-        if read:
-            if not self._channels:
-                self._channels = list(channels)
-            else:
-                self._channels.extend(channels)
-
-        if isinstance(channels, str):
-            self._triggers.append(channels)
-        else:
-            self._triggers.extend(channels)
-
-        return self
-
-    def read_from(
-        self,
-        *channels: str,
-    ) -> Self:
-        """Adds the specified channels to read from, without subscribing to them."""
-        assert isinstance(self._channels, list), (
-            "Cannot read additional channels when subscribed to single channels"
-        )
-        self._channels.extend(channels)
-        return self
-
-    def do(
-        self,
-        node: RunnableLike,
-    ) -> Self:
-        """Adds the specified node."""
-        if self._bound is not DEFAULT_BOUND:
-            self._bound = RunnableSeq(
-                self._bound, coerce_to_runnable(node, name=None, trace=True)
-            )
-        else:
-            self._bound = coerce_to_runnable(node, name=None, trace=True)
-        return self
-
-    def write_to(
-        self,
-        *channels: str | ChannelWriteEntry,
-        **kwargs: _WriteValue,
-    ) -> Self:
-        """Add channel writes.
-
-        Args:
-            *channels: Channel names to write to.
-            **kwargs: Channel name and value mappings.
-
-        Returns:
-            Self for chaining
-        """
-        self._writes.extend(
-            ChannelWriteEntry(c) if isinstance(c, str) else c for c in channels
-        )
-        self._writes.extend(
-            ChannelWriteEntry(k, mapper=v)
-            if callable(v)
-            else ChannelWriteEntry(k, value=v)
-            for k, v in kwargs.items()
-        )
-
-        return self
-
-    def meta(self, *tags: str, **metadata: Any) -> Self:
-        """Add tags or metadata to the node."""
-        self._tags.extend(tags)
-        self._metadata.update(metadata)
-        return self
-
-    def add_retry_policies(self, *policies: RetryPolicy) -> Self:
-        """Adds retry policies to the node."""
-        self._retry_policy.extend(policies)
-        return self
-
-    def add_cache_policy(self, policy: CachePolicy) -> Self:
-        """Adds cache policies to the node."""
-        self._cache_policy = policy
-        return self
-
-    def set_timeout(self, timeout: float | timedelta | TimeoutPolicy | None) -> Self:
-        """Set the per-attempt timeout policy for this node."""
-        self._timeout = coerce_timeout_policy(timeout)
-        return self
-
-    def build(self) -> PregelNode:
-        """Builds the node."""
-        return PregelNode(
-            channels=self._channels,
-            triggers=self._triggers,
-            tags=self._tags,
-            metadata=self._metadata,
-            writers=[ChannelWrite(self._writes)],
-            bound=self._bound,
-            retry_policy=self._retry_policy,
-            cache_policy=self._cache_policy,
-            timeout=self._timeout,
-        )
+__all__ = ("Pregel",)
 
 
 # Kwargs that ``stream_events(version="v3")`` / ``astream_events(version="v3")``
@@ -461,255 +275,22 @@ class Pregel(
     PregelProtocol[StateT, ContextT, InputT, OutputT],
     Generic[StateT, ContextT, InputT, OutputT],
 ):
-    """Pregel manages the runtime behavior for LangGraph applications.
+    """The compiled graph that runs a LangGraph application.
 
-    ## Overview
+    `StateGraph.compile()` and the `@entrypoint` decorator return an instance of
+    this class. It provides the methods that run and inspect the application,
+    such as `invoke`, `stream`, `get_state` and `update_state`.
 
-    Pregel combines [**actors**](https://en.wikipedia.org/wiki/Actor_model)
-    and **channels** into a single application.
-    **Actors** read data from channels and write data to channels.
-    Pregel organizes the execution of the application into multiple steps,
-    following the **Pregel Algorithm**/**Bulk Synchronous Parallel** model.
+    The runtime is named after Google's
+    [Pregel algorithm](https://research.google/pubs/pub37252/). It runs the
+    application in steps. In each step it plans which nodes to run, runs them in
+    parallel, and then applies their writes to the graph's state, so writes
+    become visible only in the next step. The run ends when no node is
+    scheduled, or when the recursion limit is reached.
 
-    Each step consists of three phases:
-
-    - **Plan**: Determine which **actors** to execute in this step. For example,
-        in the first step, select the **actors** that subscribe to the special
-        **input** channels; in subsequent steps,
-        select the **actors** that subscribe to channels updated in the previous step.
-    - **Execution**: Execute all selected **actors** in parallel,
-        until all complete, or one fails, or a timeout is reached. During this
-        phase, channel updates are invisible to actors until the next step.
-    - **Update**: Update the channels with the values written by the **actors**
-        in this step.
-
-    Repeat until no **actors** are selected for execution, or a maximum number of
-    steps is reached.
-
-    ## Actors
-
-    An **actor** is a `PregelNode`.
-    It subscribes to channels, reads data from them, and writes data to them.
-    It can be thought of as an **actor** in the Pregel algorithm.
-    `PregelNodes` implement LangChain's
-    Runnable interface.
-
-    ## Channels
-
-    Channels are used to communicate between actors (`PregelNodes`).
-    Each channel has a value type, an update type, and an update function – which
-    takes a sequence of updates and
-    modifies the stored value. Channels can be used to send data from one chain to
-    another, or to send data from a chain to itself in a future step. LangGraph
-    provides a number of built-in channels:
-
-    ### Basic channels: LastValue and Topic
-
-    - `LastValue`: The default channel, stores the last value sent to the channel,
-       useful for input and output values, or for sending data from one step to the next
-    - `Topic`: A configurable PubSub Topic, useful for sending multiple values
-       between *actors*, or for accumulating output. Can be configured to deduplicate
-       values, and/or to accumulate values over the course of multiple steps.
-
-    ### Advanced channels: Context and BinaryOperatorAggregate
-
-    - `Context`: exposes the value of a context manager, managing its lifecycle.
-        Useful for accessing external resources that require setup and/or teardown. e.g.
-        `client = Context(httpx.Client)`
-    - `BinaryOperatorAggregate`: stores a persistent value, updated by applying
-        a binary operator to the current value and each update
-        sent to the channel, useful for computing aggregates over multiple steps. e.g.
-        `total = BinaryOperatorAggregate(int, operator.add)`
-
-    ## Examples
-
-    Most users will interact with Pregel via a
-    [StateGraph (Graph API)][langgraph.graph.StateGraph] or via an
-    [entrypoint (Functional API)][langgraph.func.entrypoint].
-
-    However, for **advanced** use cases, Pregel can be used directly. If you're
-    not sure whether you need to use Pregel directly, then the answer is probably no
-    - you should use the Graph API or Functional API instead. These are higher-level
-    interfaces that will compile down to Pregel under the hood.
-
-    Here are some examples to give you a sense of how it works:
-
-    Example: Single node application
-        ```python
-        from langgraph.channels import EphemeralValue
-        from langgraph.pregel import Pregel, NodeBuilder
-
-        node1 = (
-            NodeBuilder().subscribe_only("a")
-            .do(lambda x: x + x)
-            .write_to("b")
-        )
-
-        app = Pregel(
-            nodes={"node1": node1},
-            channels={
-                "a": EphemeralValue(str),
-                "b": EphemeralValue(str),
-            },
-            input_channels=["a"],
-            output_channels=["b"],
-        )
-
-        app.invoke({"a": "foo"})
-        ```
-
-        ```con
-        {'b': 'foofoo'}
-        ```
-
-    Example: Using multiple nodes and multiple output channels
-        ```python
-        from langgraph.channels import LastValue, EphemeralValue
-        from langgraph.pregel import Pregel, NodeBuilder
-
-        node1 = (
-            NodeBuilder().subscribe_only("a")
-            .do(lambda x: x + x)
-            .write_to("b")
-        )
-
-        node2 = (
-            NodeBuilder().subscribe_to("b")
-            .do(lambda x: x["b"] + x["b"])
-            .write_to("c")
-        )
-
-
-        app = Pregel(
-            nodes={"node1": node1, "node2": node2},
-            channels={
-                "a": EphemeralValue(str),
-                "b": LastValue(str),
-                "c": EphemeralValue(str),
-            },
-            input_channels=["a"],
-            output_channels=["b", "c"],
-        )
-
-        app.invoke({"a": "foo"})
-        ```
-
-        ```con
-        {'b': 'foofoo', 'c': 'foofoofoofoo'}
-        ```
-
-    Example: Using a Topic channel
-        ```python
-        from langgraph.channels import LastValue, EphemeralValue, Topic
-        from langgraph.pregel import Pregel, NodeBuilder
-
-        node1 = (
-            NodeBuilder().subscribe_only("a")
-            .do(lambda x: x + x)
-            .write_to("b", "c")
-        )
-
-        node2 = (
-            NodeBuilder().subscribe_only("b")
-            .do(lambda x: x + x)
-            .write_to("c")
-        )
-
-
-        app = Pregel(
-            nodes={"node1": node1, "node2": node2},
-            channels={
-                "a": EphemeralValue(str),
-                "b": EphemeralValue(str),
-                "c": Topic(str, accumulate=True),
-            },
-            input_channels=["a"],
-            output_channels=["c"],
-        )
-
-        app.invoke({"a": "foo"})
-        ```
-
-        ```pycon
-        {"c": ["foofoo", "foofoofoofoo"]}
-        ```
-
-    Example: Using a `BinaryOperatorAggregate` channel
-        ```python
-        from langgraph.channels import EphemeralValue, BinaryOperatorAggregate
-        from langgraph.pregel import Pregel, NodeBuilder
-
-
-        node1 = (
-            NodeBuilder().subscribe_only("a")
-            .do(lambda x: x + x)
-            .write_to("b", "c")
-        )
-
-        node2 = (
-            NodeBuilder().subscribe_only("b")
-            .do(lambda x: x + x)
-            .write_to("c")
-        )
-
-
-        def reducer(current, update):
-            if current:
-                return current + " | " + update
-            else:
-                return update
-
-
-        app = Pregel(
-            nodes={"node1": node1, "node2": node2},
-            channels={
-                "a": EphemeralValue(str),
-                "b": EphemeralValue(str),
-                "c": BinaryOperatorAggregate(str, operator=reducer),
-            },
-            input_channels=["a"],
-            output_channels=["c"],
-        )
-
-        app.invoke({"a": "foo"})
-        ```
-
-        ```con
-        {'c': 'foofoo | foofoofoofoo'}
-        ```
-
-    Example: Introducing a cycle
-        This example demonstrates how to introduce a cycle in the graph, by having
-        a chain write to a channel it subscribes to.
-
-        Execution will continue until a `None` value is written to the channel.
-
-        ```python
-        from langgraph.channels import EphemeralValue
-        from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry
-
-        example_node = (
-            NodeBuilder()
-            .subscribe_only("value")
-            .do(lambda x: x + x if len(x) < 10 else None)
-            .write_to(ChannelWriteEntry(channel="value", skip_none=True))
-        )
-
-        app = Pregel(
-            nodes={"example_node": example_node},
-            channels={
-                "value": EphemeralValue(str),
-            },
-            input_channels=["value"],
-            output_channels=["value"],
-        )
-
-        app.invoke({"value": "a"})
-        ```
-
-        ```con
-        {'value': 'aaaaaaaaaaaaaaaa'}
-        ```
+    `Pregel` is not constructed directly. Build graphs with
+    [`StateGraph`][langgraph.graph.StateGraph] or
+    [`@entrypoint`][langgraph.func.entrypoint].
     """
 
     nodes: dict[str, PregelNode]
@@ -768,7 +349,7 @@ class Pregel(
     def __init__(
         self,
         *,
-        nodes: dict[str, PregelNode | NodeBuilder],
+        nodes: dict[str, PregelNode],
         channels: dict[str, BaseChannel | ManagedValueSpec] | None,
         auto_validate: bool = True,
         stream_mode: StreamMode = "values",
@@ -793,6 +374,13 @@ class Pregel(
         stream_transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
+        # LangGraph builds graphs through subclasses: `CompiledStateGraph` and
+        # the graph `@entrypoint` returns.
+        if type(self) is Pregel:
+            raise TypeError(
+                "`Pregel` cannot be constructed directly. "
+                "Build graphs with `StateGraph` or `@entrypoint`."
+            )
         if (
             config_type := deprecated_kwargs.get("config_type", MISSING)
         ) is not MISSING:
@@ -807,9 +395,7 @@ class Pregel(
 
         checkpointer = ensure_valid_checkpointer(checkpointer)
 
-        self.nodes = {
-            k: v.build() if isinstance(v, NodeBuilder) else v for k, v in nodes.items()
-        }
+        self.nodes = dict(nodes)
         self.channels = channels or {}
         if TASKS in self.channels and not isinstance(self.channels[TASKS], Topic):
             raise ValueError(
@@ -3310,7 +2896,9 @@ class Pregel(
                         True
                         for h in run_manager.handlers
                         if isinstance(h, _StreamingCallbackHandler)
-                        and not isinstance(h, StreamMessagesHandler)
+                        and not isinstance(
+                            h, (StreamMessagesHandler, StreamToolCallHandler)
+                        )
                     ),
                     False,
                 )

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.17"
+version = "1.2.18"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/__snapshots__/test_large_cases.ambr

@@ -133,6 +133,11 @@
             "ToolNode"
           ],
           "name": "tools"
+        },
+        "metadata": {
+          "__tools": [
+            "search_api"
+          ]
         }
       },
       {
@@ -221,6 +226,11 @@
             "ToolNode"
           ],
           "name": "tools"
+        },
+        "metadata": {
+          "__tools": [
+            "search_api"
+          ]
         }
       },
       {

libs/langgraph/tests/raw_pregel.py

@@ -0,0 +1,222 @@
+"""Build graphs directly on the Pregel engine, for tests of engine behavior.
+
+LangGraph does not expose this: applications build graphs with `StateGraph` or
+`@entrypoint`. Tests use it to set up channels, triggers and writes that those
+APIs do not produce on their own.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import timedelta
+from typing import Any
+
+from langchain_core.runnables.base import Input, Output
+from typing_extensions import Self
+
+from langgraph._internal._runnable import (
+    Runnable,
+    RunnableLike,
+    RunnableSeq,
+    coerce_to_runnable,
+)
+from langgraph._internal._timeout import coerce_timeout_policy
+from langgraph.pregel import Pregel
+from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
+from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
+from langgraph.types import CachePolicy, RetryPolicy, TimeoutPolicy
+
+__all__ = ("NodeBuilder", "RawPregel")
+
+_WriteValue = Callable[[Input], Output] | Any
+
+
+class NodeBuilder:
+    """Builds a `PregelNode` that reads and writes named channels."""
+
+    __slots__ = (
+        "_channels",
+        "_triggers",
+        "_tags",
+        "_metadata",
+        "_writes",
+        "_bound",
+        "_retry_policy",
+        "_cache_policy",
+        "_timeout",
+    )
+
+    _channels: str | list[str]
+    _triggers: list[str]
+    _tags: list[str]
+    _metadata: dict[str, Any]
+    _writes: list[ChannelWriteEntry]
+    _bound: Runnable
+    _retry_policy: list[RetryPolicy]
+    _cache_policy: CachePolicy | None
+    _timeout: TimeoutPolicy | None
+
+    def __init__(
+        self,
+    ) -> None:
+        self._channels = []
+        self._triggers = []
+        self._tags = []
+        self._metadata = {}
+        self._writes = []
+        self._bound = DEFAULT_BOUND
+        self._retry_policy = []
+        self._cache_policy = None
+        self._timeout = None
+
+    def subscribe_only(
+        self,
+        channel: str,
+    ) -> Self:
+        """Subscribe to a single channel."""
+        if not self._channels:
+            self._channels = channel
+        else:
+            raise ValueError(
+                "Cannot subscribe to single channels when other channels are already subscribed to"
+            )
+
+        self._triggers.append(channel)
+
+        return self
+
+    def subscribe_to(
+        self,
+        *channels: str,
+        read: bool = True,
+    ) -> Self:
+        """Add channels to subscribe to.
+
+        Node will be invoked when any of these channels are updated, with a dict of the
+        channel values as input.
+
+        Args:
+            channels: Channel name(s) to subscribe to
+            read: If `True`, the channels will be included in the input to the node.
+                Otherwise, they will trigger the node without being sent in input.
+
+        Returns:
+            Self for chaining
+        """
+        if isinstance(self._channels, str):
+            raise ValueError(
+                "Cannot subscribe to channels when subscribed to a single channel"
+            )
+        if read:
+            if not self._channels:
+                self._channels = list(channels)
+            else:
+                self._channels.extend(channels)
+
+        if isinstance(channels, str):
+            self._triggers.append(channels)
+        else:
+            self._triggers.extend(channels)
+
+        return self
+
+    def read_from(
+        self,
+        *channels: str,
+    ) -> Self:
+        """Adds the specified channels to read from, without subscribing to them."""
+        assert isinstance(self._channels, list), (
+            "Cannot read additional channels when subscribed to single channels"
+        )
+        self._channels.extend(channels)
+        return self
+
+    def do(
+        self,
+        node: RunnableLike,
+    ) -> Self:
+        """Adds the specified node."""
+        if self._bound is not DEFAULT_BOUND:
+            self._bound = RunnableSeq(
+                self._bound, coerce_to_runnable(node, name=None, trace=True)
+            )
+        else:
+            self._bound = coerce_to_runnable(node, name=None, trace=True)
+        return self
+
+    def write_to(
+        self,
+        *channels: str | ChannelWriteEntry,
+        **kwargs: _WriteValue,
+    ) -> Self:
+        """Add channel writes.
+
+        Args:
+            *channels: Channel names to write to.
+            **kwargs: Channel name and value mappings.
+
+        Returns:
+            Self for chaining
+        """
+        self._writes.extend(
+            ChannelWriteEntry(c) if isinstance(c, str) else c for c in channels
+        )
+        self._writes.extend(
+            ChannelWriteEntry(k, mapper=v)
+            if callable(v)
+            else ChannelWriteEntry(k, value=v)
+            for k, v in kwargs.items()
+        )
+
+        return self
+
+    def meta(self, *tags: str, **metadata: Any) -> Self:
+        """Add tags or metadata to the node."""
+        self._tags.extend(tags)
+        self._metadata.update(metadata)
+        return self
+
+    def add_retry_policies(self, *policies: RetryPolicy) -> Self:
+        """Adds retry policies to the node."""
+        self._retry_policy.extend(policies)
+        return self
+
+    def add_cache_policy(self, policy: CachePolicy) -> Self:
+        """Adds cache policies to the node."""
+        self._cache_policy = policy
+        return self
+
+    def set_timeout(self, timeout: float | timedelta | TimeoutPolicy | None) -> Self:
+        """Set the per-attempt timeout policy for this node."""
+        self._timeout = coerce_timeout_policy(timeout)
+        return self
+
+    def build(self) -> PregelNode:
+        """Builds the node."""
+        return PregelNode(
+            channels=self._channels,
+            triggers=self._triggers,
+            tags=self._tags,
+            metadata=self._metadata,
+            writers=[ChannelWrite(self._writes)],
+            bound=self._bound,
+            retry_policy=self._retry_policy,
+            cache_policy=self._cache_policy,
+            timeout=self._timeout,
+        )
+
+
+class RawPregel(Pregel):
+    """A `Pregel` built from hand-written nodes and channels.
+
+    Accepts `NodeBuilder` values in `nodes`, and builds them.
+    """
+
+    def __init__(self, *, nodes: dict[str, PregelNode | NodeBuilder], **kwargs: Any):
+        super().__init__(
+            nodes={
+                k: v.build() if isinstance(v, NodeBuilder) else v
+                for k, v in nodes.items()
+            },
+            **kwargs,
+        )

libs/langgraph/tests/test_channels.py

@@ -20,6 +20,7 @@ from langgraph.errors import EmptyChannelError, InvalidUpdateError
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
 from langgraph.graph.state import _get_channel
+from langgraph.pregel._checkpoint import delta_channels_to_snapshot
 from langgraph.types import Overwrite
 
 pytestmark = pytest.mark.anyio
@@ -342,6 +343,26 @@ def test_delta_channel_checkpoint_returns_missing() -> None:
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.parametrize("frequency", [None, 50, 1000])
+def test_delta_channel_snapshot_frequency(frequency: int | None) -> None:
+    spec = (
+        DeltaChannel(_messages_delta_reducer)
+        if frequency is None
+        else DeltaChannel(_messages_delta_reducer, snapshot_frequency=frequency)
+    )
+    expected = 100 if frequency is None else frequency
+    assert spec.snapshot_frequency == expected
+    channel = spec.from_checkpoint(MISSING)
+    channel.update([HumanMessage(content="hi", id="h1")])
+    assert channel.copy().snapshot_frequency == expected
+    for updates in (expected - 1, expected):
+        assert delta_channels_to_snapshot(
+            {"messages": channel},
+            {"messages": (updates, updates)},
+            {"messages": 1},
+        ) == ({"messages"} if updates == expected else set())
+
+
 def test_delta_channel_snapshot_version_based() -> None:
     """Snapshots fire when a channel accumulates `snapshot_frequency` updates.
 

libs/langgraph/tests/test_delta_channel_id_stability.py

@@ -19,8 +19,8 @@ from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
 from langgraph.graph import END, START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.types import Command, Durability, Overwrite
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -279,7 +279,7 @@ def _pregel_graph(seen: list[list[str | None]]) -> Any:
         .do(look)
         .write_to("messages")
     )
-    return Pregel(
+    return RawPregel(
         nodes={"look": node},
         channels={
             "messages": DeltaChannel(_messages_delta_reducer),

libs/langgraph/tests/test_delta_channel_run_input.py

@@ -7,8 +7,9 @@ from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
-from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.pregel import Pregel
 from langgraph.types import Durability
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -19,7 +20,7 @@ def _sorted_extend(current: list, writes: list) -> list:
 
 def _delta_input_graph() -> Pregel:
     node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log", "plain")
-    return Pregel(
+    return RawPregel(
         nodes={"n": node},
         channels={
             "log": DeltaChannel(_sorted_extend),

libs/langgraph/tests/test_delta_channel_seal_subscribers.py

@@ -3,7 +3,8 @@ from langgraph.checkpoint.memory import InMemorySaver
 
 from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
-from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.pregel import Pregel
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -17,7 +18,7 @@ def _extend(current: list, writes: list) -> list:
 def _graph(reads: list) -> Pregel:
     writer = NodeBuilder().subscribe_only("a").do(lambda _: [1]).write_to("d")
     reader = NodeBuilder().subscribe_only("d").do(lambda d: reads.append(list(d)))
-    return Pregel(
+    return RawPregel(
         nodes={"writer": writer, "reader": reader},
         channels={"a": LastValue(str), "d": DeltaChannel(_extend)},
         input_channels="a",

libs/langgraph/tests/test_delta_channel_update_overwrite.py

@@ -13,8 +13,9 @@ from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
-from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.pregel import Pregel
 from langgraph.types import Overwrite
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -36,7 +37,7 @@ def _extend(current: list, writes: list) -> list:
 
 def _delta_input_graph(saver: InMemorySaver) -> Pregel:
     node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log")
-    return Pregel(
+    return RawPregel(
         nodes={"n": node},
         channels={"log": DeltaChannel(_extend), "go": LastValue(int)},
         input_channels=["log", "go"],

libs/langgraph/tests/test_delta_channel_update_state.py

@@ -30,8 +30,8 @@ from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.types import Overwrite, StateSnapshot, StateUpdate
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -549,7 +549,7 @@ def _sorted_extend(current: list, writes: list) -> list:
 
 def _delta_input_graph(snapshot_frequency: int = 1000) -> Any:
     node = NodeBuilder().subscribe_only("go").do(lambda _: [2]).write_to("log", "plain")
-    return Pregel(
+    return RawPregel(
         nodes={"n": node},
         channels={
             "log": DeltaChannel(_sorted_extend, snapshot_frequency=snapshot_frequency),
@@ -649,7 +649,7 @@ def _messages_input_graph() -> Any:
         .do(lambda _: [HumanMessage("n", id="n")])
         .write_to("messages")
     )
-    return Pregel(
+    return RawPregel(
         nodes={"n": node},
         channels={
             "messages": DeltaChannel(_messages_delta_reducer),

libs/langgraph/tests/test_deprecation.py

@@ -13,13 +13,13 @@ from langgraph.errors import NodeInterrupt
 from langgraph.func import entrypoint, task
 from langgraph.graph import StateGraph
 from langgraph.graph.message import MessageGraph
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.types import GraphOutput, Interrupt, RetryPolicy
 from langgraph.warnings import (
     LangGraphDeprecatedSinceV05,
     LangGraphDeprecatedSinceV10,
     LangGraphDeprecatedSinceV11,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 
 class PlainState(TypedDict): ...
@@ -157,7 +157,7 @@ def test_config_type_deprecation_pregel(mocker: MockerFixture) -> None:
         LangGraphDeprecatedSinceV10,
         match="`config_type` is deprecated and will be removed. Please use `context_schema` instead.",
     ):
-        instance = Pregel(
+        instance = RawPregel(
             nodes={
                 "one": chain,
             },

libs/langgraph/tests/test_large_cases.py

@@ -39,7 +39,6 @@ from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.constants import END, START
 from langgraph.graph import StateGraph
 from langgraph.graph.message import MessagesState, add_messages
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.types import (
     Command,
     Durability,
@@ -62,6 +61,7 @@ from tests.messages import (
     _AnyIdHumanMessage,
     _AnyIdToolMessage,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 
 def test_invoke_two_processes_in_out_interrupt(
@@ -72,7 +72,7 @@ def test_invoke_two_processes_in_out_interrupt(
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("inbox")
     two = NodeBuilder().subscribe_only("inbox").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "inbox": LastValue(int),

libs/langgraph/tests/test_large_cases_async.py

@@ -39,7 +39,6 @@ from langgraph.constants import END, START
 from langgraph.graph import MessagesState
 from langgraph.graph.message import add_messages
 from langgraph.graph.state import StateGraph
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.types import PregelTask, Send, StateSnapshot, StreamWriter
 from tests.any_int import AnyInt
 from tests.any_str import AnyDict, AnyStr, UnsortedSequence
@@ -51,6 +50,7 @@ from tests.messages import (
     _AnyIdHumanMessage,
     _AnyIdToolMessage,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -61,7 +61,7 @@ async def test_invoke_two_processes_in_out_interrupt(
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("inbox")
     two = NodeBuilder().subscribe_only("inbox").do(add_one).write_to("output")
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "inbox": LastValue(int),

libs/langgraph/tests/test_pregel.py

@@ -66,10 +66,7 @@ from langgraph.errors import GraphRecursionError, InvalidUpdateError, ParentComm
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph
 from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
-from langgraph.pregel import (
-    NodeBuilder,
-    Pregel,
-)
+from langgraph.pregel import Pregel
 from langgraph.pregel._loop import PregelLoop, SyncPregelLoop
 from langgraph.pregel._runner import PregelRunner
 from langgraph.runtime import RunControl
@@ -95,6 +92,7 @@ from tests.messages import (
     _AnyIdHumanMessage,
     _AnyIdToolMessage,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 pytestmark = pytest.mark.anyio
 
@@ -257,7 +255,7 @@ def test_context_json_schema() -> None:
         x: int
         y: str = field(default="foo")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": chain,
         },
@@ -449,7 +447,7 @@ def test_invoke_single_process_in_out(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": chain,
         },
@@ -485,7 +483,7 @@ def test_invoke_single_process_in_write_kwargs(mocker: MockerFixture) -> None:
         .write_to("output", fixed=5, output_plus_one=lambda x: x + 1)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={
             "input": LastValue(int),
@@ -521,7 +519,7 @@ def test_invoke_single_process_in_out_dict(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={"input": LastValue(int), "output": LastValue(int)},
         input_channels="input",
@@ -546,7 +544,7 @@ def test_invoke_single_process_in_dict_out_dict(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={"input": LastValue(int), "output": LastValue(int)},
         input_channels=["input"],
@@ -572,7 +570,7 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("inbox")
     two = NodeBuilder().subscribe_only("inbox").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "inbox": LastValue(int),
@@ -677,7 +675,7 @@ def test_batch_two_processes_in_out() -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one_with_delay).write_to("one")
     two = NodeBuilder().subscribe_only("one").do(add_one_with_delay).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "one": LastValue(int),
@@ -709,7 +707,7 @@ def test_invoke_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = NodeBuilder().subscribe_only(str(i)).do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes=nodes,
         channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
         | {"input": LastValue(int), "output": LastValue(int)},
@@ -737,7 +735,7 @@ def test_batch_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = NodeBuilder().subscribe_only(str(i)).do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes=nodes,
         channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
         | {"input": LastValue(int), "output": LastValue(int)},
@@ -770,7 +768,7 @@ def test_invoke_two_processes_two_in_two_out_invalid(mocker: MockerFixture) -> N
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
     two = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={"output": LastValue(int), "input": LastValue(int)},
         input_channels="input",
@@ -803,7 +801,7 @@ def test_invoke_two_processes_two_in_two_out_valid(mocker: MockerFixture) -> Non
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
     two = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -844,7 +842,7 @@ def test_invoke_checkpoint_two(
         .do(raise_if_above_10)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one},
         channels={
             "total": BinaryOperatorAggregate(int, operator.add),
@@ -1451,7 +1449,7 @@ def test_invoke_checkpoint_three(
         .do(raise_if_above_10)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one},
         channels={
             "total": BinaryOperatorAggregate(int, operator.add),
@@ -1576,7 +1574,7 @@ def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture) -> None
         NodeBuilder().subscribe_only("inbox").do(add_10_each).write_to("output")
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": one,
             "chain_three": chain_three,
@@ -1607,7 +1605,7 @@ def test_invoke_join_then_call_other_pregel(
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x])
 
-    inner_app = Pregel(
+    inner_app = RawPregel(
         nodes={
             "one": NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
         },
@@ -1628,7 +1626,7 @@ def test_invoke_join_then_call_other_pregel(
     )
     chain_three = NodeBuilder().subscribe_only("outbox_one").do(sum).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": one,
             "two": two,
@@ -1669,7 +1667,7 @@ def test_invoke_two_processes_one_in_two_out(mocker: MockerFixture) -> None:
     )
     two = NodeBuilder().subscribe_only("between").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -1696,7 +1694,7 @@ def test_invoke_two_processes_no_out(mocker: MockerFixture) -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("between")
     two = NodeBuilder().subscribe_only("between").do(add_one)
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -1719,7 +1717,7 @@ def test_invoke_two_processes_no_in(mocker: MockerFixture) -> None:
     two = NodeBuilder().subscribe_only("between").do(add_one)
 
     with pytest.raises(TypeError):
-        Pregel(nodes={"one": one, "two": two})
+        RawPregel(nodes={"one": one, "two": two})
 
 
 def test_conditional_entrypoint_to_multiple_state_graph(
@@ -4128,7 +4126,7 @@ def test_xray_lance(snapshot: SnapshotAssertion):
 def test_channel_values(sync_checkpointer: BaseCheckpointSaver) -> None:
     config = {"configurable": {"thread_id": "1"}}
     chain = NodeBuilder().subscribe_only("input").write_to("output")
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": chain,
         },
@@ -4346,6 +4344,25 @@ def test_enum_node_names():
     assert graph.invoke({"foo": "hello"}) == {"foo": "hello", "bar": "hello!"}
 
 
+@pytest.mark.parametrize("stream_mode", ["tasks", "debug"])
+def test_failing_task_result_is_streamed_before_the_error(stream_mode: str) -> None:
+    class State(TypedDict):
+        x: str
+
+    def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    graph = StateGraph(State).add_node("failing", failing).add_edge(START, "failing")
+    errors = []
+    with pytest.raises(ValueError, match="boom"):
+        for chunk in graph.compile().stream({"x": ""}, stream_mode=stream_mode):
+            payload = chunk["payload"] if stream_mode == "debug" else chunk
+            if "error" in payload:
+                errors.append(payload["error"])
+
+    assert [type(error) for error in errors] == [ValueError]
+
+
 def test_debug_retry(sync_checkpointer: BaseCheckpointSaver):
     class State(TypedDict):
         messages: Annotated[list[str], operator.add]
@@ -9488,7 +9505,7 @@ class _IntVersionSaver(InMemorySaver):
 
 
 def _build_chain_after_a_delta_channel() -> Pregel:
-    return Pregel(
+    return RawPregel(
         nodes={
             "a": NodeBuilder().subscribe_only("inp").do(lambda _: ["a"]).write_to("d"),
             "b": NodeBuilder().subscribe_only("d").do(lambda _: "b").write_to("x"),
@@ -9533,7 +9550,7 @@ async def test_aupdate_state_after_an_exit_snapshot_infers_the_last_writer() ->
 
 
 def test_update_state_after_a_fork_seal_infers_the_subscriber_that_ran() -> None:
-    graph = Pregel(
+    graph = RawPregel(
         nodes={
             "a": NodeBuilder()
             .subscribe_only("inp")
@@ -9568,7 +9585,7 @@ def test_update_state_after_a_fork_seal_infers_the_subscriber_that_ran() -> None
 def test_update_state_after_a_fork_seal_infers_the_node_whose_read_it_advanced() -> (
     None
 ):
-    graph = Pregel(
+    graph = RawPregel(
         nodes={
             "a": NodeBuilder().subscribe_only("inp").do(lambda _: ["a"]).write_to("d"),
             "b": NodeBuilder().subscribe_only("d").do(lambda _: "b").write_to("out"),

libs/langgraph/tests/test_pregel_async.py

@@ -69,7 +69,6 @@ from langgraph.errors import (
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph
 from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.pregel._loop import AsyncPregelLoop, PregelLoop
 from langgraph.pregel._runner import PregelRunner
 from langgraph.runtime import RunControl
@@ -95,6 +94,7 @@ from tests.messages import (
     _AnyIdHumanMessage,
     _AnyIdToolMessage,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 logger = logging.getLogger(__name__)
 
@@ -1446,7 +1446,7 @@ async def test_invoke_single_process_in_out(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": chain,
         },
@@ -1479,7 +1479,7 @@ async def test_invoke_single_process_in_write_kwargs(mocker: MockerFixture) -> N
         .write_to("output", fixed=5, output_plus_one=lambda x: x + 1)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={
             "input": LastValue(int),
@@ -1515,7 +1515,7 @@ async def test_invoke_single_process_in_out_dict(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={"input": LastValue(int), "output": LastValue(int)},
         input_channels="input",
@@ -1540,7 +1540,7 @@ async def test_invoke_single_process_in_dict_out_dict(mocker: MockerFixture) ->
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": chain},
         channels={"input": LastValue(int), "output": LastValue(int)},
         input_channels=["input"],
@@ -1567,7 +1567,7 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("inbox")
     two = NodeBuilder().subscribe_only("inbox").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "inbox": LastValue(int),
@@ -1607,7 +1607,7 @@ async def test_batch_two_processes_in_out() -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one_with_delay).write_to("one")
     two = NodeBuilder().subscribe_only("one").do(add_one_with_delay).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "one": LastValue(int),
@@ -1639,7 +1639,7 @@ async def test_invoke_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = NodeBuilder().subscribe_only(str(i)).do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes=nodes,
         channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
         | {"input": LastValue(int), "output": LastValue(int)},
@@ -1668,7 +1668,7 @@ async def test_batch_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = NodeBuilder().subscribe_only(str(i)).do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes=nodes,
         channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
         | {"input": LastValue(int), "output": LastValue(int)},
@@ -1704,7 +1704,7 @@ async def test_invoke_two_processes_two_in_two_out_invalid(
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
     two = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={"output": LastValue(int), "input": LastValue(int)},
         input_channels="input",
@@ -1722,7 +1722,7 @@ async def test_invoke_two_processes_two_in_two_out_valid(mocker: MockerFixture)
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
     two = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -1763,7 +1763,7 @@ async def test_invoke_checkpoint(
         .do(raise_if_above_10)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one},
         channels={
             "total": BinaryOperatorAggregate(int, operator.add),
@@ -3641,7 +3641,7 @@ async def test_invoke_checkpoint_three(
         .do(raise_if_above_10)
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one},
         channels={
             "total": BinaryOperatorAggregate(int, operator.add),
@@ -3765,7 +3765,7 @@ async def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture) -
         NodeBuilder().subscribe_only("inbox").do(add_10_each).write_to("output")
     )
 
-    app = Pregel(
+    app = RawPregel(
         nodes={
             "one": one,
             "chain_three": chain_three,
@@ -3799,7 +3799,7 @@ async def test_invoke_two_processes_one_in_two_out(mocker: MockerFixture) -> Non
     )
     two = NodeBuilder().subscribe_only("between").do(add_one).write_to("output")
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -3823,7 +3823,7 @@ async def test_invoke_two_processes_no_out(mocker: MockerFixture) -> None:
     one = NodeBuilder().subscribe_only("input").do(add_one).write_to("between")
     two = NodeBuilder().subscribe_only("between").do(add_one)
 
-    app = Pregel(
+    app = RawPregel(
         nodes={"one": one, "two": two},
         channels={
             "input": LastValue(int),
@@ -5809,6 +5809,60 @@ async def test_store_injected_async(
     )  # still overwriting the same one
 
 
+@pytest.mark.parametrize("stream_mode", ["tasks", "debug"])
+async def test_failing_task_result_is_streamed_before_the_error(
+    stream_mode: str,
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    async def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    graph = StateGraph(State).add_node("failing", failing).add_edge(START, "failing")
+    errors = []
+    with pytest.raises(ValueError, match="boom"):
+        async for chunk in graph.compile().astream({"x": ""}, stream_mode=stream_mode):
+            payload = chunk["payload"] if stream_mode == "debug" else chunk
+            if "error" in payload:
+                errors.append(payload["error"])
+
+    assert [type(error) for error in errors] == [ValueError]
+
+
+@pytest.mark.parametrize(
+    ("parallel", "stream_mode"), [(False, "custom"), (True, "values")]
+)
+async def test_graph_error_handler_recovers_on_the_executor_path(
+    parallel: bool, stream_mode: str
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    async def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    async def handler(state: State) -> Command:
+        return Command(update={"x": "handled"}, goto=END)
+
+    builder = (
+        StateGraph(State)
+        .add_node("failing", failing, error_handler=handler)
+        .add_edge(START, "failing")
+    )
+    if parallel:
+        builder.add_node("sibling", lambda state: {}).add_edge(START, "sibling")
+
+    chunks = [
+        chunk
+        async for chunk in builder.compile().astream(
+            {"x": ""}, stream_mode=[stream_mode, "values"]
+        )
+    ]
+
+    assert [c[-1] for c in chunks if c[0] == "values"][-1] == {"x": "handled"}
+
+
 async def test_debug_retry(async_checkpointer: BaseCheckpointSaver):
     class State(TypedDict):
         messages: Annotated[list[str], operator.add]
@@ -6780,6 +6834,31 @@ async def test_multiple_updates() -> None:
     ]
 
 
+@NEEDS_CONTEXTVARS
+async def test_task_in_subgraph_node_returns_its_value_with_tools_mode() -> None:
+    class State(TypedDict):
+        out: int | None
+
+    @task
+    async def double(x: int) -> int:
+        return x * 2
+
+    async def node(state: State) -> State:
+        return {"out": await double(21)}
+
+    inner = StateGraph(State).add_node("node", node).add_edge(START, "node").compile()
+    graph = (
+        StateGraph(State).add_node("inner", inner).add_edge(START, "inner").compile()
+    )
+
+    chunks = [
+        chunk
+        async for chunk in graph.astream({"out": None}, stream_mode=["values", "tools"])
+    ]
+
+    assert chunks[-1] == ("values", {"out": 42})
+
+
 @NEEDS_CONTEXTVARS
 async def test_falsy_return_from_task(async_checkpointer: BaseCheckpointSaver) -> None:
     """Test with a falsy return from a task."""
@@ -7544,6 +7623,40 @@ async def test_entrypoint_from_async_generator() -> None:
             yield "b"
 
 
+@NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("timeout", [None, 30])
+@pytest.mark.parametrize("version", ["v1", "v2"])
+async def test_awaited_task_returns_its_value_under_astream_events(
+    version: Literal["v1", "v2"], timeout: int | None
+) -> None:
+    class State(TypedDict):
+        doubled: int
+        zeroed: int
+
+    @task(timeout=timeout)
+    async def double(x: int) -> int:
+        return x * 2
+
+    @task(timeout=timeout)
+    async def zero() -> int:
+        return 0
+
+    async def node(state: State) -> State:
+        return {"doubled": await double(21), "zeroed": await zero()}
+
+    graph = StateGraph(State).add_node("node", node).add_edge(START, "node").compile()
+
+    outputs = [
+        event["data"]["output"]
+        async for event in graph.astream_events(
+            {"doubled": 0, "zeroed": 1}, version=version
+        )
+        if event["event"] == "on_chain_end" and event["name"] == "node"
+    ]
+
+    assert outputs == [{"doubled": 42, "zeroed": 0}]
+
+
 @NEEDS_CONTEXTVARS
 async def test_named_tasks_functional() -> None:
     class Foo:

libs/langgraph/tests/test_pregel_construction.py

@@ -0,0 +1,56 @@
+"""`Pregel` is the type of compiled graphs, built only by `StateGraph` and
+`@entrypoint`."""
+
+import pytest
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.memory import InMemorySaver
+from typing_extensions import TypedDict
+
+import langgraph.pregel
+from langgraph.channels.ephemeral_value import EphemeralValue
+from langgraph.func import entrypoint
+from langgraph.graph import START, StateGraph
+from langgraph.pregel import Pregel
+from langgraph.pregel._read import PregelNode
+
+
+def test_pregel_cannot_be_constructed_directly() -> None:
+    with pytest.raises(TypeError, match="`Pregel` cannot be constructed directly"):
+        Pregel(
+            nodes={"node": PregelNode(channels="a", triggers=["a"])},
+            channels={"a": EphemeralValue(str)},
+            input_channels="a",
+            output_channels="a",
+        )
+
+
+def test_node_builder_is_not_exported() -> None:
+    assert langgraph.pregel.__all__ == ("Pregel",)
+    assert not hasattr(langgraph.pregel, "NodeBuilder")
+
+
+def test_compiled_graphs_are_pregel_instances() -> None:
+    class State(TypedDict):
+        value: int
+
+    def increment(state: State) -> State:
+        return {"value": state["value"] + 1}
+
+    builder = StateGraph(State)
+    builder.add_node("increment", increment)
+    builder.add_edge(START, "increment")
+    graph = builder.compile(checkpointer=InMemorySaver())
+
+    @entrypoint(checkpointer=InMemorySaver())
+    def workflow(value: int) -> int:
+        return value + 1
+
+    assert isinstance(graph, Pregel)
+    assert isinstance(workflow, Pregel)
+
+    # `with_config` copies the graph by calling its constructor again.
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    assert graph.with_config(recursion_limit=5).invoke({"value": 1}, config) == {
+        "value": 2
+    }
+    assert workflow.with_config(recursion_limit=5).invoke(1, config) == 2

libs/langgraph/tests/test_retry.py

@@ -47,7 +47,6 @@ from langgraph.errors import (
 )
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph, add_messages
-from langgraph.pregel import NodeBuilder, Pregel
 from langgraph.pregel._read import PregelNode
 from langgraph.pregel._retry import (
     _checkpoint_ns_for_parent_command,
@@ -67,6 +66,7 @@ from langgraph.types import (
     TimeoutPolicy,
     interrupt,
 )
+from tests.raw_pregel import NodeBuilder, RawPregel
 
 NEEDS_CONTEXTVARS = pytest.mark.skipif(
     sys.version_info < (3, 11),
@@ -1260,7 +1260,7 @@ def test_pregel_validate_rejects_sync_writer_timeout():
         return value
 
     with pytest.raises(ValueError, match="only supported for async nodes"):
-        Pregel(
+        RawPregel(
             nodes={
                 "slow": PregelNode(
                     channels="input",
@@ -1284,7 +1284,7 @@ def test_pregel_validate_rejects_wrapped_sync_runnable_lambda_timeout():
         return value + 1
 
     with pytest.raises(ValueError, match="only supported for async nodes"):
-        Pregel(
+        RawPregel(
             nodes={
                 "slow": (
                     NodeBuilder()
@@ -1307,7 +1307,7 @@ def test_pregel_validate_accepts_wrapped_async_runnable_lambda_timeout():
     async def slow(value: int) -> int:
         return value + 1
 
-    Pregel(
+    RawPregel(
         nodes={
             "slow": (
                 NodeBuilder()
@@ -1334,7 +1334,7 @@ def test_pregel_validate_rejects_parallel_sync_branch_timeout():
         return value + 1
 
     with pytest.raises(ValueError, match="only supported for async nodes"):
-        Pregel(
+        RawPregel(
             nodes={
                 "parallel": (
                     NodeBuilder()
@@ -1363,7 +1363,7 @@ def test_pregel_validate_rejects_sync_node_timeout():
         return value + 1
 
     with pytest.raises(ValueError, match="only supported for async nodes"):
-        Pregel(
+        RawPregel(
             nodes={
                 "slow": (
                     NodeBuilder()
@@ -1388,7 +1388,7 @@ async def test_pregel_validate_accepts_async_runnable_lambda_timeout():
         await asyncio.sleep(0.2)
         return value + 1
 
-    graph = Pregel(
+    graph = RawPregel(
         nodes={
             "slow": (
                 NodeBuilder()
@@ -1418,7 +1418,7 @@ async def test_pregel_validate_accepts_runnable_callable_with_sync_and_async_tim
         await asyncio.sleep(0.2)
         return value + 1
 
-    graph = Pregel(
+    graph = RawPregel(
         nodes={
             "slow": (
                 NodeBuilder()
@@ -1632,7 +1632,7 @@ async def test_node_builder_timeout_e2e():
         await asyncio.sleep(0.2)
         return value + 1
 
-    graph = Pregel(
+    graph = RawPregel(
         nodes={
             "slow": (
                 NodeBuilder()
@@ -2097,6 +2097,47 @@ def test_graph_error_handler_can_route_with_command():
     assert result["foo"] == "handled_next"
 
 
+# Each case runs the failing node on the executor path instead of the
+# single-task fast path.
+@pytest.mark.parametrize(
+    ("parallel", "stream_mode", "subgraphs"),
+    [
+        (False, "custom", False),
+        (False, "messages", False),
+        (False, "values", True),
+        (True, "values", False),
+    ],
+)
+def test_graph_error_handler_recovers_on_the_executor_path(
+    parallel: bool, stream_mode: str, subgraphs: bool
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    def handler(state: State) -> Command:
+        return Command(update={"x": "handled"}, goto=END)
+
+    builder = (
+        StateGraph(State)
+        .add_node("failing", failing, error_handler=handler)
+        .add_edge(START, "failing")
+    )
+    if parallel:
+        builder.add_node("sibling", lambda state: {}).add_edge(START, "sibling")
+    graph = builder.compile()
+
+    chunks = list(
+        graph.stream(
+            {"x": ""}, stream_mode=[stream_mode, "values"], subgraphs=subgraphs
+        )
+    )
+
+    assert [c[-1] for c in chunks if c[-2] == "values"][-1] == {"x": "handled"}
+
+
 def test_graph_error_handler_failure_fails_run():
     class State(TypedDict):
         foo: str

libs/langgraph/tests/test_tool_stream_handler.py

@@ -13,13 +13,17 @@ from typing import Annotated, Any
 import pytest
 from langchain_core.messages import AIMessage
 from langchain_core.tools import tool
+from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.prebuilt import ToolNode, ToolRuntime
 from typing_extensions import TypedDict
 
 from langgraph.constants import END, START
+from langgraph.errors import GraphDrained
 from langgraph.graph import StateGraph
 from langgraph.graph.message import add_messages
 from langgraph.pregel._tools import _tool_call_writer
+from langgraph.runtime import RunControl
+from langgraph.types import Command, interrupt
 
 
 class _State(TypedDict):
@@ -54,14 +58,14 @@ def _caller_async(tool_name: str, tool_args: dict[str, Any], tc_id: str = "tc1")
     return caller
 
 
-def _build_graph(caller, tools) -> Any:
+def _build_graph(caller, tools, checkpointer=None) -> Any:
     sg = StateGraph(_State)
     sg.add_node("caller", caller)
     sg.add_node("tools", ToolNode(tools))
     sg.add_edge(START, "caller")
     sg.add_edge("caller", "tools")
     sg.add_edge("tools", END)
-    return sg.compile()
+    return sg.compile(checkpointer=checkpointer)
 
 
 def _tool_events(stream) -> list[tuple[tuple[str, ...], dict]]:
@@ -146,6 +150,105 @@ class TestSyncGraphSyncTool:
         assert kinds == ["tool-started", "tool-error"]
         assert events[1][1]["message"] == "nope"
 
+    def test_interrupt_in_tool_is_not_a_tool_error(self) -> None:
+        @tool
+        def ask() -> str:
+            """asks."""
+            return interrupt("ok?")
+
+        graph = _build_graph(_caller_sync("ask", {}), [ask], InMemorySaver())
+        config = {"configurable": {"thread_id": "1"}}
+        paused = _tool_events(
+            graph.stream(
+                {"messages": []}, config, stream_mode=["tools"], subgraphs=True
+            )
+        )
+        resumed = _tool_events(
+            graph.stream(
+                Command(resume="yes"), config, stream_mode=["tools"], subgraphs=True
+            )
+        )
+
+        assert [p["event"] for _, p in paused] == ["tool-started"]
+        assert [(p["event"], p["tool_call_id"]) for _, p in resumed] == [
+            ("tool-started", "tc1"),
+            ("tool-finished", "tc1"),
+        ]
+
+    def test_tool_handing_off_to_the_parent_still_ends_its_call(self) -> None:
+        class State(TypedDict):
+            messages: Annotated[list, add_messages]
+            done: bool
+
+        def hand_off(state: State) -> Command:
+            return Command(graph=Command.PARENT, goto="finish", update={"done": True})
+
+        inner = StateGraph(State).add_node("hand_off", hand_off)
+        inner = inner.add_edge(START, "hand_off").compile()
+
+        @tool
+        def handoff() -> str:
+            """hands off."""
+            inner.invoke({"messages": [], "done": False})
+            return "unreachable"
+
+        graph = (
+            StateGraph(State)
+            .add_node("caller", _caller_sync("handoff", {}))
+            .add_node("tools", ToolNode([handoff]))
+            .add_node("finish", lambda state: {"done": True})
+            .add_edge(START, "caller")
+            .add_edge("caller", "tools")
+            .add_edge("tools", END)
+            .add_edge("finish", END)
+            .compile()
+        )
+        events = [
+            payload["event"]
+            for _, payload in _tool_events(
+                graph.stream(
+                    {"messages": [], "done": False},
+                    stream_mode=["tools"],
+                    subgraphs=True,
+                )
+            )
+        ]
+
+        assert events[0] == "tool-started"
+        assert events[-1] == "tool-error", events
+
+    def test_drained_tool_is_not_a_tool_error(self) -> None:
+        control = RunControl()
+
+        def request_drain(state: _State) -> dict:
+            control.request_drain()
+            return {}
+
+        inner = (
+            StateGraph(_State)
+            .add_node("request_drain", request_drain)
+            .add_node("after", lambda state: {})
+            .add_edge(START, "request_drain")
+            .add_edge("request_drain", "after")
+            .compile()
+        )
+
+        @tool
+        def drained() -> str:
+            """drains."""
+            inner.invoke({"messages": []}, control=control)
+            return "unreachable"
+
+        graph = _build_graph(_caller_sync("drained", {}), [drained])
+        chunks: list = []
+        with pytest.raises(GraphDrained):
+            for chunk in graph.stream(
+                {"messages": []}, stream_mode=["tools"], subgraphs=True
+            ):
+                chunks.append(chunk)
+
+        assert [p["event"] for _, p in _tool_events(chunks)] == ["tool-started"]
+
     def test_writer_unset_outside_tool(self) -> None:
         # Outside any tool body the ContextVar that ToolRuntime reads
         # is unset — emitting from there would be a no-op.
@@ -196,6 +299,25 @@ class TestAsyncGraphAsyncTool:
         assert kinds == ["tool-started", "tool-output-delta", "tool-finished"]
         assert events[1][1]["delta"] == "hi"
 
+    @pytest.mark.anyio
+    async def test_interrupt_in_async_tool_is_not_a_tool_error(self) -> None:
+        @tool
+        async def aask() -> str:
+            """asks."""
+            return interrupt("ok?")
+
+        graph = _build_graph(_caller_async("aask", {}), [aask], InMemorySaver())
+        config = {"configurable": {"thread_id": "1"}}
+        events = []
+        for graph_input in ({"messages": []}, Command(resume="yes")):
+            async for ns, mode, payload in graph.astream(
+                graph_input, config, stream_mode=["tools"], subgraphs=True
+            ):
+                if mode == "tools":
+                    events.append(payload["event"])
+
+        assert events == ["tool-started", "tool-started", "tool-finished"]
+
 
 class TestConcurrentToolCalls:
     def test_parallel_tool_calls_do_not_bleed(self) -> None:

libs/langgraph/uv.lock

@@ -1439,7 +1439,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.17"
+version = "1.2.18"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1625,7 +1625,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.17"
+version = "1.2.18"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -370,7 +370,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/__init__.py

@@ -5,7 +5,7 @@ from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 from langgraph_sdk.priority import Priority
 from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
 
-__version__ = "0.4.9"
+__version__ = "0.4.10"
 
 __all__ = [
     "Auth",

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

@@ -217,6 +217,12 @@ class HttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
+            if last_event_id is not None:
+                current_headers = {
+                    key: value
+                    for key, value in current_headers.items()
+                    if key.lower() != "last-event-id"
+                }
             if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 

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

@@ -219,6 +219,12 @@ class SyncHttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
+            if last_event_id is not None:
+                current_headers = {
+                    key: value
+                    for key, value in current_headers.items()
+                    if key.lower() != "last-event-id"
+                }
             if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 

libs/sdk-py/langgraph_sdk/sse.py

@@ -104,7 +104,7 @@ class SSEDecoder:
             if (
                 not self._event
                 and not self._data
-                and not self._last_event_id
+                and self._last_event_id is None
                 and self._retry is None
             ):
                 return None

libs/sdk-py/tests/test_client_stream.py

@@ -312,18 +312,14 @@ async def test_http_client_stream_recovers_after_disconnect():
     ]
 
 
-def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
+def _drop_after(chunks: list[bytes], stream_cls: type, reconnects: list[httpx.Request]):
     def handler(request: httpx.Request) -> httpx.Response:
         if request.method == "POST":
             return httpx.Response(
                 200,
                 headers={"Content-Type": "text/event-stream", "Location": "/reconnect"},
                 stream=stream_cls(
-                    [
-                        b"id: 1\nevent: values\ndata: 1\n\n",
-                        b"id:\nevent: values\ndata: 2\n\n",
-                    ],
-                    httpx.RemoteProtocolError("incomplete chunked read"),
+                    chunks, httpx.RemoteProtocolError("incomplete chunked read")
                 ),
             )
         reconnects.append(request)
@@ -336,26 +332,99 @@ def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
     return handler
 
 
-def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
+def _sync_reconnect(
+    chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
     reconnects: list[httpx.Request] = []
-    transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream, reconnects))
+    transport = httpx.MockTransport(_drop_after(chunks, ListByteStream, reconnects))
     with httpx.Client(transport=transport, base_url="https://example.com") as client:
-        list(SyncHttpClient(client).stream("/stream", "POST", json={}))
-
-    assert "last-event-id" not in reconnects[0].headers
+        list(SyncHttpClient(client).stream("/stream", "POST", json={}, headers=headers))
+    return reconnects[0]
 
 
-async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
+async def _async_reconnect(
+    chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
     reconnects: list[httpx.Request] = []
     transport = httpx.MockTransport(
-        _drop_after_an_empty_id(AsyncListByteStream, reconnects)
+        _drop_after(chunks, AsyncListByteStream, reconnects)
     )
     async with httpx.AsyncClient(
         transport=transport, base_url="https://example.com"
     ) as client:
-        [part async for part in HttpClient(client).stream("/stream", "POST", json={})]
+        [
+            part
+            async for part in HttpClient(client).stream(
+                "/stream", "POST", json={}, headers=headers
+            )
+        ]
+    return reconnects[0]
 
-    assert "last-event-id" not in reconnects[0].headers
+
+_EMPTY_IDS = pytest.mark.parametrize(
+    "empty_id",
+    [b"id:\nevent: values\ndata: 2\n\n", b"id:\n\n"],
+    ids=["with-an-event", "alone"],
+)
+_CALLER_CURSORS = pytest.mark.parametrize(
+    "headers",
+    [None, {"Last-Event-ID": "5"}, {"last-event-id": "5"}],
+    ids=["no-caller-cursor", "caller-cursor", "lowercase-caller-cursor"],
+)
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    reconnect = _sync_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+    )
+
+    assert "last-event-id" not in reconnect.headers
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    reconnect = await _async_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+    )
+
+    assert "last-event-id" not in reconnect.headers
+
+
+def test_sync_stream_id_replaces_a_lowercase_caller_cursor():
+    reconnect = _sync_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
+async def test_async_stream_id_replaces_a_lowercase_caller_cursor():
+    reconnect = await _async_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
+def test_sync_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+    reconnect = _sync_reconnect([b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"})
+
+    assert reconnect.headers.get_list("last-event-id") == ["5"]
+
+
+async def test_async_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+    reconnect = await _async_reconnect(
+        [b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["5"]
 
 
 # --- _sse_to_v2_dict conversion ---

libs/sdk-py/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.17"
+version = "1.2.18"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -385,7 +385,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.0"
+version = "4.3.1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },