docs(langgraph): say which task paths graph_namespace counts as calls

A node that streams a graph with `stream_mode="messages"` got that graph's
model tokens but not the messages its nodes return, such as a `ToolNode`'s
`ToolMessage`s: the node-output check had no exception for the handler's own
graph, unlike the model, tool and `push_message` checks. Streaming the same
graph at the root returned both.

The task tests now run from an entrypoint and from a task inside a task.

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/graph/message.py

@@ -22,7 +22,7 @@ from langchain_core.messages import (
 )
 from typing_extensions import TypedDict, deprecated
 
-from langgraph._internal._constants import CONF, CONFIG_KEY_SEND, NS_SEP
+from langgraph._internal._constants import CONF, CONFIG_KEY_SEND
 from langgraph.graph.state import StateGraph
 from langgraph.pregel._messages import (
     _messages_delta_reducer as _messages_delta_reducer,
@@ -343,7 +343,7 @@ def push_message(
     )
 
     from langgraph.config import get_config
-    from langgraph.pregel._messages import StreamMessagesHandler
+    from langgraph.pregel._messages import StreamMessagesHandler, graph_namespace
 
     config = get_config()
     message = next(x for x in convert_to_messages([message]))
@@ -359,15 +359,12 @@ def push_message(
     ):
         handlers = config["callbacks"]
 
-    if stream_handler := next(
-        (x for x in handlers if isinstance(x, StreamMessagesHandler)), None
-    ):
-        metadata = config["metadata"]
-        message_meta = (
-            tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP)),
-            metadata,
-        )
-        stream_handler._emit(message_meta, message, dedupe=False)
+    metadata = config["metadata"]
+    for handler in handlers:
+        if isinstance(handler, StreamMessagesHandler):
+            ns = graph_namespace(metadata)
+            if handler.subgraphs or not ns or ns == handler.parent_ns:
+                handler._emit((ns, metadata), message, dedupe=False)
 
     if state_key:
         config[CONF][CONFIG_KEY_SEND]([(state_key, message)])

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/_messages.py

@@ -23,7 +23,7 @@ from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResul
 from pydantic import BaseModel
 
 from langgraph._internal._config import filter_to_user_tags
-from langgraph._internal._constants import NS_SEP
+from langgraph._internal._constants import NS_SEP, PUSH
 from langgraph.channels.binop import _get_overwrite
 from langgraph.channels.delta import DeltaChannel
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
@@ -55,6 +55,23 @@ def _state_values(obj: Any) -> Sequence[Any]:
     return ()
 
 
+def graph_namespace(metadata: Mapping[str, Any]) -> tuple[str, ...]:
+    """Namespace of the graph a task runs in, from the task's run metadata.
+
+    Drops the task's own segment, plus one per enclosing `@task` call: a call
+    runs in its caller's graph but nests its segment under the caller's. A
+    call's `langgraph_path` is `(PUSH, <caller's path>, <index>, True)`, while
+    a `Send` task's ends in `False`.
+    """
+    ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))
+    path = metadata.get("langgraph_path", ())
+    drop = 1
+    while len(path) == 4 and path[0] == PUSH and path[-1] is True:
+        drop += 1
+        path = path[1]
+    return ns[:-drop]
+
+
 class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
     """A callback handler that implements stream_mode=messages.
 
@@ -148,9 +165,7 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         **kwargs: Any,
     ) -> Any:
         if metadata and (not tags or (TAG_NOSTREAM not in tags)):
-            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
-                :-1
-            ]
+            ns = graph_namespace(metadata)
             if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
                 return
             if (filtered_tags := filter_to_user_tags(tags)) is not None:
@@ -213,10 +228,8 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             and kwargs.get("name") == metadata.get("langgraph_node")
             and (not tags or TAG_HIDDEN not in tags)
         ):
-            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
-                :-1
-            ]
-            if not self.subgraphs and len(ns) > 0:
+            ns = graph_namespace(metadata)
+            if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
                 return
             self.metadata[run_id] = (ns, metadata)
             for value in _state_values(inputs):

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

@@ -7,8 +7,9 @@ from uuid import UUID
 
 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._messages import graph_namespace
 from langgraph.pregel.protocol import StreamChunk
 
 try:
@@ -91,12 +92,9 @@ class StreamToolCallHandler(BaseCallbackHandler, _StreamingCallbackHandler):
     ) -> tuple[str, ...] | None:
         """Resolve the namespace this tool call should emit at, or `None` to skip.
 
-        Mirrors `StreamMessagesHandler.on_chat_model_start`'s namespace
-        derivation: parses `langgraph_checkpoint_ns` (which ends with
-        the `node_name:task_id` of the calling node), drops that
-        trailing segment, and returns the containing subgraph's own
-        namespace. Returns `None` when the call should be silently
-        suppressed:
+        Mirrors `StreamMessagesHandler.on_chat_model_start`: the namespace
+        is that of the graph the tool runs in (`graph_namespace`). Returns
+        `None` when the call should be silently suppressed:
 
         - `metadata` is missing — handler is attached to a context
           without Pregel routing info.
@@ -109,11 +107,9 @@ class StreamToolCallHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             return None
         if tags and TAG_NOSTREAM in tags:
             return None
-        nskey = metadata.get("langgraph_checkpoint_ns")
-        if not nskey:
-            ns: tuple[str, ...] = ()
-        else:
-            ns = tuple(cast(str, nskey).split(NS_SEP))[:-1]
+        ns = (
+            graph_namespace(metadata) if metadata.get("langgraph_checkpoint_ns") else ()
+        )
         if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
             return None
         return ns
@@ -188,6 +184,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/main.py

@@ -2880,6 +2880,20 @@ class Pregel(
                     for h in callback_manager.inheritable_handlers
                     if not isinstance(h, StreamMessagesHandlerV2)
                 ]
+            if CONFIG_KEY_TASK_ID not in config[CONF]:
+                # Not run as a task (a child invoked with its own thread, say),
+                # this graph's namespace starts at the root, so a caller's
+                # handler would take this graph's events for its own graph's.
+                callback_manager.handlers = [
+                    h
+                    for h in callback_manager.handlers
+                    if not _streams_only_its_graph(h)
+                ]
+                callback_manager.inheritable_handlers = [
+                    h
+                    for h in callback_manager.inheritable_handlers
+                    if not _streams_only_its_graph(h)
+                ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
             run_manager = callback_manager.on_chain_start(
@@ -2898,9 +2912,9 @@ class Pregel(
                 )
             # set up subgraph checkpointing
             config = self._own_checkpoint_config(config)
+            stream_handlers: list[StreamMessagesHandler | StreamToolCallHandler] = []
             # set up messages stream mode
             if "messages" in stream_modes:
-                ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
                 use_stream_messages_v2 = bool(
                     version == "v2" and config[CONF].get(CONFIG_KEY_STREAM_MESSAGES_V2)
                 )
@@ -2909,24 +2923,12 @@ class Pregel(
                     if use_stream_messages_v2
                     else StreamMessagesHandler
                 )
-                run_manager.inheritable_handlers.append(
-                    messages_handler_cls(
-                        stream.put,
-                        subgraphs,
-                        parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
-                    )
-                )
+                stream_handlers.append(messages_handler_cls(stream.put, subgraphs))
 
             # set up tools stream mode
             if "tools" in stream_modes:
-                ns_tools = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
-                run_manager.inheritable_handlers.append(
-                    StreamToolCallHandler(
-                        stream.put,
-                        subgraphs,
-                        parent_ns=tuple(ns_tools.split(NS_SEP)) if ns_tools else None,
-                    )
-                )
+                stream_handlers.append(StreamToolCallHandler(stream.put, subgraphs))
+            run_manager.inheritable_handlers.extend(stream_handlers)
 
             # set up custom stream mode
             if "custom" in stream_modes:
@@ -3009,6 +3011,10 @@ class Pregel(
                 cache_policy=self.cache_policy,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
+                # Known only now: the loop numbers the namespace of every
+                # subgraph a task runs after its first.
+                for handler in stream_handlers:
+                    handler.parent_ns = loop.checkpoint_ns
                 emit_graph_lifecycle_events(loop)
                 # create runner
                 runner = PregelRunner(
@@ -3291,6 +3297,20 @@ class Pregel(
                     for h in callback_manager.inheritable_handlers
                     if not isinstance(h, StreamMessagesHandlerV2)
                 ]
+            if CONFIG_KEY_TASK_ID not in config[CONF]:
+                # Not run as a task (a child invoked with its own thread, say),
+                # this graph's namespace starts at the root, so a caller's
+                # handler would take this graph's events for its own graph's.
+                callback_manager.handlers = [
+                    h
+                    for h in callback_manager.handlers
+                    if not _streams_only_its_graph(h)
+                ]
+                callback_manager.inheritable_handlers = [
+                    h
+                    for h in callback_manager.inheritable_handlers
+                    if not _streams_only_its_graph(h)
+                ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
             run_manager = await callback_manager.on_chain_start(
@@ -3310,7 +3330,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,
                 )
@@ -3323,10 +3345,9 @@ class Pregel(
                 )
             # set up subgraph checkpointing
             config = self._own_checkpoint_config(config)
+            stream_handlers: list[StreamMessagesHandler | StreamToolCallHandler] = []
             # set up messages stream mode
             if "messages" in stream_modes:
-                # namespace can be None in a root level graph?
-                ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
                 use_stream_messages_v2 = bool(
                     version == "v2" and config[CONF].get(CONFIG_KEY_STREAM_MESSAGES_V2)
                 )
@@ -3335,24 +3356,12 @@ class Pregel(
                     if use_stream_messages_v2
                     else StreamMessagesHandler
                 )
-                run_manager.inheritable_handlers.append(
-                    messages_handler_cls(
-                        stream_put,
-                        subgraphs,
-                        parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
-                    )
-                )
+                stream_handlers.append(messages_handler_cls(stream_put, subgraphs))
 
             # set up tools stream mode
             if "tools" in stream_modes:
-                ns_tools = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
-                run_manager.inheritable_handlers.append(
-                    StreamToolCallHandler(
-                        stream_put,
-                        subgraphs,
-                        parent_ns=tuple(ns_tools.split(NS_SEP)) if ns_tools else None,
-                    )
-                )
+                stream_handlers.append(StreamToolCallHandler(stream_put, subgraphs))
+            run_manager.inheritable_handlers.extend(stream_handlers)
 
             # set up custom stream mode
             def stream_writer(c: Any) -> None:
@@ -3461,6 +3470,10 @@ class Pregel(
                 cache_policy=self.cache_policy,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
+                # Known only now: the loop numbers the namespace of every
+                # subgraph a task runs after its first.
+                for handler in stream_handlers:
+                    handler.parent_ns = loop.checkpoint_ns
                 await aemit_graph_lifecycle_events(loop)
                 # create runner
                 runner = PregelRunner(
@@ -4458,6 +4471,13 @@ def _coerce_checkpoint_values(payload: Any, mapper: Callable[[Any], Any]) -> Non
         payload["values"] = mapper(payload["values"])
 
 
+def _streams_only_its_graph(handler: object) -> bool:
+    return (
+        isinstance(handler, (StreamMessagesHandler, StreamToolCallHandler))
+        and not handler.subgraphs
+    )
+
+
 def _resolve_parent_ns(
     graph_config: RunnableConfig | None, call_config: RunnableConfig | None
 ) -> tuple[str, ...]:

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/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_pregel.py

@@ -4346,6 +4346,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]

libs/langgraph/tests/test_pregel_async.py

@@ -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_retry.py

@@ -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_stream_namespaces.py

@@ -0,0 +1,291 @@
+from typing import Any
+
+import pytest
+from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+from langchain_core.messages import AIMessage
+from langchain_core.tools import tool
+
+from langgraph.func import entrypoint, task
+from langgraph.graph import START, MessagesState, StateGraph
+from langgraph.graph.message import push_message
+from langgraph.graph.state import CompiledStateGraph
+
+pytestmark = pytest.mark.anyio
+
+
+@tool
+def echo(text: str) -> str:
+    """Echo the text."""
+    return text
+
+
+def _one_node_graph(node: Any) -> CompiledStateGraph:
+    return (
+        StateGraph(MessagesState)
+        .add_node(node)
+        .add_edge(START, node.__name__)
+        .compile()
+    )
+
+
+def _chat_and_echo(state: MessagesState) -> dict:
+    echo.invoke({"text": "done"})
+    return {"messages": [GenericFakeChatModel(messages=iter(["one two"])).invoke("hi")]}
+
+
+async def _achat_and_echo(state: MessagesState) -> dict:
+    await echo.ainvoke({"text": "done"})
+    model = GenericFakeChatModel(messages=iter(["one two"]))
+    return {"messages": [await model.ainvoke("hi")]}
+
+
+def _segment_names(ns: tuple[str, ...]) -> tuple[str, ...]:
+    return tuple(segment.split(":")[0] for segment in ns)
+
+
+def _summary(part: dict) -> tuple[tuple[str, ...], str]:
+    if part["type"] == "messages":
+        return _segment_names(part["ns"]), part["data"][0].content
+    return _segment_names(part["ns"]), part["data"]["event"]
+
+
+_CHAT_AND_ECHO_EVENTS = [
+    ((), "tool-started"),
+    ((), "tool-finished"),
+    ((), "one"),
+    ((), " "),
+    ((), "two"),
+]
+
+_CHILD_CONFIGS = pytest.mark.parametrize(
+    "child_config",
+    [
+        None,
+        {"configurable": {"thread_id": "child"}},
+        {"configurable": {"thread_id": "parent"}},
+    ],
+    ids=["no_config", "own_thread", "caller_thread"],
+)
+
+
+_TASK_CALLERS = pytest.mark.parametrize(
+    "nested", [False, True], ids=["from_entrypoint", "from_task_in_task"]
+)
+
+
+@_TASK_CALLERS
+@pytest.mark.parametrize("subgraphs", [False, True])
+def test_task_events_stream_under_the_graph_namespace(
+    nested: bool, subgraphs: bool
+) -> None:
+    @task
+    def chat_and_echo() -> None:
+        _chat_and_echo({"messages": []})
+
+    @task
+    def outer() -> None:
+        chat_and_echo().result()
+
+    @entrypoint()
+    def workflow(inputs: dict) -> None:
+        (outer() if nested else chat_and_echo()).result()
+
+    parts = workflow.stream(
+        {}, stream_mode=["messages", "tools"], subgraphs=subgraphs, version="v2"
+    )
+
+    assert [_summary(part) for part in parts] == _CHAT_AND_ECHO_EVENTS
+
+
+@_TASK_CALLERS
+@pytest.mark.parametrize("subgraphs", [False, True])
+async def test_task_events_stream_under_the_graph_namespace_async(
+    nested: bool, subgraphs: bool
+) -> None:
+    @task
+    async def chat_and_echo() -> None:
+        await _achat_and_echo({"messages": []})
+
+    @task
+    async def outer() -> None:
+        await chat_and_echo()
+
+    @entrypoint()
+    async def workflow(inputs: dict) -> None:
+        await (outer() if nested else chat_and_echo())
+
+    parts = workflow.astream(
+        {}, stream_mode=["messages", "tools"], subgraphs=subgraphs, version="v2"
+    )
+
+    assert [_summary(part) async for part in parts] == _CHAT_AND_ECHO_EVENTS
+
+
+def test_task_events_stream_under_the_namespace_of_their_subgraph() -> None:
+    @task
+    def chat_and_echo() -> None:
+        _chat_and_echo({"messages": []})
+
+    def node(state: MessagesState) -> dict:
+        chat_and_echo().result()
+        return {}
+
+    sub = _one_node_graph(node)
+    parent = StateGraph(MessagesState).add_node("sub", sub).add_edge(START, "sub")
+
+    parts = parent.compile().stream(
+        {"messages": []},
+        stream_mode=["messages", "tools"],
+        subgraphs=True,
+        version="v2",
+    )
+
+    assert [_summary(part) for part in parts] == [
+        (("sub",), event) for _, event in _CHAT_AND_ECHO_EVENTS
+    ]
+
+
+@_CHILD_CONFIGS
+def test_child_graph_streams_only_to_the_node_streaming_it(
+    child_config: dict | None,
+) -> None:
+    child = _one_node_graph(_chat_and_echo)
+    received = []
+
+    def node(state: MessagesState) -> dict:
+        received.extend(
+            child.stream(
+                {"messages": []}, child_config, stream_mode=["messages", "tools"]
+            )
+        )
+        return {}
+
+    parts = _one_node_graph(node).stream(
+        {"messages": []},
+        {"configurable": {"thread_id": "parent"}},
+        stream_mode=["messages", "tools"],
+    )
+
+    assert list(parts) == []
+    assert len(received) == len(_CHAT_AND_ECHO_EVENTS)
+
+
+@_CHILD_CONFIGS
+async def test_child_graph_streams_only_to_the_node_streaming_it_async(
+    child_config: dict | None,
+) -> None:
+    child = _one_node_graph(_achat_and_echo)
+    received = []
+
+    async def node(state: MessagesState) -> dict:
+        stream = child.astream(
+            {"messages": []}, child_config, stream_mode=["messages", "tools"]
+        )
+        received.extend([part async for part in stream])
+        return {}
+
+    parts = _one_node_graph(node).astream(
+        {"messages": []},
+        {"configurable": {"thread_id": "parent"}},
+        stream_mode=["messages", "tools"],
+    )
+
+    assert [part async for part in parts] == []
+    assert len(received) == len(_CHAT_AND_ECHO_EVENTS)
+
+
+def _reply(state: MessagesState) -> dict:
+    return {"messages": [AIMessage(id="reply", content="hi")]}
+
+
+def test_node_gets_the_messages_returned_in_the_graph_it_streams() -> None:
+    sub = _one_node_graph(_reply)
+    received = []
+
+    def node(state: MessagesState) -> dict:
+        for message, _ in sub.stream({"messages": []}, stream_mode="messages"):
+            received.append(message.id)
+        return {}
+
+    parts = _one_node_graph(node).stream({"messages": []}, stream_mode="messages")
+
+    assert list(parts) == []
+    assert received == ["reply"]
+
+
+def test_node_gets_the_events_of_every_subgraph_it_streams() -> None:
+    sub = _one_node_graph(_chat_and_echo)
+    counts = []
+
+    def node(state: MessagesState) -> dict:
+        for _ in range(2):
+            stream = sub.stream({"messages": []}, stream_mode=["messages", "tools"])
+            counts.append(len(list(stream)))
+        return {}
+
+    _one_node_graph(node).invoke({"messages": []})
+
+    assert counts == [len(_CHAT_AND_ECHO_EVENTS)] * 2
+
+
+async def test_node_gets_the_events_of_every_subgraph_it_streams_async() -> None:
+    sub = _one_node_graph(_achat_and_echo)
+    counts = []
+
+    async def node(state: MessagesState) -> dict:
+        for _ in range(2):
+            stream = sub.astream({"messages": []}, stream_mode=["messages", "tools"])
+            counts.append(len([part async for part in stream]))
+        return {}
+
+    await _one_node_graph(node).ainvoke({"messages": []})
+
+    assert counts == [len(_CHAT_AND_ECHO_EVENTS)] * 2
+
+
+def _push(state: MessagesState) -> dict:
+    push_message(AIMessage(id="pushed", content="hi"), state_key=None)
+    return {}
+
+
+def _namespaces(graph: CompiledStateGraph, subgraphs: bool) -> list[tuple[str, ...]]:
+    parts = graph.stream(
+        {"messages": []}, stream_mode="messages", subgraphs=subgraphs, version="v2"
+    )
+    return [_segment_names(part["ns"]) for part in parts]
+
+
+def _streams_pushing_graph(received: list) -> Any:
+    pushing = _one_node_graph(_push)
+
+    def node(state: MessagesState) -> dict:
+        for message, _ in pushing.stream({"messages": []}, stream_mode="messages"):
+            received.append(message.id)
+        return {}
+
+    return node
+
+
+def test_pushed_message_streams_under_the_namespace_of_its_graph() -> None:
+    pushing = _one_node_graph(_push)
+    parent = StateGraph(MessagesState).add_node("sub", pushing).add_edge(START, "sub")
+
+    assert _namespaces(pushing, subgraphs=True) == [()]
+    assert _namespaces(parent.compile(), subgraphs=False) == []
+    assert _namespaces(parent.compile(), subgraphs=True) == [("sub",)]
+
+
+def test_pushed_message_streams_to_the_node_that_streams_its_graph() -> None:
+    received: list[str] = []
+    graph = _one_node_graph(_streams_pushing_graph(received))
+
+    assert _namespaces(graph, subgraphs=False) == []
+    assert received == ["pushed"]
+
+
+def test_pushed_message_streams_to_every_stream_that_takes_it() -> None:
+    received: list[str] = []
+    graph = _one_node_graph(_streams_pushing_graph(received))
+
+    assert _namespaces(graph, subgraphs=True) == [("node",)]
+    assert received == ["pushed"]

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" },