release(langgraph): 1.2.16 (#9334)

Bumps langgraph from 1.2.15 to 1.2.16.

Includes:

- Keep a DeltaChannel's value when its field changes to a plain reducer
(#9302)
- Save a resumed exit run's pending writes once (#9314)
- Stop a subgraph from rerunning its finished tasks on resume (#9315)
- Apply a resumed `Command`'s update once (#9322)
- Let other tasks finish when a resume value is invalid (#9265)

libs/langgraph/langgraph/pregel/_runner.py

@@ -41,7 +41,7 @@ from langgraph._internal._future import chain_future, run_coroutine_threadsafe
 from langgraph._internal._scratchpad import PregelScratchpad
 from langgraph._internal._typing import MISSING
 from langgraph.constants import TAG_HIDDEN
-from langgraph.errors import GraphBubbleUp, GraphInterrupt
+from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume
 from langgraph.pregel._algo import Call
 from langgraph.pregel._executor import Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
@@ -621,7 +621,11 @@ def _should_stop_others(
     handled_exception_ids: set[int] | None = None,
 ) -> bool:
     """Check if any task failed, if so, cancel all other tasks.
-    GraphInterrupts are not considered failures."""
+    GraphInterrupts are not considered failures.
+
+    Neither is an invalid resume value: it's raised once the other tasks finish, so a
+    task answered in the same resume isn't cut off after its side effects and then run
+    again with its saved answer."""
     for fut in done:
         if fut.cancelled():
             continue
@@ -629,6 +633,7 @@ def _should_stop_others(
             if (
                 id(exc) not in (handled_exception_ids or set())
                 and not isinstance(exc, GraphBubbleUp)
+                and not is_invalid_resume(exc)
                 and fut not in SKIP_RERAISE_SET
             ):
                 return True

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_interruption.py

@@ -1,5 +1,8 @@
+import asyncio
+import operator
+import sys
 from dataclasses import dataclass
-from typing import Any
+from typing import Annotated, Any
 
 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
@@ -251,3 +254,61 @@ def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt(
     assert graph.invoke(resume({"approved": True}), config) == {
         "answer": ["ok", Decision(approved=True)]
     }
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+async def test_invalid_resume_lets_other_answered_tasks_finish(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """An invalid answer fails the resume without cutting off the other answered tasks.
+
+    A task cancelled after its side effect keeps its answer and runs again on the next
+    resume, so the side effect would happen twice.
+    """
+    sent: list[str] = []
+
+    class State(TypedDict):
+        log: Annotated[list[str], operator.add]
+
+    async def send(state: State) -> State:
+        interrupt("send?", response_schema=Decision)
+        sent.append("email")
+        await asyncio.sleep(0.05)  # waiting on the reply to a request already sent
+        return {"log": ["sent"]}
+
+    async def check(state: State) -> State:
+        interrupt("check?", response_schema=Decision)
+        return {"log": ["checked"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("send", send)
+        .add_node("check", check)
+        .add_edge(START, "send")
+        .add_edge(START, "check")
+        .compile(checkpointer=async_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await graph.ainvoke({"log": []}, config)
+    ids = {i.value: i.id for i in (await graph.aget_state(config)).interrupts}
+
+    with pytest.raises(ValidationError, match="approved"):
+        await graph.ainvoke(
+            Command(
+                resume={
+                    ids["send?"]: {"approved": True},
+                    ids["check?"]: {"approved": "nope"},
+                }
+            ),
+            config,
+        )
+    pending = [i.value for i in (await graph.aget_state(config)).interrupts]
+    assert (pending, sent) == (["check?"], ["email"])
+
+    result = await graph.ainvoke(
+        Command(resume={ids["check?"]: {"approved": True}}), config
+    )
+    assert (sorted(result["log"]), sent) == (["checked", "sent"], ["email"])

libs/langgraph/uv.lock

@@ -1451,7 +1451,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/__init__.py

@@ -3,7 +3,7 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 
-__version__ = "0.4.7"
+__version__ = "0.4.8"
 
 __all__ = [
     "Auth",

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

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import logging
 import random
 from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
 from dataclasses import dataclass, field
@@ -36,6 +37,11 @@ from langgraph_sdk.stream.decoders import (
     ToolCallsDecoder,
     validate_interleave_channels,
 )
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    protocol_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.transport import (
     AsyncProtocolTransport,
@@ -44,6 +50,8 @@ from langgraph_sdk.stream.transport import (
     ProtocolWebSocketTransport,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class InterruptPayload(TypedDict):
     """Payload surfaced when the server requests human input for a thread."""
@@ -193,6 +201,7 @@ class RunModule:
             if not gate.done():
                 gate.set_result(None)
             self._owner._run_seen = True
+            self._owner._ensure_lifecycle_watcher_running()
             return result
         except BaseException as err:
             # Why: gate MUST reject on any exit type, including CancelledError,
@@ -1203,6 +1212,7 @@ class AsyncThreadStream:
         # observes a `completed`/`errored` event.
         self._interrupts_lock = asyncio.Lock()
         self._lifecycle_watcher_task: asyncio.Task[None] | None = None
+        self._lifecycle_watcher_rejected = False
         self._lifecycle_watcher_handle: EventStreamHandle | None = None
         self._lifecycle_cursor: int | None = None
         self._lifecycle_max_reconnect_attempts = 5
@@ -1261,8 +1271,11 @@ class AsyncThreadStream:
         # Create the run-done future here (async context guarantees a running loop).
         self._run_done = asyncio.get_running_loop().create_future()
         # Start the lifecycle watcher immediately so reattach and thread.output
-        # work without a preceding run.start call.
-        self._ensure_lifecycle_watcher_running()
+        # work without a preceding run.start call. A minted thread_id names no
+        # thread until run.start creates it, and with custom auth a stream for a
+        # thread that doesn't exist yet fails, so run.start starts the watcher.
+        if self._explicit_thread_id:
+            self._ensure_lifecycle_watcher_running()
         return self
 
     async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1630,6 +1643,13 @@ class AsyncThreadStream:
                 async for event in self._dedup_iter(shared.events):
                     if self._closed:
                         break
+                    if (frame_error := stream_error_frame(event)) is not None:
+                        logger.warning(
+                            "Thread %s event stream failed: %s",
+                            self.thread_id,
+                            frame_error,
+                        )
+                        continue
                     self._observe_event(event)
                     for sub in list(self._subscriptions.values()):
                         if matches_subscription(event, sub.params):
@@ -1655,6 +1675,7 @@ class AsyncThreadStream:
                 if (
                     err is not None
                     and not isinstance(err, asyncio.CancelledError)
+                    and not is_non_retryable_stream_error(err)
                     and not self._closed
                 ):
                     with contextlib.suppress(Exception):
@@ -1681,7 +1702,8 @@ class AsyncThreadStream:
 
         Returns:
             `True` if a new stream was opened (caller should resume fanout),
-            `False` if all reconnect attempts were exhausted or the controller
+            `False` if all reconnect attempts were exhausted, the server
+            rejected the stream with a non-retryable 4xx, or the controller
             was closed in the meantime.
         """
         if self._transport is None:
@@ -1702,7 +1724,9 @@ class AsyncThreadStream:
                 await new_stream.ready
             except asyncio.CancelledError:
                 raise
-            except Exception:
+            except Exception as err:
+                if is_non_retryable_stream_error(err):
+                    return False
                 await self._reconnect_sleep(attempt)
                 continue
             self._shared_stream = new_stream
@@ -1800,9 +1824,7 @@ class AsyncThreadStream:
             # 202/204 — no body. Caller gets an empty result.
             return {}
         if response.get("type") == "error":
-            code = response.get("error", "unknown")
-            message = response.get("message", "")
-            raise RuntimeError(f"Protocol error [{code}]: {message}")
+            raise protocol_error(response)
         meta = response.get("meta")
         if isinstance(meta, dict):
             applied_through_seq = meta.get("applied_through_seq")
@@ -1826,8 +1848,15 @@ class AsyncThreadStream:
             await asyncio.wait_for(asyncio.shield(gate), timeout=timeout)
 
     def _ensure_lifecycle_watcher_running(self) -> None:
-        if self._lifecycle_watcher_task is not None:
+        task = self._lifecycle_watcher_task
+        if task is not None and not self._lifecycle_watcher_rejected:
             return
+        if task is not None:
+            # A 4xx stopped the watcher before this run started, typically the
+            # 404 for a thread this run.start just created, so the result it
+            # left isn't this run's.
+            self._run_done = asyncio.get_running_loop().create_future()
+            self._lifecycle_watcher_rejected = False
         self._lifecycle_watcher_task = asyncio.create_task(
             self._run_lifecycle_watcher()
         )
@@ -1850,7 +1879,11 @@ class AsyncThreadStream:
         if self._transport is None:
             return
         reconnect_attempts = 0
+        frame_error: RuntimeError | None = None
         while not self._closed:
+            # A stream opened before run.start can 404 for the thread that
+            # run.start creates, so it gets one more try once a run has started.
+            opened_before_run = not self._run_seen
             try:
                 handle = self._transport.open_event_stream(
                     self._lifecycle_stream_params()
@@ -1860,6 +1893,7 @@ class AsyncThreadStream:
                 async for event in handle.events:
                     if self._closed:
                         return
+                    frame_error = stream_error_frame(event) or frame_error
                     self._observe_lifecycle_event(event)
                     await self._apply_lifecycle_event(event)
                 err = await handle.done
@@ -1868,38 +1902,43 @@ class AsyncThreadStream:
                     # event. Resolve `_run_done` as errored so awaiters of
                     # `thread.output` don't hang.
                     if err is None:
-                        run_done = self._run_done
-                        if run_done is not None and not run_done.done():
-                            run_done.set_result(
-                                _RunTerminal(
-                                    status="errored",
-                                    error=RuntimeError(
-                                        "lifecycle stream ended before terminal event"
-                                    ),
-                                )
+                        self._fail_run_done(
+                            frame_error
+                            or RuntimeError(
+                                "lifecycle stream ended before terminal event"
                             )
+                        )
                     return
                 reconnect_attempts += 1
-                if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+                if (
+                    is_non_retryable_stream_error(err)
+                    or reconnect_attempts > self._lifecycle_max_reconnect_attempts
+                ):
                     raise err
                 await asyncio.sleep(0.05)
             except asyncio.CancelledError:
                 raise
             except Exception as exc:
+                if is_non_retryable_stream_error(exc):
+                    if opened_before_run and self._run_seen:
+                        continue
+                    self._lifecycle_watcher_rejected = True
+                    self._fail_run_done(exc)
+                    return
                 reconnect_attempts += 1
                 if reconnect_attempts <= self._lifecycle_max_reconnect_attempts:
                     await asyncio.sleep(0.05)
                     continue
-                run_done = self._run_done
-                if run_done is not None and not run_done.done():
-                    run_done.set_result(
-                        _RunTerminal(
-                            status="errored",
-                            error=RuntimeError(f"Lifecycle transport failed: {exc}"),
-                        )
-                    )
+                self._fail_run_done(
+                    frame_error or RuntimeError(f"Lifecycle transport failed: {exc}")
+                )
                 return
 
+    def _fail_run_done(self, error: BaseException) -> None:
+        run_done = self._run_done
+        if run_done is not None and not run_done.done():
+            run_done.set_result(_RunTerminal(status="errored", error=error))
+
     async def _fetch_state(self) -> dict[str, Any]:
         """Fetch the current thread state from the REST endpoint."""
         return await self._http.get(

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

@@ -34,6 +34,11 @@ from langgraph_sdk.stream.decoders import (
     ToolCallsDecoder,
     validate_interleave_channels,
 )
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    protocol_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.sync_controller import (
     SyncStreamController,
@@ -237,6 +242,7 @@ class SyncRunModule:
         controller = self._owner._controller
         if controller is not None and controller._run_start_gate is not None:
             controller._run_start_gate.set()
+        self._owner._ensure_lifecycle_watcher_running()
         return result
 
     def respond(
@@ -1111,6 +1117,7 @@ class SyncThreadStream:
         self.interrupted: bool = False
         self.interrupts: list[InterruptPayload] = []
         self._lifecycle_watcher_thread: threading.Thread | None = None
+        self._lifecycle_watcher_lock = threading.Lock()
         self._lifecycle_watcher_handle: SyncEventStreamHandle | None = None
         self._lifecycle_cursor: int | None = None
         self._lifecycle_max_reconnect_attempts = 5
@@ -1149,7 +1156,11 @@ class SyncThreadStream:
             self._transport, run_start_gate=run_start_gate
         )
         self._run_done = _BlockingResult()
-        self._ensure_lifecycle_watcher_running()
+        # A minted thread_id names no thread until run.start creates it, and
+        # with custom auth a stream for a thread that doesn't exist yet fails,
+        # so run.start starts the watcher instead.
+        if self._explicit_thread_id:
+            self._ensure_lifecycle_watcher_running()
         return self
 
     def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1485,9 +1496,7 @@ class SyncThreadStream:
         if response is None:
             return {}
         if response.get("type") == "error":
-            code = response.get("error", "unknown")
-            message = response.get("message", "")
-            raise RuntimeError(f"Protocol error [{code}]: {message}")
+            raise protocol_error(response)
         meta = response.get("meta")
         if isinstance(meta, dict):
             applied_through_seq = meta.get("applied_through_seq")
@@ -1496,14 +1505,20 @@ class SyncThreadStream:
         return response.get("result", {})
 
     def _ensure_lifecycle_watcher_running(self) -> None:
-        if self._lifecycle_watcher_thread is not None:
-            return
-        self._lifecycle_watcher_thread = threading.Thread(
-            target=self._run_lifecycle_watcher,
-            name="langgraph-sdk-sync-lifecycle",
-            daemon=True,
-        )
-        self._lifecycle_watcher_thread.start()
+        with self._lifecycle_watcher_lock:
+            if self._lifecycle_watcher_thread is not None:
+                return
+            if self._run_done is not None and self._run_done.done():
+                # A 4xx stopped the watcher before this run started, typically
+                # the 404 for a thread this run.start just created, so the
+                # result it left isn't this run's.
+                self._run_done = _BlockingResult()
+            self._lifecycle_watcher_thread = threading.Thread(
+                target=self._run_lifecycle_watcher,
+                name="langgraph-sdk-sync-lifecycle",
+                daemon=True,
+            )
+            self._lifecycle_watcher_thread.start()
 
     def _observe_lifecycle_event(self, event: Event) -> None:
         seq = event.get("seq")
@@ -1523,7 +1538,11 @@ class SyncThreadStream:
         if self._transport is None:
             return
         reconnect_attempts = 0
+        frame_error: RuntimeError | None = None
         while not self._closed:
+            # A stream opened before run.start can 404 for the thread that
+            # run.start creates, so it gets one more try once a run has started.
+            opened_before_run = not self._run_seen
             try:
                 handle = self._transport.open_event_stream(
                     self._lifecycle_stream_params()
@@ -1532,28 +1551,64 @@ class SyncThreadStream:
                 for event in handle.events:
                     if self._closed:
                         return
+                    frame_error = stream_error_frame(event) or frame_error
                     self._observe_lifecycle_event(event)
                     self._apply_lifecycle_event(event)
                 err = handle.error()
                 if err is None:
+                    # Clean EOF: stream ended without a terminal lifecycle
+                    # event. Resolve `_run_done` as errored so `thread.output`
+                    # doesn't block forever.
+                    if not self._closed:
+                        self._fail_run_done(
+                            frame_error
+                            or RuntimeError(
+                                "lifecycle stream ended before terminal event"
+                            )
+                        )
                     return
                 reconnect_attempts += 1
-                if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+                if (
+                    is_non_retryable_stream_error(err)
+                    or reconnect_attempts > self._lifecycle_max_reconnect_attempts
+                ):
                     raise err
             except Exception as exc:
+                if is_non_retryable_stream_error(exc):
+                    if self._stop_rejected_lifecycle_watcher(
+                        exc, unless_run_started=opened_before_run
+                    ):
+                        return
+                    continue
                 reconnect_attempts += 1
                 if reconnect_attempts <= self._lifecycle_max_reconnect_attempts:
                     continue
-                run_done = self._run_done
-                if run_done is not None and not run_done.done():
-                    run_done.set_result(
-                        _RunTerminal(
-                            status="errored",
-                            error=RuntimeError(f"Lifecycle transport failed: {exc}"),
-                        )
-                    )
+                self._fail_run_done(
+                    frame_error or RuntimeError(f"Lifecycle transport failed: {exc}")
+                )
                 return
 
+    def _stop_rejected_lifecycle_watcher(
+        self, error: BaseException, *, unless_run_started: bool = False
+    ) -> bool:
+        """Fail the run with a 4xx `error` and clear the watcher, so run.start starts another.
+
+        Returns False, changing nothing, when `unless_run_started` is set and a
+        run has started. Holding the lock keeps a run.start from seeing this
+        watcher as still running once it has decided to stop.
+        """
+        with self._lifecycle_watcher_lock:
+            if unless_run_started and self._run_seen:
+                return False
+            self._fail_run_done(error)
+            self._lifecycle_watcher_thread = None
+            return True
+
+    def _fail_run_done(self, error: BaseException) -> None:
+        run_done = self._run_done
+        if run_done is not None and not run_done.done():
+            run_done.set_result(_RunTerminal(status="errored", error=error))
+
     def _fetch_state(self) -> dict[str, Any]:
         """Fetch the current thread state from the REST endpoint."""
         return self._http.get(

libs/sdk-py/langgraph_sdk/stream/errors.py

@@ -0,0 +1,46 @@
+"""Thread-stream failures, shared by the async and sync clients."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+from websockets.exceptions import ConnectionClosed, InvalidStatus
+
+
+def protocol_error(response: dict[str, Any]) -> RuntimeError:
+    """Build the error for a protocol `{"type": "error"}` envelope."""
+    code = response.get("error", "unknown")
+    message = response.get("message", "")
+    return RuntimeError(f"Protocol error [{code}]: {message}")
+
+
+def stream_error_frame(event: Any) -> RuntimeError | None:
+    """Return the error in an unsolicited `{"type": "error", "id": null}` frame.
+
+    The server sends one just before it closes a thread stream whose read failed.
+    """
+    if (
+        isinstance(event, dict)
+        and event.get("type") == "error"
+        and event.get("id") is None
+    ):
+        return protocol_error(event)
+    return None
+
+
+def is_non_retryable_stream_error(error: BaseException | None) -> bool:
+    """Whether reopening the stream would fail the same way.
+
+    True for a 4xx other than 408 and 429: the status of a refused stream request
+    or WebSocket upgrade, or a WebSocket close code of 4000 plus that status. A
+    server that rejects a WebSocket after accepting it closes with such a code,
+    because a browser can't read the status of a refused upgrade.
+    """
+    if isinstance(error, (httpx.HTTPStatusError, InvalidStatus)):
+        status = error.response.status_code
+    elif isinstance(error, ConnectionClosed) and error.rcvd is not None:
+        status = error.rcvd.code - 4000
+    else:
+        return False
+    return 400 <= status < 500 and status not in (408, 429)

libs/sdk-py/langgraph_sdk/stream/sync_controller.py

@@ -13,6 +13,10 @@ from typing import Any
 
 from langchain_protocol import Event, SubscribeParams
 
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, filter_covers
 from langgraph_sdk.stream.transport import (
     SyncEventStreamHandle,
@@ -167,6 +171,13 @@ class SyncStreamController:
                 return
             try:
                 for event in self._dedup_iter(shared.events):
+                    if (frame_error := stream_error_frame(event)) is not None:
+                        _logger.warning(
+                            "Thread %s event stream failed: %s",
+                            self._transport.thread_id,
+                            frame_error,
+                        )
+                        continue
                     with self._lock:
                         if self._closed:
                             break
@@ -190,7 +201,11 @@ class SyncStreamController:
                     continue  # rotation happened; pick up new stream
                 # No rotation — check if this was a transport drop
             err = shared.error()
-            if err is not None and not self._closed:
+            if (
+                err is not None
+                and not is_non_retryable_stream_error(err)
+                and not self._closed
+            ):
                 if self._reconnect_shared_stream():
                     continue
             break
@@ -199,25 +214,6 @@ class SyncStreamController:
             for sub in self._subscriptions.values():
                 sub.queue.put(None)
 
-    def _reconnect_shared_stream(self) -> bool:
-        with self._lock:
-            base_filter = self._shared_stream_filter
-        if base_filter is None:
-            return False
-        for _ in range(self._max_reconnect_attempts):
-            with self._lock:
-                if self._closed:
-                    return False
-                params = self._filter_with_since(base_filter)
-            try:
-                new_stream = self._transport.open_event_stream(params)
-            except Exception:
-                continue
-            with self._lock:
-                self._shared_stream = new_stream
-            return True
-        return False
-
     def _compute_current_union(
         self, extra: SubscribeParams | None = None
     ) -> dict[str, Any]:
@@ -296,7 +292,8 @@ class SyncStreamController:
         """Attempt to reopen the shared stream after a transport drop.
 
         Returns True if a new stream was successfully opened, False if all
-        reconnect attempts were exhausted or the controller was closed.
+        reconnect attempts were exhausted, the server rejected the stream with a
+        non-retryable 4xx, or the controller was closed.
         """
         base_filter = self._shared_stream_filter
         if base_filter is None:
@@ -317,6 +314,8 @@ class SyncStreamController:
                         old.close()
                 return True
             except Exception as err:
+                if is_non_retryable_stream_error(err):
+                    return False
                 _logger.debug("sync reconnect attempt %d failed: %r", attempt, err)
         return False
 

libs/sdk-py/tests/streaming/_events.py

@@ -69,6 +69,16 @@ def lifecycle_errored_event(
     return _base(seq, "lifecycle", namespace or [], {"event": "failed", "error": error})
 
 
+def thread_stream_failed_frame() -> dict[str, Any]:
+    """Unsolicited error frame the server sends before closing a failed stream."""
+    return {
+        "type": "error",
+        "id": None,
+        "error": "unknown_error",
+        "message": "Thread event stream failed.",
+    }
+
+
 def values_event(
     seq: int = 0, namespace: list[str] | None = None, **data: Any
 ) -> dict[str, Any]:

libs/sdk-py/tests/streaming/_fake_server.py

@@ -60,6 +60,7 @@ class _StreamScript:
     events: list[dict[str, Any]]
     delay: float = 0.0
     fail_after: int | None = None
+    status: int = 200
 
 
 class FakeServer:
@@ -263,6 +264,8 @@ class FakeServer:
                     events=list(self.scripted_events), delay=self._stream_delay
                 )
             )
+            if script.status != 200:
+                return httpx.Response(script.status, json={"detail": "scripted"})
             self.open_event_streams += 1
             self._open_event_streams_max = max(
                 self._open_event_streams_max, self.open_event_streams

libs/sdk-py/tests/streaming/_sync_fake_server.py

@@ -14,6 +14,7 @@ import orjson
 class SyncStreamScript:
     events: list[dict[str, Any]]
     fail_after: int | None = None
+    status: int = 200
 
 
 class _SseByteStream(httpx.SyncByteStream):
@@ -113,6 +114,8 @@ class SyncFakeServer:
                 if self._stream_scripts
                 else SyncStreamScript(events=list(self.scripted_events))
             )
+            if script.status != 200:
+                return httpx.Response(script.status, json={"detail": "scripted"})
             return httpx.Response(
                 200,
                 headers={"content-type": "text/event-stream"},

libs/sdk-py/tests/streaming/test_lifecycle_watcher.py

@@ -17,6 +17,7 @@ from streaming._events import (
     lifecycle_completed_event,
     lifecycle_errored_event,
     lifecycle_event,
+    thread_stream_failed_frame,
 )
 from streaming._fake_server import FakeServer, _StreamScript
 
@@ -48,7 +49,7 @@ async def test_interrupts_populated_from_input_requested_event():
 
 
 async def test_aenter_starts_lifecycle_watcher():
-    """Entering AsyncThreadStream opens lifecycle/input SSE before run.start."""
+    """Entering with an explicit thread_id opens the lifecycle/input SSE before run.start."""
     fake = FakeServer()
     fake.script([lifecycle_event(seq=0, phase="started")])
     asgi = httpx.ASGITransport(app=fake.app)
@@ -274,3 +275,137 @@ async def test_lifecycle_watcher_reconnects_with_since_after_transport_drop():
     assert fake.stream_request_bodies[0]["channels"] == ["lifecycle", "input"]
     assert fake.stream_request_bodies[1]["channels"] == ["lifecycle", "input"]
     assert fake.stream_request_bodies[1]["since"] == 1
+
+
+async def test_lifecycle_error_frame_becomes_the_run_error():
+    fake = FakeServer()
+    fake.script([lifecycle_event(seq=0, phase="running"), thread_stream_failed_frame()])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+async def test_lifecycle_watcher_stops_on_a_404_without_retrying():
+    fake = FakeServer()
+    fake.script_sequence([_StreamScript(events=[], status=404)])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert isinstance(terminal.error, httpx.HTTPStatusError)
+    assert terminal.error.response.status_code == 404
+    assert len(fake.stream_request_bodies) == 1
+
+
+async def test_minted_thread_opens_the_lifecycle_stream_after_run_start():
+    fake = FakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            assert thread._lifecycle_watcher_task is None
+            await thread.run.start(input={})
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert terminal.status == "completed"
+
+
+def _new_thread_scripts() -> list[_StreamScript]:
+    return [
+        _StreamScript(events=[], status=404),
+        _StreamScript(events=[lifecycle_completed_event(seq=1)]),
+    ]
+
+
+async def test_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+    fake = FakeServer()
+    fake.script_sequence(_new_thread_scripts())
+    fake.set_state({"ok": True})
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            rejected = thread._run_done
+            assert rejected is not None
+            await asyncio.wait_for(rejected, timeout=2.0)
+            await thread.run.start(input={})
+            output = await thread.output.with_timeout(2.0)
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+    fake = FakeServer()
+    fake.script_sequence(_new_thread_scripts())
+    fake.set_state({"ok": True})
+    stream_opened = asyncio.Event()
+    run_started = asyncio.Event()
+
+    async def handle(request: httpx.Request) -> httpx.Response:
+        if (
+            request.url.path.endswith("/stream/events")
+            and not fake.stream_request_bodies
+        ):
+            stream_opened.set()
+            await run_started.wait()
+        return await fake._handle_request(request)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handle), base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            await asyncio.wait_for(stream_opened.wait(), timeout=2.0)
+            await thread.run.start(input={})
+            run_started.set()
+            output = await thread.output.with_timeout(2.0)
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+    fake = FakeServer()
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        stream = threads.stream(thread_id="t-1", assistant_id="agent")
+        stream._lifecycle_max_reconnect_attempts = 1
+        fake.script_sequence(
+            [
+                _StreamScript(events=[thread_stream_failed_frame()], fail_after=1)
+                for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+            ]
+        )
+        async with stream as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )

libs/sdk-py/tests/streaming/test_shared_stream.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, AsyncIterator
 from typing import Any, cast
 from unittest.mock import MagicMock
 
 import httpx
+from websockets.exceptions import ConnectionClosedError
+from websockets.frames import Close
 
 from langgraph_sdk._async.http import HttpClient
 from langgraph_sdk._async.threads import ThreadsClient
@@ -348,3 +350,66 @@ async def test_send_command_applied_through_seq_seeds_shared_stream_since():
     ]
     assert len(values_requests) == 1
     assert values_requests[0]["since"] == 17
+
+
+async def _drain(events: AsyncIterator[Any]) -> list[Any]:
+    return [event async for event in events]
+
+
+async def test_shared_stream_stops_reconnecting_on_a_404():
+    fake = FakeServer()
+    fake.script_sequence(
+        [
+            _StreamScript(events=[values_event(seq=1)], fail_after=1),
+            _StreamScript(events=[], status=404),
+        ]
+    )
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            received = await asyncio.wait_for(
+                _drain(thread.subscribe(["values"])), timeout=5.0
+            )
+
+    assert [event["seq"] for event in received] == [1]
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_shared_stream_does_not_reconnect_after_a_4404_close(
+    monkeypatch: Any,
+) -> None:
+    opened: list[dict[str, Any]] = []
+
+    def open_then_close_with_4404(
+        _self: ProtocolSseTransport, params: dict[str, Any]
+    ) -> EventStreamHandle:
+        opened.append(params)
+        loop = asyncio.get_running_loop()
+        ready: asyncio.Future[None] = loop.create_future()
+        ready.set_result(None)
+        done: asyncio.Future[BaseException | None] = loop.create_future()
+        done.set_result(ConnectionClosedError(Close(4404, "not found"), None))
+
+        async def events() -> AsyncGenerator[Any, None]:
+            await asyncio.sleep(0)
+            yield values_event(seq=1)
+
+        async def close() -> None:
+            return None
+
+        return EventStreamHandle(events=events(), ready=ready, done=done, close=close)
+
+    monkeypatch.setattr(
+        ProtocolSseTransport, "open_event_stream", open_then_close_with_4404
+    )
+    async with httpx.AsyncClient(base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            received = await asyncio.wait_for(
+                _drain(thread.subscribe(["values"])), timeout=5.0
+            )
+
+    assert [event["seq"] for event in received] == [1]
+    assert len(opened) == 1

libs/sdk-py/tests/streaming/test_stream_errors.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+from websockets.datastructures import Headers
+from websockets.exceptions import ConnectionClosedError, InvalidStatus
+from websockets.frames import Close
+from websockets.http11 import Response
+
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    stream_error_frame,
+)
+
+
+def _http_error(status: int) -> httpx.HTTPStatusError:
+    request = httpx.Request("POST", "http://test/threads/t-1/stream/events")
+    response = httpx.Response(status, request=request)
+    return httpx.HTTPStatusError("scripted", request=request, response=response)
+
+
+def _ws_close(code: int) -> ConnectionClosedError:
+    return ConnectionClosedError(Close(code, "scripted"), None)
+
+
+@pytest.mark.parametrize(
+    ("error", "expected"),
+    [
+        pytest.param(_http_error(403), True, id="http-403"),
+        pytest.param(_http_error(404), True, id="http-404"),
+        pytest.param(_http_error(408), False, id="http-408"),
+        pytest.param(_http_error(429), False, id="http-429"),
+        pytest.param(_http_error(503), False, id="http-503"),
+        pytest.param(_ws_close(4404), True, id="ws-close-4404"),
+        pytest.param(_ws_close(4422), True, id="ws-close-4422"),
+        pytest.param(_ws_close(4429), False, id="ws-close-4429"),
+        pytest.param(_ws_close(4503), False, id="ws-close-4503"),
+        pytest.param(_ws_close(1011), False, id="ws-close-1011"),
+        pytest.param(
+            InvalidStatus(Response(403, "Forbidden", Headers())),
+            True,
+            id="ws-upgrade-403",
+        ),
+        pytest.param(RuntimeError("network down"), False, id="no-status"),
+        pytest.param(None, False, id="none"),
+    ],
+)
+def test_is_non_retryable_stream_error(
+    error: BaseException | None, expected: bool
+) -> None:
+    assert is_non_retryable_stream_error(error) is expected
+
+
+def test_stream_error_frame_reads_only_unsolicited_error_frames() -> None:
+    frame = {
+        "type": "error",
+        "id": None,
+        "error": "unknown_error",
+        "message": "Thread event stream failed.",
+    }
+
+    assert (
+        str(stream_error_frame(frame))
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+    assert stream_error_frame({**frame, "id": 3}) is None
+    assert stream_error_frame({"type": "event", "method": "values"}) is None

libs/sdk-py/tests/streaming/test_sync_thread_stream.py

@@ -35,6 +35,7 @@ from streaming._events import (
     message_text_delta_event,
     message_text_finish_event,
     tasks_start_event,
+    thread_stream_failed_frame,
     tool_finished_event,
     tool_started_event,
     updates_event,
@@ -868,3 +869,196 @@ def test_interleave_projections_data_channel_scoped_to_root_namespace():
             ]
     assert {"scope": "root"} in checkpoints
     assert {"scope": "child"} not in checkpoints
+
+
+def test_sync_lifecycle_clean_eof_fails_the_run_instead_of_blocking():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_started_event(seq=0)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert "ended before terminal" in str(terminal.error)
+
+
+def test_sync_lifecycle_error_frame_becomes_the_run_error():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_event(seq=0, phase="running"), thread_stream_failed_frame()])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+def test_sync_lifecycle_watcher_stops_on_a_404_without_retrying():
+    fake = SyncFakeServer()
+    fake.script_sequence([SyncStreamScript(events=[], status=404)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert isinstance(terminal.error, httpx.HTTPStatusError)
+    assert terminal.error.response.status_code == 404
+    assert len(fake.stream_request_bodies) == 1
+
+
+def test_sync_minted_thread_opens_the_lifecycle_stream_after_run_start():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(assistant_id="agent") as thread:
+            assert thread._lifecycle_watcher_thread is None
+            thread.run.start(input={})
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "completed"
+
+
+def _sync_new_thread_scripts() -> list[SyncStreamScript]:
+    return [
+        SyncStreamScript(events=[], status=404),
+        SyncStreamScript(events=[lifecycle_completed_event(seq=1)]),
+    ]
+
+
+def test_sync_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+    fake = SyncFakeServer()
+    fake.script_sequence(_sync_new_thread_scripts())
+    fake.set_state({"ok": True})
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            rejected = thread._run_done
+            assert rejected is not None
+            rejected.result(timeout=2.0)
+            thread.run.start(input={})
+            output = thread.output
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+def test_sync_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+    fake = SyncFakeServer()
+    fake.script_sequence(_sync_new_thread_scripts())
+    fake.set_state({"ok": True})
+    stream_opened = threading.Event()
+    run_started = threading.Event()
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        if (
+            request.url.path.endswith("/stream/events")
+            and not fake.stream_request_bodies
+        ):
+            stream_opened.set()
+            run_started.wait(timeout=2.0)
+        return fake._handle(request)
+
+    with httpx.Client(
+        transport=httpx.MockTransport(handle), base_url="http://test"
+    ) as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            assert stream_opened.wait(timeout=2.0)
+            thread.run.start(input={})
+            run_started.set()
+            output = thread.output
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+def test_sync_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+    fake = SyncFakeServer()
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        stream = threads.stream(thread_id="t-1", assistant_id="agent")
+        stream._lifecycle_max_reconnect_attempts = 1
+        fake.script_sequence(
+            [
+                SyncStreamScript(events=[thread_stream_failed_frame()], fail_after=1)
+                for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+            ]
+        )
+        with stream as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+def test_sync_concurrent_run_starts_start_one_lifecycle_watcher(monkeypatch):
+    real_thread = threading.Thread
+    created_watchers: list[str] = []
+
+    class _SlowToCreateThread(real_thread):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            if kwargs.get("name") == "langgraph-sdk-sync-lifecycle":
+                created_watchers.append(kwargs["name"])
+                time.sleep(0.05)
+            super().__init__(*args, **kwargs)
+
+    monkeypatch.setattr(threading, "Thread", _SlowToCreateThread)
+    fake = SyncFakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(assistant_id="agent") as thread:
+            barrier = threading.Barrier(2)
+
+            def start_run() -> None:
+                barrier.wait()
+                thread.run.start(input={})
+
+            runners = [real_thread(target=start_run) for _ in range(2)]
+            for runner in runners:
+                runner.start()
+            for runner in runners:
+                runner.join()
+
+    assert len(created_watchers) == 1
+
+
+def test_sync_reconnect_stops_on_a_404(monkeypatch):
+    monkeypatch.setattr(_ctrl_mod.time, "sleep", lambda _delay: None)
+    opened: list[dict[str, Any]] = []
+
+    class _NotFoundTransport(SyncProtocolSseTransport):
+        def open_event_stream(self, params: dict) -> SyncEventStreamHandle:
+            opened.append(params)
+            request = httpx.Request("POST", "http://test/threads/t-1/stream/events")
+            raise httpx.HTTPStatusError(
+                "not found",
+                request=request,
+                response=httpx.Response(404, request=request),
+            )
+
+    with httpx.Client(base_url="http://test") as raw:
+        transport = _NotFoundTransport(client=raw, thread_id="t-1")
+        controller = SyncStreamController(transport, max_reconnect_attempts=5)
+        controller._shared_stream_filter = {"channels": ["values"]}
+        assert controller._reconnect_shared_stream() is False
+
+    assert len(opened) == 1

libs/sdk-py/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },