fix(langgraph): let error_handler recover on every path and stream failed task results

- A node's error_handler ran, but the original error was still raised when
  the node ran on the executor path: streaming messages or custom, with
  subgraphs=True, or next to a parallel node. The runner marks a handled
  future as such, but the executors re-raised every failed future on exit
  without checking. The marker set now lives in the executor module, and
  both executors skip those futures.
- With only tasks or debug streaming, a failing node's result was queued but
  never streamed, because the single-task fast path raised before the caller
  drained the queue. It now yields once before re-raising.

Ideas from #8611 by SnowingFox and #6984 by Br1an67.

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/_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/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]

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