fix(sdk-py): retry runs.wait, join and cancel(wait=True) reconnects with backoff (#9481)

`runs.wait`, `runs.join` and `runs.cancel(wait=True)` reconnect to the
run's `Location` when the response is cut off, but with no delay: a 429,
a 5xx or a refused connection on the reconnect raised at once, and a
dropped body reconnected again immediately, so they could not ride out a
server restart or a deploy. #9462 fixed the same for streams.

The reconnect now waits with the stream path's backoff and retries those
failures within the same five attempts; any other status and a failed
first request still raise. Each attempt is one GET with
`reconnect_limit=0`, so the loop owns every retry (the join `Location`
points back at itself, so nothing is lost by not following it).
`request_reconnect` raises plain `httpx.HTTPStatusError`, so it checks
429 and 5xx directly, the same set the stream path catches as
`RateLimitError` and `InternalServerError`.

`test_wait_retries_a_failed_reconnect_after_a_backoff` covers each
failure, sync and async, and fails on `main`; the other two new tests
check the budget and that a 404 still raises. `no_sleep` moves to
`conftest.py` so both test files can use it.

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/cli/js-monorepo-example/yarn.lock

@@ -2148,6 +2148,6 @@ yocto-queue@^0.1.0:
   integrity sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==
 
 "zod@^3.25.76 || ^4":
-  version "4.3.6"
-  resolved "https://registry.yarnpkg.com/zod/-/zod-4.3.6.tgz#89c56e0aa7d2b05107d894412227087885ab112a"
-  integrity sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==
+  version "4.6.7"
+  resolved "https://registry.yarnpkg.com/zod/-/zod-4.6.7.tgz#e806cf64a7f4e37635ce9f2f52c04c3766f75e5b"
+  integrity sha512-g8W2D2kU6a8untc053fX1uVBKzM7yqAxW/3zwk83+6RyF2+SKOuV+c0wBDhUwzV1PaGapmZH+rmcUzPf+3CGwg==

libs/cli/langgraph_cli/config.py

@@ -1589,8 +1589,6 @@ ADD {relpath} /deps/{name}
     docker_file_contents.extend(
         [
             f"FROM {image_str}",
-            # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
-            "ARG USER_ENV_VARS_HASH",
             "",
             os.linesep.join(config["dockerfile_lines"]),
             "",
@@ -1681,8 +1679,6 @@ def node_config_to_docker(
 
     docker_file_contents = [
         f"FROM {image_str}",
-        # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
-        "ARG USER_ENV_VARS_HASH",
         "",
         os.linesep.join(config["dockerfile_lines"]),
         "",

libs/cli/langgraph_cli/uv_lock.py

@@ -955,8 +955,6 @@ def python_config_to_docker_uv_lock(
         docker_plan.add_blank()
 
     docker_plan.add_instruction("FROM", image_str)
-    # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
-    docker_plan.add_instruction("ARG", "USER_ENV_VARS_HASH")
     docker_plan.add_blank()
     docker_plan.extend_nonempty(config["dockerfile_lines"])
     if config["dockerfile_lines"]:

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -148,7 +148,6 @@ services:
             dockerfile_inline: |
                 # syntax=docker/dockerfile:1.4
                 FROM langchain/langgraph-api:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding local package . --
                 ADD . /deps/cli
                 # -- End of local package . --

libs/cli/tests/unit_tests/test_config.py

@@ -910,7 +910,6 @@ def test_config_to_docker_simple():
     expected_docker_stdin = f"""\
 # syntax=docker/dockerfile:1.4
 FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 # -- Installing local requirements --
 COPY --from=outer-requirements.txt requirements.txt /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
 RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt -r /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
@@ -975,7 +974,6 @@ def test_config_to_docker_outside_path():
         """\
 # syntax=docker/dockerfile:1.4
 FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
 RUN set -ex && \\
@@ -1036,7 +1034,6 @@ def test_config_to_docker_pipconfig():
     expected_docker_stdin = (
         """\
 FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 ADD pipconfig.txt /pipconfig.txt
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -1100,7 +1097,6 @@ def test_config_to_docker_local_deps():
     )
     expected_docker_stdin = f"""\
 FROM langchain/langgraph-api-custom:3.11
-ARG USER_ENV_VARS_HASH
 # -- Adding non-package dependency graphs --
 ADD ./graphs /deps/outer-graphs/src
 RUN set -ex && \\
@@ -1148,7 +1144,6 @@ dependencies = ["langchain"]"""
     os.remove(pyproject_path)
     expected_docker_stdin = (
         """FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 # -- Adding local package . --
 ADD . /deps/unit_tests
 # -- End of local package . --
@@ -1182,7 +1177,6 @@ def test_config_to_docker_end_to_end():
         base_image="langchain/langgraph-api",
     )
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.12
-ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD pipconfig.txt /pipconfig.txt
@@ -1228,7 +1222,6 @@ def test_config_to_docker_nodejs():
         base_image="langchain/langgraphjs-api",
     )
     expected_docker_stdin = """FROM langchain/langgraphjs-api:20
-ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD . /deps/unit_tests
@@ -1420,7 +1413,6 @@ def test_config_to_docker_nodejs_internal_docker_tag():
         base_image="langchain/langgraphjs-api",
     )
     expected_docker_stdin = """FROM langchain/langgraphjs-api:my-tag
-ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD . /deps/unit_tests
@@ -1532,7 +1524,6 @@ def test_config_to_docker_gen_ui_python():
     )
 
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 RUN /storage/install-node.sh
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -1580,7 +1571,6 @@ def test_config_to_docker_multiplatform():
     )
 
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.11
-ARG USER_ENV_VARS_HASH
 RUN /storage/install-node.sh
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -2853,7 +2843,6 @@ def test_config_to_compose_simple_config():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2895,7 +2884,6 @@ def test_config_to_compose_env_vars():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api-custom:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2941,7 +2929,6 @@ def test_config_to_compose_env_file():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2980,7 +2967,6 @@ def test_config_to_compose_watch():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -3028,7 +3014,6 @@ def test_config_to_compose_end_to_end():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
-                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\

libs/cli/tests/unit_tests/test_env_cache.py

@@ -1,142 +0,0 @@
-"""Generated Dockerfile cache keys; opt into the real BuildKit test locally."""
-
-import hashlib
-import os
-import pathlib
-import subprocess
-import uuid
-
-import pytest
-
-from langgraph_cli.config import config_to_docker, validate_config
-
-
-def generated_prefix(tmp_path: pathlib.Path, runtime: str) -> str:
-    config = {
-        "graphs": {"agent": "agent:graph"},
-        "dockerfile_lines": [
-            f"RUN --mount=type=secret,id=build_marker echo {tmp_path.name} >/dev/null; "
-            "sha256sum /run/secrets/build_marker | cut -d ' ' -f 1 > /result"
-        ],
-    }
-    if runtime == "node":
-        config["node_version"] = "20"
-        config["graphs"] = {"agent": "./agent.js:graph"}
-    else:
-        config.update(python_version="3.11", dependencies=["requests"])
-        if runtime == "uv":
-            config.pop("dependencies")
-            config["source"] = {"kind": "uv", "root": "."}
-            (tmp_path / "uv.lock").write_text("version = 1\n")
-            (tmp_path / "pyproject.toml").write_text(
-                '[project]\nname = "agent"\nversion = "0.1.0"\n'
-                '[build-system]\nrequires = ["setuptools"]\n'
-                'build-backend = "setuptools.build_meta"\n'
-            )
-    dockerfile, _ = config_to_docker(
-        tmp_path / "langgraph.json",
-        validate_config(config),
-        base_image="langchain/langgraph-api",
-    )
-    # Keep the generated stage header and custom RUN exactly as emitted. Avoid
-    # installing real graph dependencies: this test exercises the cache boundary.
-    return dockerfile.split(" > /result", 1)[0] + " > /result\n"
-
-
-@pytest.mark.parametrize("runtime", ["python", "uv", "node"])
-def test_env_hash_in_build_stage_before_custom_run(tmp_path, runtime):
-    prefix = generated_prefix(tmp_path, runtime)
-    assert prefix.index("FROM ") < prefix.index("ARG USER_ENV_VARS_HASH")
-    assert prefix.index("ARG USER_ENV_VARS_HASH") < prefix.index("RUN ")
-
-
-@pytest.mark.skipif(
-    os.environ.get("LANGGRAPH_TEST_BUILDKIT") != "1",
-    reason="Set LANGGRAPH_TEST_BUILDKIT=1 with a local Docker BuildKit daemon",
-)
-@pytest.mark.parametrize("runtime", ["python", "uv", "node"])
-def test_secret_hash_invalidates_real_buildkit_cache(tmp_path, runtime, request):
-    prefix = generated_prefix(tmp_path, runtime)
-    lines = prefix.splitlines()
-    lines[0] = "FROM alpine:3.21"
-    prefix = "\n".join(lines) + "\n"
-    dockerfile = tmp_path / "Dockerfile"
-    marker = tmp_path / "build-marker"
-    (tmp_path / ".dockerignore").write_text("*\n!Dockerfile\n")
-    output_counter = 0
-    cache_image = "langgraph-cli-cache-test:" + uuid.uuid4().hex
-    request.addfinalizer(
-        lambda: subprocess.run(
-            ["docker", "image", "rm", cache_image], capture_output=True, check=False
-        )
-    )
-
-    def build(contents, value, *, pass_hash=True, export_image=False):
-        nonlocal output_counter
-        output_counter += 1
-        marker.write_text(value)
-        dockerfile.write_text(
-            contents + "FROM scratch\nCOPY --from=0 /result /result\n"
-        )
-        output = tmp_path / f"output-{output_counter}"
-        args = [
-            "docker",
-            "buildx",
-            "build",
-            "--progress=plain",
-            "--secret",
-            f"id=build_marker,src={marker}",
-            "--output",
-            f"type=local,dest={output}",
-            "-f",
-            str(dockerfile),
-            str(tmp_path),
-        ]
-        if pass_hash:
-            args.extend(
-                [
-                    "--build-arg",
-                    f"USER_ENV_VARS_HASH={hashlib.sha256(value.encode()).hexdigest()}",
-                ]
-            )
-        if output_counter > 1:
-            args.extend(["--cache-from", cache_image])
-        if export_image:
-            # Export a legacy image with inline cache, then explicitly import it.
-            image_args = args.copy()
-            output_index = image_args.index("--output")
-            del image_args[output_index : output_index + 2]
-            image_args.extend(
-                ["--load", "--tag", cache_image, "--cache-to", "type=inline"]
-            )
-            subprocess.run(image_args, capture_output=True, text=True, check=True)
-        # List arguments only; all inputs are local public test markers.
-        result = subprocess.run(args, capture_output=True, text=True, check=True)
-        return (output / "result").read_text().strip(), result.stderr
-
-    first = f"{runtime}-{tmp_path.name}-first"
-    second = f"{runtime}-{tmp_path.name}-second"
-    legacy = prefix.replace("ARG USER_ENV_VARS_HASH\n", "")
-    legacy_output, _ = build(legacy, first, export_image=True)
-    assert legacy_output == hashlib.sha256(first.encode()).hexdigest()
-    stale_output, _ = build(legacy, second)
-    assert stale_output == legacy_output  # Secret contents alone do not bust cache.
-
-    fixed_output, _ = build(prefix, second)
-    assert fixed_output == hashlib.sha256(second.encode()).hexdigest()
-    assert fixed_output != stale_output  # Old cache is safe when adopting the ARG.
-    repeated_output, repeated_log = build(prefix, second)
-    assert repeated_output == fixed_output
-    run_vertex = next(
-        line.split()[0]
-        for line in repeated_log.splitlines()
-        if "] RUN " in line and "sha256sum" in line
-    )
-    assert f"{run_vertex} CACHED" in repeated_log
-    changed_output, _ = build(prefix, first)
-    assert changed_output == hashlib.sha256(first.encode()).hexdigest()
-    assert changed_output != repeated_output
-
-    # Ordinary CLI callers can still build without supplying the optional hash.
-    unset_output, _ = build(prefix, first, pass_hash=False)
-    assert unset_output == hashlib.sha256(first.encode()).hexdigest()

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

@@ -216,7 +216,7 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
                 :-1
             ]
-            if not self.subgraphs and len(ns) > 0:
+            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/_tools.py

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

libs/langgraph/langgraph/pregel/main.py

@@ -3310,7 +3310,9 @@ class Pregel(
                         True
                         for h in run_manager.handlers
                         if isinstance(h, _StreamingCallbackHandler)
-                        and not isinstance(h, StreamMessagesHandler)
+                        and not isinstance(
+                            h, (StreamMessagesHandler, StreamToolCallHandler)
+                        )
                     ),
                     False,
                 )

libs/langgraph/langgraph/pregel/remote.py

@@ -847,6 +847,11 @@ class RemoteGraph(PregelProtocol):
             # filter for what was actually requested
             if mode not in requested:
                 continue
+            # subgraph events may have been requested only for the parent stream.
+            # messages are exempt: the server dedupes them across namespaces, so
+            # dropping them here would also drop the root-level message
+            if not subgraphs and NS_SEP in chunk.event and mode != "messages":
+                continue
 
             if chunk.event.startswith("messages"):
                 chunk = chunk._replace(data=tuple(chunk.data))
@@ -997,6 +1002,11 @@ class RemoteGraph(PregelProtocol):
             # filter for what was actually requested
             if mode not in requested:
                 continue
+            # subgraph events may have been requested only for the parent stream.
+            # messages are exempt: the server dedupes them across namespaces, so
+            # dropping them here would also drop the root-level message
+            if not subgraphs and NS_SEP in chunk.event and mode != "messages":
+                continue
 
             if chunk.event.startswith("messages"):
                 chunk = chunk._replace(data=tuple(chunk.data))

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

@@ -8950,6 +8950,39 @@ def test_subgraph_streaming_sync() -> None:
     assert result["num_chunks"] == 9
 
 
+def test_subgraph_streaming_node_output_messages_sync() -> None:
+    """Messages returned by a node (not streamed by an LLM) are emitted when a
+    graph is explicitly streamed with `stream_mode="messages"` inside a node."""
+
+    def respond(state: MessagesState) -> dict:
+        return {"messages": [AIMessage("hi there", id="ai1")]}
+
+    subgraph = (
+        StateGraph(MessagesState)
+        .add_node("respond", respond)
+        .add_edge(START, "respond")
+        .compile()
+    )
+    msgs = {"messages": [("user", "hello")]}
+
+    top_level = [msg for msg, _ in subgraph.stream(msgs, stream_mode="messages")]
+
+    def parent_node(state: MessagesState, config: RunnableConfig) -> dict:
+        nonlocal in_node
+        in_node = [
+            msg for msg, _ in subgraph.stream(msgs, config, stream_mode="messages")
+        ]
+        return {}
+
+    in_node: list = []
+    StateGraph(MessagesState).add_node("parent", parent_node).add_edge(
+        START, "parent"
+    ).compile().invoke(msgs)
+
+    assert [m.content for m in top_level] == ["hi there"]
+    assert [m.content for m in in_node] == ["hi there"]
+
+
 def test_get_graph_nonterminal_last_step_source(snapshot: SnapshotAssertion) -> None:
     class State(TypedDict):
         messages: list[str]

libs/langgraph/tests/test_pregel_async.py

@@ -6834,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."""
@@ -7598,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:
@@ -9444,6 +9503,44 @@ async def test_subgraph_streaming_async() -> None:
     assert result["num_chunks"] == 9
 
 
+async def test_subgraph_streaming_node_output_messages_async() -> None:
+    """Messages returned by a node (not streamed by an LLM) are emitted when a
+    graph is explicitly streamed with `stream_mode="messages"` inside a node."""
+
+    async def respond(state: MessagesState) -> dict:
+        return {"messages": [AIMessage("hi there", id="ai1")]}
+
+    subgraph = (
+        StateGraph(MessagesState)
+        .add_node("respond", respond)
+        .add_edge(START, "respond")
+        .compile()
+    )
+    msgs = {"messages": [("user", "hello")]}
+
+    top_level = [msg async for msg, _ in subgraph.astream(msgs, stream_mode="messages")]
+
+    async def parent_node(state: MessagesState, config: RunnableConfig) -> dict:
+        nonlocal in_node
+        in_node = [
+            msg
+            async for msg, _ in subgraph.astream(msgs, config, stream_mode="messages")
+        ]
+        return {}
+
+    in_node: list = []
+    await (
+        StateGraph(MessagesState)
+        .add_node("parent", parent_node)
+        .add_edge(START, "parent")
+        .compile()
+        .ainvoke(msgs)
+    )
+
+    assert [m.content for m in top_level] == ["hi there"]
+    assert [m.content for m in in_node] == ["hi there"]
+
+
 @NEEDS_CONTEXTVARS
 async def test_null_resume_disallowed_with_multiple_interrupts(
     async_checkpointer: BaseCheckpointSaver,

libs/langgraph/tests/test_remote_graph.py

@@ -16,9 +16,11 @@ from langgraph_sdk.schema import StreamPart
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
+from langgraph._internal._constants import CONFIG_KEY_STREAM
 from langgraph.errors import GraphInterrupt
 from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
 from langgraph.pregel import Pregel
+from langgraph.pregel.protocol import StreamProtocol
 from langgraph.pregel.remote import RemoteGraph
 from langgraph.types import Interrupt, StateSnapshot
 from tests.any_str import AnyStr
@@ -863,6 +865,87 @@ async def test_astream_subgraphs_keeps_caller_namespace():
     assert stream_parts == _CALLER_PREFIXED
 
 
+# Subgraph events requested only on behalf of a streaming parent graph
+_PARENT_STREAMED_PARTS = [
+    StreamPart(event="updates|sub:abc", data={"inner": {"chunk": "inner"}}),
+    StreamPart(event="messages|sub:abc", data=[{"content": "token"}, {}]),
+    StreamPart(event="updates", data={"sub": {"chunk": "root"}}),
+]
+
+
+def _parent_streamed_config(parent_parts: list) -> RunnableConfig:
+    return {
+        "configurable": {
+            "thread_id": "thread_1",
+            "checkpoint_ns": "outer:123",
+            CONFIG_KEY_STREAM: StreamProtocol(
+                parent_parts.append, {"updates", "messages"}
+            ),
+        }
+    }
+
+
+def test_stream_without_subgraphs_skips_parent_requested_subgraph_events():
+    mock_sync_client = MagicMock()
+    mock_sync_client.runs.stream.return_value = _PARENT_STREAMED_PARTS
+    remote_pregel = RemoteGraph("test_graph_id", sync_client=mock_sync_client)
+
+    for stream_mode, expected in [
+        ("updates", [{"sub": {"chunk": "root"}}]),
+        # messages are not filtered: the server dedupes them across namespaces
+        ("messages", [({"content": "token"}, {})]),
+    ]:
+        parent_parts: list = []
+        stream_parts = list(
+            remote_pregel.stream(
+                {"input": "data"},
+                _parent_streamed_config(parent_parts),
+                stream_mode=stream_mode,
+            )
+        )
+
+        assert stream_parts == expected
+        # subgraph events are still requested and forwarded to the parent
+        assert mock_sync_client.runs.stream.call_args.kwargs["stream_subgraphs"]
+        assert [ns for ns, _, _ in parent_parts] == [
+            ("outer:123", "sub:abc"),
+            ("outer:123", "sub:abc"),
+            ("outer:123",),
+        ]
+
+
+async def test_astream_without_subgraphs_skips_parent_requested_subgraph_events():
+    mock_async_client = MagicMock()
+    async_iter = MagicMock()
+    async_iter.__aiter__.return_value = _PARENT_STREAMED_PARTS
+    mock_async_client.runs.stream.return_value = async_iter
+    remote_pregel = RemoteGraph("test_graph_id", client=mock_async_client)
+
+    for stream_mode, expected in [
+        ("updates", [{"sub": {"chunk": "root"}}]),
+        # messages are not filtered: the server dedupes them across namespaces
+        ("messages", [({"content": "token"}, {})]),
+    ]:
+        parent_parts: list = []
+        stream_parts = [
+            part
+            async for part in remote_pregel.astream(
+                {"input": "data"},
+                _parent_streamed_config(parent_parts),
+                stream_mode=stream_mode,
+            )
+        ]
+
+        assert stream_parts == expected
+        # subgraph events are still requested and forwarded to the parent
+        assert mock_async_client.runs.stream.call_args.kwargs["stream_subgraphs"]
+        assert [ns for ns, _, _ in parent_parts] == [
+            ("outer:123", "sub:abc"),
+            ("outer:123", "sub:abc"),
+            ("outer:123",),
+        ]
+
+
 def test_invoke():
     # set up test
     mock_sync_client = MagicMock()

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" },
@@ -3419,11 +3419,11 @@ wheels = [
 
 [[package]]
 name = "setuptools"
-version = "83.0.0"
+version = "84.1.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/34/26/f5d29e25ffdb535afef2d35cdb55b325298f96debd670da4c325e08d70f4/setuptools-83.0.0.tar.gz", hash = "sha256:025bccbbf0fa05b6192bc64ae1e7b16e001fd6d6d4d5de03c97b1c1ade523bef", size = 1154254, upload-time = "2026-07-04T15:31:22.699Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/73/eb/7e5f628ee2eb168fecbf8026de7856b2543d6d609cc85f086b9a91ac93d6/setuptools-84.1.0.tar.gz", hash = "sha256:23657205b63d4d31395ab917815247f694aafad28cd21316bf5fd45a51048ea0", size = 1171783, upload-time = "2026-10-19T20:11:43.489Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/5d/40/e1e72872c6354b306daef1703549e8e83b4d43cfea356311bf722a043752/setuptools-83.0.0-py3-none-any.whl", hash = "sha256:29b23c360f22f414dc7336bb39178cc7bcbf6021ed2733cde173f09dba19abb3", size = 1008090, upload-time = "2026-07-04T15:31:20.885Z" },
+    { url = "https://files.pythonhosted.org/packages/18/02/b4bbb90f8f29e334898ff45c6d9b1fffaa6f378017052b3023a06262b5f7/setuptools-84.1.0-py3-none-any.whl", hash = "sha256:5231689c6ac8156cd155cfb98c3fd6fb252c3e7e9bd3d78b64d3c55f132624c9", size = 819380, upload-time = "2026-10-19T20:11:41.684Z" },
 ]
 
 [[package]]

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

@@ -14,9 +14,14 @@ import orjson
 
 from langgraph_sdk._shared.utilities import (
     _orjson_default,
+    _reconnect_delay,
     _validate_reconnect_location,
 )
-from langgraph_sdk.errors import _araise_for_status_typed
+from langgraph_sdk.errors import (
+    InternalServerError,
+    RateLimitError,
+    _araise_for_status_typed,
+)
 from langgraph_sdk.schema import QueryParamTypes, StreamPart
 from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw
 
@@ -146,7 +151,8 @@ class HttpClient:
         on_response: Callable[[httpx.Response], None] | None = None,
         reconnect_limit: int = 5,
     ) -> Any:
-        """Send a request that automatically reconnects to Location header."""
+        """Send a request that reconnects to its Location header if the response is
+        cut off, with backoff between attempts."""
         request_headers, content = await _aencode_json(json)
         if headers:
             request_headers.update(headers)
@@ -176,13 +182,27 @@ class HttpClient:
                     stacklevel=2,
                 )
                 await r.aclose()
-                return await self.request_reconnect(
-                    loc,
-                    "GET",
-                    headers=headers,
-                    # don't pass on_response so it's only called once
-                    reconnect_limit=reconnect_limit - 1,
-                )
+                for attempt in range(reconnect_limit):
+                    await asyncio.sleep(_reconnect_delay(attempt, base=1.0, cap=5.0))
+                    # A restart or a deploy looks like a refused connection, a
+                    # dropped body, a 429 or a 5xx; other errors won't go away.
+                    try:
+                        return await self.request_reconnect(
+                            loc,
+                            "GET",
+                            headers=headers,
+                            # don't pass on_response so it's only called once
+                            reconnect_limit=0,
+                        )
+                    except httpx.HTTPStatusError as err:
+                        status = err.response.status_code
+                        if attempt == reconnect_limit - 1 or (
+                            status != 429 and status < 500
+                        ):
+                            raise
+                    except httpx.TransportError:
+                        if attempt == reconnect_limit - 1:
+                            raise
 
     async def stream(
         self,
@@ -217,72 +237,104 @@ 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
+            if reconnect_path is not None and not any(
+                key.lower() == "last-event-id" and value
+                for key, value in current_headers.items()
+            ):
+                warnings.warn(
+                    "Reconnecting without a Last-Event-ID, so events sent while "
+                    "disconnected are not replayed. Only runs created with "
+                    "stream_resumable=True send the event IDs a reconnect resumes "
+                    "from.",
+                    stacklevel=2,
+                )
 
             current_method = method if reconnect_path is None else "GET"
             current_content = content if reconnect_path is None else None
             current_params = params if reconnect_path is None else None
 
-            retry = False
-            async with self.client.stream(
-                current_method,
-                reconnect_path or path,
-                headers=current_headers,
-                content=current_content,
-                params=current_params,
-            ) as res:
-                if reconnect_path is None and on_response:
-                    on_response(res)
-                # check status
-                await _araise_for_status_typed(res)
-                # check content type
-                content_type = res.headers.get("content-type", "").partition(";")[0]
-                if "text/event-stream" not in content_type:
-                    raise httpx.TransportError(
-                        "Expected response header Content-Type to contain 'text/event-stream', "
-                        f"got {content_type!r}"
-                    )
+            reconnect_cause: httpx.HTTPError | None = None
+            try:
+                async with self.client.stream(
+                    current_method,
+                    reconnect_path or path,
+                    headers=current_headers,
+                    content=current_content,
+                    params=current_params,
+                ) as res:
+                    if reconnect_path is None and on_response:
+                        on_response(res)
+                    # check status
+                    await _araise_for_status_typed(res)
+                    # check content type
+                    content_type = res.headers.get("content-type", "").partition(";")[0]
+                    if "text/event-stream" not in content_type:
+                        raise httpx.TransportError(
+                            "Expected response header Content-Type to contain 'text/event-stream', "
+                            f"got {content_type!r}"
+                        )
 
-                reconnect_location = res.headers.get("location")
-                if reconnect_location:
-                    _validate_reconnect_location(
-                        self.client.base_url, reconnect_location
-                    )
-                    reconnect_path = reconnect_location
+                    reconnect_location = res.headers.get("location")
+                    if reconnect_location:
+                        _validate_reconnect_location(
+                            self.client.base_url, reconnect_location
+                        )
+                        reconnect_path = reconnect_location
 
-                # parse SSE
-                decoder = SSEDecoder()
-                try:
-                    async for line in aiter_lines_raw(res):
-                        sse = decoder.decode(line=cast("bytes", line).rstrip(b"\n"))
-                        if sse is not None:
+                    # parse SSE
+                    decoder = SSEDecoder(last_event_id=last_event_id)
+                    try:
+                        async for line in aiter_lines_raw(res):
+                            sse = decoder.decode(line=cast("bytes", line).rstrip(b"\n"))
+                            if sse is not None:
+                                if decoder.last_event_id is not None:
+                                    last_event_id = decoder.last_event_id
+                                if sse.event or decoder.has_data:
+                                    # Progress since the last drop earns a fresh
+                                    # set of reconnect attempts.
+                                    reconnect_attempts = 0
+                                    yield sse
+                    except httpx.HTTPError as err:
+                        # httpx.TransportError inherits from HTTPError, so transient
+                        # disconnects during streaming land here.
+                        if reconnect_path is None:
+                            raise
+                        reconnect_cause = err
+                    else:
+                        if sse := decoder.decode(b""):
                             if decoder.last_event_id is not None:
                                 last_event_id = decoder.last_event_id
-                            if sse.event or sse.data is not None:
+                            if sse.event or decoder.has_data:
+                                # decoder.decode(b"") flushes the in-flight event and may
+                                # return an empty placeholder when there is no pending
+                                # message. Skip these no-op events so the stream doesn't
+                                # emit a trailing blank item after reconnects.
                                 yield sse
-                except httpx.HTTPError:
-                    # httpx.TransportError inherits from HTTPError, so transient
-                    # disconnects during streaming land here.
-                    if reconnect_path is None:
-                        raise
-                    retry = True
-                else:
-                    if sse := decoder.decode(b""):
-                        if decoder.last_event_id is not None:
-                            last_event_id = decoder.last_event_id
-                        if sse.event or sse.data is not None:
-                            # decoder.decode(b"") flushes the in-flight event and may
-                            # return an empty placeholder when there is no pending
-                            # message. Skip these no-op events so the stream doesn't
-                            # emit a trailing blank item after reconnects.
-                            yield sse
-            if retry:
+            except (httpx.TransportError, RateLimitError, InternalServerError) as err:
+                # A reconnect that can't open (connection refused, 429, 5xx or a
+                # non-SSE body) usually means the server is restarting or
+                # mid-deploy, so retry it like a drop. Before the server sends a
+                # Location there is nothing to reconnect to, so raise.
+                if reconnect_path is None:
+                    raise
+                reconnect_cause = err
+            if reconnect_cause is not None:
                 reconnect_attempts += 1
                 if reconnect_attempts > max_reconnect_attempts:
                     raise httpx.TransportError(
                         "Exceeded maximum SSE reconnection attempts"
-                    )
+                    ) from reconnect_cause
+                await asyncio.sleep(
+                    _reconnect_delay(reconnect_attempts - 1, base=1.0, cap=5.0)
+                )
                 continue
             break
 

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

@@ -16,7 +16,6 @@ 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
 from typing import Any, Literal, TypedDict, cast
@@ -25,7 +24,7 @@ from langchain_core.language_models.chat_model_stream import AsyncChatModelStrea
 from langchain_protocol import Event, SubscribeParams
 
 from langgraph_sdk._async.http import HttpClient
-from langgraph_sdk._shared.utilities import _quote_path_param
+from langgraph_sdk._shared.utilities import _quote_path_param, _reconnect_delay
 from langgraph_sdk.schema import LangSmithTracing, QueryParamTypes
 from langgraph_sdk.stream.controller import _SeenEventIds
 from langgraph_sdk.stream.decoders import (
@@ -1691,11 +1690,13 @@ class AsyncThreadStream:
 
     async def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff and jitter for reconnect attempt `attempt`."""
-        base = self._shared_reconnect_backoff_base
-        cap = self._shared_reconnect_backoff_cap
-        delay = min(cap, base * (2**attempt))
-        jitter = random.uniform(0, delay * 0.25)
-        await asyncio.sleep(delay + jitter)
+        await asyncio.sleep(
+            _reconnect_delay(
+                attempt,
+                base=self._shared_reconnect_backoff_base,
+                cap=self._shared_reconnect_backoff_cap,
+            )
+        )
 
     async def _reconnect_shared_stream(self) -> bool:
         """Attempt to reopen the shared stream after a post-ready transport drop.

libs/sdk-py/langgraph_sdk/_shared/utilities.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import functools
 import os
+import random
 import re
 from collections.abc import Mapping
 from datetime import tzinfo
@@ -194,6 +195,15 @@ def _validate_reconnect_location(base_url: httpx.URL, location: str) -> str:
     return location
 
 
+def _reconnect_delay(attempt: int, *, base: float, cap: float) -> float:
+    """Seconds to wait before reconnect attempt `attempt` (0-based).
+
+    Exponential backoff capped at `cap`, plus up to 25% jitter.
+    """
+    delay = min(cap, base * (2**attempt))
+    return delay + random.uniform(0, delay * 0.25)
+
+
 def _provided_vals(d: Mapping[str, Any]) -> dict[str, Any]:
     return {k: v for k, v in d.items() if v is not None}
 

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

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import sys
+import time
 import warnings
 from collections.abc import Callable, Iterator, Mapping
 from typing import Any, cast
@@ -13,9 +14,14 @@ import orjson
 
 from langgraph_sdk._shared.utilities import (
     _orjson_default,
+    _reconnect_delay,
     _validate_reconnect_location,
 )
-from langgraph_sdk.errors import _raise_for_status_typed
+from langgraph_sdk.errors import (
+    InternalServerError,
+    RateLimitError,
+    _raise_for_status_typed,
+)
 from langgraph_sdk.schema import QueryParamTypes, StreamPart
 from langgraph_sdk.sse import SSEDecoder, iter_lines_raw
 
@@ -146,7 +152,8 @@ class SyncHttpClient:
         on_response: Callable[[httpx.Response], None] | None = None,
         reconnect_limit: int = 5,
     ) -> Any:
-        """Send a request that automatically reconnects to Location header."""
+        """Send a request that reconnects to its Location header if the response is
+        cut off, with backoff between attempts."""
         request_headers, content = _encode_json(json)
         if headers:
             request_headers.update(headers)
@@ -176,13 +183,27 @@ class SyncHttpClient:
                     stacklevel=2,
                 )
                 r.close()
-                return self.request_reconnect(
-                    loc,
-                    "GET",
-                    headers=headers,
-                    # don't pass on_response so it's only called once
-                    reconnect_limit=reconnect_limit - 1,
-                )
+                for attempt in range(reconnect_limit):
+                    time.sleep(_reconnect_delay(attempt, base=1.0, cap=5.0))
+                    # A restart or a deploy looks like a refused connection, a
+                    # dropped body, a 429 or a 5xx; other errors won't go away.
+                    try:
+                        return self.request_reconnect(
+                            loc,
+                            "GET",
+                            headers=headers,
+                            # don't pass on_response so it's only called once
+                            reconnect_limit=0,
+                        )
+                    except httpx.HTTPStatusError as err:
+                        status = err.response.status_code
+                        if attempt == reconnect_limit - 1 or (
+                            status != 429 and status < 500
+                        ):
+                            raise
+                    except httpx.TransportError:
+                        if attempt == reconnect_limit - 1:
+                            raise
 
     def stream(
         self,
@@ -219,69 +240,99 @@ 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
+            if reconnect_path is not None and not any(
+                key.lower() == "last-event-id" and value
+                for key, value in current_headers.items()
+            ):
+                warnings.warn(
+                    "Reconnecting without a Last-Event-ID, so events sent while "
+                    "disconnected are not replayed. Only runs created with "
+                    "stream_resumable=True send the event IDs a reconnect resumes "
+                    "from.",
+                    stacklevel=2,
+                )
 
             current_method = method if reconnect_path is None else "GET"
             current_content = content if reconnect_path is None else None
             current_params = params if reconnect_path is None else None
 
-            retry = False
-            with self.client.stream(
-                current_method,
-                reconnect_path or path,
-                headers=current_headers,
-                content=current_content,
-                params=current_params,
-            ) as res:
-                if reconnect_path is None and on_response:
-                    on_response(res)
-                # check status
-                _raise_for_status_typed(res)
-                # check content type
-                content_type = res.headers.get("content-type", "").partition(";")[0]
-                if "text/event-stream" not in content_type:
-                    raise httpx.TransportError(
-                        "Expected response header Content-Type to contain 'text/event-stream', "
-                        f"got {content_type!r}"
-                    )
+            reconnect_cause: httpx.HTTPError | None = None
+            try:
+                with self.client.stream(
+                    current_method,
+                    reconnect_path or path,
+                    headers=current_headers,
+                    content=current_content,
+                    params=current_params,
+                ) as res:
+                    if reconnect_path is None and on_response:
+                        on_response(res)
+                    # check status
+                    _raise_for_status_typed(res)
+                    # check content type
+                    content_type = res.headers.get("content-type", "").partition(";")[0]
+                    if "text/event-stream" not in content_type:
+                        raise httpx.TransportError(
+                            "Expected response header Content-Type to contain 'text/event-stream', "
+                            f"got {content_type!r}"
+                        )
 
-                reconnect_location = res.headers.get("location")
-                if reconnect_location:
-                    _validate_reconnect_location(
-                        self.client.base_url, reconnect_location
-                    )
-                    reconnect_path = reconnect_location
+                    reconnect_location = res.headers.get("location")
+                    if reconnect_location:
+                        _validate_reconnect_location(
+                            self.client.base_url, reconnect_location
+                        )
+                        reconnect_path = reconnect_location
 
-                decoder = SSEDecoder()
-                try:
-                    for line in iter_lines_raw(res):
-                        sse = decoder.decode(cast(bytes, line).rstrip(b"\n"))
-                        if sse is not None:
+                    decoder = SSEDecoder(last_event_id=last_event_id)
+                    try:
+                        for line in iter_lines_raw(res):
+                            sse = decoder.decode(cast(bytes, line).rstrip(b"\n"))
+                            if sse is not None:
+                                if decoder.last_event_id is not None:
+                                    last_event_id = decoder.last_event_id
+                                if sse.event or decoder.has_data:
+                                    # Progress since the last drop earns a fresh
+                                    # set of reconnect attempts.
+                                    reconnect_attempts = 0
+                                    yield sse
+                    except httpx.HTTPError as err:
+                        # httpx.TransportError inherits from HTTPError, so transient
+                        # disconnects during streaming land here.
+                        if reconnect_path is None:
+                            raise
+                        reconnect_cause = err
+                    else:
+                        if sse := decoder.decode(b""):
                             if decoder.last_event_id is not None:
                                 last_event_id = decoder.last_event_id
-                            if sse.event or sse.data is not None:
+                            if sse.event or decoder.has_data:
+                                # See async stream implementation for rationale on
+                                # skipping empty flush events.
                                 yield sse
-                except httpx.HTTPError:
-                    # httpx.TransportError inherits from HTTPError, so transient
-                    # disconnects during streaming land here.
-                    if reconnect_path is None:
-                        raise
-                    retry = True
-                else:
-                    if sse := decoder.decode(b""):
-                        if decoder.last_event_id is not None:
-                            last_event_id = decoder.last_event_id
-                        if sse.event or sse.data is not None:
-                            # See async stream implementation for rationale on
-                            # skipping empty flush events.
-                            yield sse
-            if retry:
+            except (httpx.TransportError, RateLimitError, InternalServerError) as err:
+                # A reconnect that can't open (connection refused, 429, 5xx or a
+                # non-SSE body) usually means the server is restarting or
+                # mid-deploy, so retry it like a drop. Before the server sends a
+                # Location there is nothing to reconnect to, so raise.
+                if reconnect_path is None:
+                    raise
+                reconnect_cause = err
+            if reconnect_cause is not None:
                 reconnect_attempts += 1
                 if reconnect_attempts > max_reconnect_attempts:
                     raise httpx.TransportError(
                         "Exceeded maximum SSE reconnection attempts"
-                    )
+                    ) from reconnect_cause
+                time.sleep(_reconnect_delay(reconnect_attempts - 1, base=1.0, cap=5.0))
                 continue
             break
 

libs/sdk-py/langgraph_sdk/auth/__init__.py

@@ -811,7 +811,7 @@ class _On:
     def __call__(
         self,
         *,
-        resources: str | Sequence[str],
+        resources: str | Sequence[str] | None = None,
         actions: str | Sequence[str] | None = None,
     ) -> Callable[[AHO], AHO]: ...
 

libs/sdk-py/langgraph_sdk/sse.py

@@ -30,18 +30,14 @@ class BytesLineDecoder:
         # See https://docs.python.org/3/glossary.html#term-universal-newlines
         NEWLINE_CHARS = b"\n\r"
 
-        # We always push a trailing `\r` into the next decode iteration.
-        if self.trailing_cr:
-            text = b"\r" + text
-            self.trailing_cr = False
-        if text.endswith(b"\r"):
-            self.trailing_cr = True
-            text = text[:-1]
+        if not text:
+            return []
+        if self.trailing_cr and text.startswith(b"\n"):
+            text = text[1:]
+        self.trailing_cr = text.endswith(b"\r")
 
         if not text:
-            # NOTE: the edge case input of empty text doesn't occur in practice,
-            # because other httpx internals filter out this value
-            return []  # pragma: no cover
+            return []
 
         trailing_newline = text[-1] in NEWLINE_CHARS
         lines = cast(list[BytesLike], text.splitlines())
@@ -66,7 +62,7 @@ class BytesLineDecoder:
         return lines
 
     def flush(self) -> list[BytesLike]:
-        if not self.buffer and not self.trailing_cr:
+        if not self.buffer:
             return []
 
         lines: list[BytesLike] = [self.buffer]
@@ -76,18 +72,25 @@ class BytesLineDecoder:
 
 
 class SSEDecoder:
-    def __init__(self) -> None:
+    def __init__(self, *, last_event_id: str | None = None) -> None:
         self._event = ""
         self._data = bytearray()
-        self._last_event_id: str | None = None
+        self._last_event_id = last_event_id
         self._retry: int | None = None
         self._at_stream_start = True
+        self._has_data = False
+        self._id_changed = False
+
+    @property
+    def has_data(self) -> bool:
+        """Whether the last decoded block contained a data field."""
+        return self._has_data
 
     @property
     def last_event_id(self) -> str | None:
         """Return the last event identifier that was seen.
 
-        `None` until the stream sends an `id` field, and `""` after an empty
+        `None` until an ID is supplied or received, and `""` after an empty
         one, which resets the reconnect cursor.
         """
 
@@ -101,17 +104,26 @@ class SSEDecoder:
             line = line.removeprefix(b"\xef\xbb\xbf")
 
         if not line:
+            self._has_data = bool(self._data)
             if (
                 not self._event
                 and not self._data
-                and not self._last_event_id
+                and not self._id_changed
                 and self._retry is None
             ):
                 return None
 
+            body = self._data[:-1]
+            data = None
+            if body and not body.isspace():
+                try:
+                    data = orjson.loads(body)
+                except orjson.JSONDecodeError:
+                    data = orjson.loads(body.decode(errors="replace"))
+
             sse = StreamPart(
                 event=self._event,
-                data=orjson.loads(self._data) if self._data else None,  # ty: ignore[invalid-argument-type]
+                data=data,  # ty: ignore[invalid-argument-type]
                 id=self._last_event_id or None,
             )
 
@@ -119,6 +131,7 @@ class SSEDecoder:
             self._event = ""
             self._data = bytearray()
             self._retry = None
+            self._id_changed = False
 
             return sse
 
@@ -131,16 +144,16 @@ class SSEDecoder:
             value = value[1:]
 
         if fieldname == b"event":
-            self._event = value.decode()
+            self._event = value.decode(errors="replace")
         elif fieldname == b"data":
-            if self._data:
-                self._data.extend(b"\n")
             self._data.extend(value)
+            self._data.extend(b"\n")
         elif fieldname == b"id":
             if b"\0" in value:
                 pass
             else:
-                self._last_event_id = value.decode()
+                self._last_event_id = value.decode(errors="replace")
+                self._id_changed = True
         elif fieldname == b"retry":
             with contextlib.suppress(TypeError, ValueError):
                 self._retry = int(value)

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

@@ -15,7 +15,6 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import logging
-import random
 from collections import OrderedDict
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass, field
@@ -23,6 +22,7 @@ from typing import Any
 
 from langchain_protocol import Event, SubscribeParams
 
+from langgraph_sdk._shared.utilities import _reconnect_delay
 from langgraph_sdk.stream.transport import AsyncProtocolTransport, EventStreamHandle
 
 # ---------------------------------------------------------------------------
@@ -275,11 +275,13 @@ class StreamController:
 
     async def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff and jitter for reconnect attempt *attempt*."""
-        base = self._reconnect_backoff_base
-        cap = self._reconnect_backoff_cap
-        delay = min(cap, base * (2**attempt))
-        jitter = random.uniform(0, delay * 0.25)
-        await asyncio.sleep(delay + jitter)
+        await asyncio.sleep(
+            _reconnect_delay(
+                attempt,
+                base=self._reconnect_backoff_base,
+                cap=self._reconnect_backoff_cap,
+            )
+        )
 
     async def _reconnect_shared_stream(self) -> bool:
         """Attempt to reopen the shared stream after a transport drop.

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

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import contextlib
 import logging
-import random
 import threading
 import time
 from dataclasses import dataclass, field
@@ -13,6 +12,7 @@ from typing import Any
 
 from langchain_protocol import Event, SubscribeParams
 
+from langgraph_sdk._shared.utilities import _reconnect_delay
 from langgraph_sdk.stream.errors import (
     is_non_retryable_stream_error,
     stream_error_frame,
@@ -282,11 +282,13 @@ class SyncStreamController:
 
     def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff + jitter before a reconnect attempt."""
-        base = self._reconnect_backoff_base
-        cap = self._reconnect_backoff_cap
-        delay = min(cap, base * (2**attempt))
-        jitter = random.uniform(0, delay * 0.25)
-        time.sleep(delay + jitter)
+        time.sleep(
+            _reconnect_delay(
+                attempt,
+                base=self._reconnect_backoff_base,
+                cap=self._reconnect_backoff_cap,
+            )
+        )
 
     def _reconnect_shared_stream(self) -> bool:
         """Attempt to reopen the shared stream after a transport drop.

libs/sdk-py/tests/conftest.py

@@ -0,0 +1,17 @@
+import asyncio
+import time
+
+import pytest
+
+
+@pytest.fixture
+def no_sleep(monkeypatch):
+    """Record backoff sleeps without actually waiting."""
+    sleeps: list[float] = []
+
+    async def fake_async_sleep(d):
+        sleeps.append(d)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_async_sleep)
+    monkeypatch.setattr(time, "sleep", lambda d: sleeps.append(d))
+    return sleeps

libs/sdk-py/tests/test_auth.py

@@ -1,8 +1,49 @@
+import ast
+import inspect
+
 import pytest
 
 from langgraph_sdk import Auth
 
 
+@pytest.mark.parametrize(
+    "kwargs",
+    [{}, {"actions": "read"}, {"resources": None}],
+)
+def test_global_handler_factory_accepts_optional_resources(kwargs) -> None:
+    auth = Auth()
+    tree = ast.parse(inspect.getsource(type(auth.on)))
+    cls = tree.body[0]
+    assert isinstance(cls, ast.ClassDef)
+    factory_overload = next(
+        node
+        for node in cls.body
+        if isinstance(node, ast.FunctionDef)
+        and node.name == "__call__"
+        and node.decorator_list
+        and any(arg.arg == "resources" for arg in node.args.kwonlyargs)
+    )
+    resources = next(
+        arg for arg in factory_overload.args.kwonlyargs if arg.arg == "resources"
+    )
+    assert resources.annotation is not None
+    assert ast.unparse(resources.annotation) == "str | Sequence[str] | None"
+    default = factory_overload.args.kw_defaults[0]
+    assert isinstance(default, ast.Constant) and default.value is None
+
+    @auth.on(**kwargs)
+    async def handler(ctx, value):
+        del ctx, value
+        return None
+
+    if kwargs.get("actions"):
+        assert auth._handlers == {("*", "read"): [handler]}
+        assert auth._global_handlers == []
+    else:
+        assert auth._global_handlers == [handler]
+        assert auth._handlers == {}
+
+
 def test_handler_multiple_resources_and_actions() -> None:
     auth = Auth()
 

libs/sdk-py/tests/test_client_stream.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Iterator, Sequence
+from collections.abc import Callable, Iterator, Sequence
 from pathlib import Path
 from typing import Any
 
@@ -13,6 +13,7 @@ from langgraph_sdk._async.runs import _wrap_stream_v2
 from langgraph_sdk._shared.utilities import _sse_to_v2_dict
 from langgraph_sdk._sync.runs import _wrap_stream_v2_sync
 from langgraph_sdk.client import HttpClient, SyncHttpClient
+from langgraph_sdk.errors import InternalServerError, NotFoundError
 from langgraph_sdk.schema import (
     CheckpointPayload,
     CheckpointsStreamPart,
@@ -130,6 +131,110 @@ def test_sse_decoder_strips_a_leading_bom():
     assert decoder.decode(b"") == StreamPart(event="values", data=1, id=None)  # ty: ignore[invalid-argument-type]
 
 
+def test_control_blocks_do_not_repeat_after_a_cursor_update():
+    decoder = SSEDecoder()
+    decoder.decode(b"id: cursor")
+    assert decoder.decode(b"") == StreamPart(event="", data=None, id="cursor")  # ty: ignore[invalid-argument-type]
+    decoder.decode(b": heartbeat")
+    assert decoder.decode(b"") is None
+    assert decoder.decode(b"") is None
+    decoder.decode(b"id:")
+    assert decoder.decode(b"") == StreamPart(event="", data=None, id=None)  # ty: ignore[invalid-argument-type]
+    assert decoder.last_event_id == ""
+    assert decoder.decode(b"") is None
+
+
+def test_cr_lines_are_emitted_without_waiting_for_another_chunk():
+    decoder = BytesLineDecoder()
+    assert decoder.decode(b"data: 1\r\r") == [b"data: 1", b""]
+    assert decoder.decode(b"\n") == []
+    assert decoder.decode(b"data: 2\r") == [b"data: 2"]
+    assert decoder.decode(b"") == []
+    assert decoder.decode(b"\n\r") == [b""]
+    assert decoder.flush() == []
+
+
+def test_empty_data_lines_preserve_their_newlines():
+    decoder = SSEDecoder()
+    for line in (b"data", b"data: 1", b"data:"):
+        decoder.decode(line)
+    assert decoder._data == b"\n1\n\n"
+    assert decoder.decode(b"") == StreamPart(event="", data=1, id=None)  # ty: ignore[invalid-argument-type]
+    decoder.decode(b"data:")
+    decoder.decode(b"data:")
+    assert decoder.decode(b"") == StreamPart(event="", data=None, id=None)  # ty: ignore[invalid-argument-type]
+    assert decoder.has_data
+    assert decoder.decode(b"") is None
+    assert not decoder.has_data
+
+
+def test_malformed_utf8_is_replaced_in_fields_and_json_strings():
+    decoder = SSEDecoder()
+    for line in (b"event: \xff", b"id: \xff", b'data: "\xff"'):
+        decoder.decode(line)
+    assert decoder.decode(b"") == StreamPart(event="\ufffd", data="\ufffd", id="\ufffd")  # ty: ignore[invalid-argument-type]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_null_and_empty_data_are_not_dropped(asynchronous: bool):
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            headers={"Content-Type": "text/event-stream"},
+            content=b"data: null\n\ndata:\n\n",
+        )
+
+    transport = httpx.MockTransport(handler)
+    if asynchronous:
+        async with httpx.AsyncClient(
+            transport=transport, base_url="https://example.com"
+        ) as client:
+            parts = [part async for part in HttpClient(client).stream("/stream", "GET")]
+    else:
+        with httpx.Client(
+            transport=transport, base_url="https://example.com"
+        ) as client:
+            parts = list(SyncHttpClient(client).stream("/stream", "GET"))
+    assert parts == [StreamPart(event="", data=None), StreamPart(event="", data=None)]  # ty: ignore[invalid-argument-type]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_cursor_is_inherited_after_reconnect(asynchronous: bool):
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        if len(requests) == 1:
+            stream_cls = AsyncListByteStream if asynchronous else ListByteStream
+            return httpx.Response(
+                200,
+                headers={"Content-Type": "text/event-stream", "Location": "/reconnect"},
+                stream=stream_cls(
+                    [b"id: cursor\nevent: values\ndata: 1\n\n"],
+                    httpx.RemoteProtocolError("incomplete chunked read"),
+                ),
+            )
+        assert (b"Last-Event-ID", b"cursor") in request.headers.raw
+        return httpx.Response(
+            200,
+            headers={"Content-Type": "text/event-stream"},
+            content=b"event: values\ndata: 2\n\n",
+        )
+
+    transport = httpx.MockTransport(handler)
+    if asynchronous:
+        async with httpx.AsyncClient(
+            transport=transport, base_url="https://example.com"
+        ) as client:
+            parts = [part async for part in HttpClient(client).stream("/stream", "GET")]
+    else:
+        with httpx.Client(
+            transport=transport, base_url="https://example.com"
+        ) as client:
+            parts = list(SyncHttpClient(client).stream("/stream", "GET"))
+    assert [part.id for part in parts] == ["cursor", "cursor"]
+
+
 # --- HTTP client streaming ---
 
 
@@ -176,6 +281,7 @@ def test_sync_http_client_stream_flushes_trailing_event():
     assert parts == [StreamPart(event="foo", data={"bar": 1})]
 
 
+@pytest.mark.usefixtures("no_sleep")
 def test_sync_http_client_stream_recovers_after_disconnect():
     reconnect_path = "/reconnect"
     first_chunks = [
@@ -241,6 +347,7 @@ def test_sync_http_client_stream_recovers_after_disconnect():
 
 
 @pytest.mark.asyncio
+@pytest.mark.usefixtures("no_sleep")
 async def test_http_client_stream_recovers_after_disconnect():
     reconnect_path = "/reconnect"
     first_chunks = [
@@ -312,18 +419,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 +439,376 @@ 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
+@pytest.mark.usefixtures("no_sleep")
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    with pytest.warns(UserWarning, match="Last-Event-ID"):
+        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
+@pytest.mark.usefixtures("no_sleep")
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    with pytest.warns(UserWarning, match="Last-Event-ID"):
+        reconnect = await _async_reconnect(
+            [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+        )
+
+    assert "last-event-id" not in reconnect.headers
+
+
+@pytest.mark.usefixtures("no_sleep")
+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"]
+
+
+@pytest.mark.usefixtures("no_sleep")
+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"]
+
+
+@pytest.mark.usefixtures("no_sleep")
+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"]
+
+
+@pytest.mark.usefixtures("no_sleep")
+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"]
+
+
+# A reconnect that sends no Last-Event-ID can't replay what was sent while it was
+# disconnected, so it warns.
+_RECONNECT_CURSORS = pytest.mark.parametrize(
+    ("chunks", "headers", "warns"),
+    [
+        ([b"event: values\ndata: 1\n\n"], None, True),
+        ([b"id: 1\nevent: values\ndata: 1\n\n"], None, False),
+        ([b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"}, False),
+    ],
+    ids=["no-cursor", "stream-cursor", "caller-cursor"],
+)
+
+
+@_RECONNECT_CURSORS
+@pytest.mark.usefixtures("no_sleep")
+def test_sync_reconnect_warns_only_without_a_cursor(
+    chunks: list[bytes],
+    headers: dict[str, str] | None,
+    warns: bool,
+    recwarn: pytest.WarningsRecorder,
+):
+    _sync_reconnect(chunks, headers)
+
+    assert any("Last-Event-ID" in str(w.message) for w in recwarn) is warns
+
+
+@_RECONNECT_CURSORS
+@pytest.mark.usefixtures("no_sleep")
+async def test_async_reconnect_warns_only_without_a_cursor(
+    chunks: list[bytes],
+    headers: dict[str, str] | None,
+    warns: bool,
+    recwarn: pytest.WarningsRecorder,
+):
+    await _async_reconnect(chunks, headers)
+
+    assert any("Last-Event-ID" in str(w.message) for w in recwarn) is warns
+
+
+# --- reconnects survive a server restart or a deploy ---
+
+SSE = {"Content-Type": "text/event-stream"}
+DROP = httpx.RemoteProtocolError("incomplete chunked read")
+
+
+def _event(i: int) -> bytes:
+    return f"id: {i}\nevent: values\ndata: {i}\n\n".encode()
+
+
+def _scripted(steps: list[Callable[[httpx.Request], httpx.Response]]):
+    """Handler that plays one step per request and records each request."""
+    seen: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        if len(seen) > len(steps):
+            raise AssertionError("unexpected request")
+        return steps[len(seen) - 1](request)
+
+    return handler, seen
+
+
+def _ok(stream_cls, chunks, *, drop=False, location=None):
+    headers = dict(SSE)
+    if location:
+        headers["Location"] = location
+
+    def step(_request):
+        return httpx.Response(
+            200, headers=headers, stream=stream_cls(chunks, DROP if drop else None)
+        )
+
+    return step
+
+
+def _status(code):
+    return lambda request: httpx.Response(
+        code, json={"detail": "upstream"}, request=request
+    )
+
+
+def _connect_error(request):
+    raise httpx.ConnectError("[Errno 61] Connection refused", request=request)
+
+
+def _not_sse(request):
+    return httpx.Response(
+        200, headers={"Content-Type": "text/html"}, text="<html>", request=request
+    )
+
+
+def _transient_steps(stream_cls, failure):
+    """POST drops after event 1; first reconnect fails; second succeeds."""
+    return [
+        _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+        failure,
+        _ok(stream_cls, [_event(2), b"event: end\ndata: null\n\n"]),
+    ]
+
+
+def _long_stream_steps(stream_cls, drops):
+    """POST + `drops` reconnects, each delivering one event then dropping."""
+    steps = [_ok(stream_cls, [_event(0)], drop=True, location="/reconnect")]
+    for i in range(1, drops):
+        steps.append(_ok(stream_cls, [_event(i)], drop=True))
+    steps.append(_ok(stream_cls, [_event(drops), b"event: end\ndata: null\n\n"]))
+    return steps
+
+
+def _run_sync(steps):
+    handler, seen = _scripted(steps)
+    with httpx.Client(
+        transport=httpx.MockTransport(handler), base_url="https://example.com"
+    ) as client:
+        parts = list(SyncHttpClient(client).stream("/stream", "POST", json={}))
+    return parts, seen
+
+
+async def _run_async(steps):
+    sync_handler, seen = _scripted(steps)
+
+    async def handler(request):
+        return sync_handler(request)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="https://example.com"
+    ) as client:
+        parts = [p async for p in HttpClient(client).stream("/stream", "POST", json={})]
+    return parts, seen
+
+
+FAILURES = {
+    "429": _status(429),
+    "502": _status(502),
+    "503": _status(503),
+    "connect_error": _connect_error,
+    "not_sse": _not_sse,
+}
+
+
+# A reconnect that hits a 429 / 502 / 503 / connect error / non-SSE body, then
+# succeeds, should recover.
+
+
+@pytest.mark.usefixtures("no_sleep")
+@pytest.mark.parametrize("failure", FAILURES.values(), ids=FAILURES.keys())
+def test_sync_reconnect_recovers_from_transient_failure(failure):
+    parts, seen = _run_sync(_transient_steps(ListByteStream, failure))
+    assert [p.event for p in parts] == ["values", "values", "end"]
+    assert len(seen) == 3
+
+
+@pytest.mark.usefixtures("no_sleep")
+@pytest.mark.parametrize("failure", FAILURES.values(), ids=FAILURES.keys())
+async def test_async_reconnect_recovers_from_transient_failure(failure):
+    parts, seen = await _run_async(_transient_steps(AsyncListByteStream, failure))
+    assert [p.event for p in parts] == ["values", "values", "end"]
+    assert len(seen) == 3
+
+
+# Any other 4xx on a reconnect won't go away by retrying, so it raises.
+
+
+def _client_error_steps(stream_cls):
+    return [
+        _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+        _status(404),
+    ]
+
+
+@pytest.mark.usefixtures("no_sleep")
+def test_sync_reconnect_does_not_retry_a_client_error():
+    with pytest.raises(NotFoundError):
+        _run_sync(_client_error_steps(ListByteStream))
+
+
+@pytest.mark.usefixtures("no_sleep")
+async def test_async_reconnect_does_not_retry_a_client_error():
+    with pytest.raises(NotFoundError):
+        await _run_async(_client_error_steps(AsyncListByteStream))
+
+
+# The first request isn't a reconnect, so it raises even when it names a Location.
+
+
+def _first_request_503(request):
+    return httpx.Response(
+        503,
+        headers={"Location": "/reconnect"},
+        json={"detail": "upstream"},
+        request=request,
+    )
+
+
+@pytest.mark.usefixtures("no_sleep")
+def test_sync_first_request_failure_is_not_retried():
+    with pytest.raises(InternalServerError):
+        _run_sync([_first_request_503])
+
+
+@pytest.mark.usefixtures("no_sleep")
+async def test_async_first_request_failure_is_not_retried():
+    with pytest.raises(InternalServerError):
+        await _run_async([_first_request_503])
+
+
+# A server that stays down gets five reconnects, backing off 1, 2, 4, 5 and 5s
+# (each plus up to 25% jitter), so about 20s, before the stream gives up. The
+# error it gives up with is caused by the last failure, so callers see why.
+
+GIVE_UP = "Exceeded maximum SSE reconnection attempts"
+
+
+def _server_down_steps(stream_cls):
+    return [
+        _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+        *[_status(502)] * 5,
+    ]
+
+
+def _assert_backed_off_for_about_20s(sleeps: list[float]) -> None:
+    assert len(sleeps) == 5
+    for slept, delay in zip(sleeps, [1, 2, 4, 5, 5], strict=True):
+        assert delay <= slept <= delay * 1.25
+
+
+def test_sync_reconnect_gives_up_after_five_backed_off_attempts(no_sleep):
+    with pytest.raises(httpx.TransportError, match=GIVE_UP) as exc_info:
+        _run_sync(_server_down_steps(ListByteStream))
+    assert isinstance(exc_info.value.__cause__, InternalServerError)
+    _assert_backed_off_for_about_20s(no_sleep)
+
+
+async def test_async_reconnect_gives_up_after_five_backed_off_attempts(no_sleep):
+    with pytest.raises(httpx.TransportError, match=GIVE_UP) as exc_info:
+        await _run_async(_server_down_steps(AsyncListByteStream))
+    assert isinstance(exc_info.value.__cause__, InternalServerError)
+    _assert_backed_off_for_about_20s(no_sleep)
+
+
+# A long stream that drops 6 times, each reconnect succeeding and making
+# progress, should not run out of attempts.
+
+
+@pytest.mark.usefixtures("no_sleep")
+def test_sync_long_stream_survives_many_spaced_out_drops():
+    parts, _ = _run_sync(_long_stream_steps(ListByteStream, drops=6))
+    assert [p.data for p in parts] == [0, 1, 2, 3, 4, 5, 6, None]
+
+
+@pytest.mark.usefixtures("no_sleep")
+async def test_async_long_stream_survives_many_spaced_out_drops():
+    parts, _ = await _run_async(_long_stream_steps(AsyncListByteStream, drops=6))
+    assert [p.data for p in parts] == [0, 1, 2, 3, 4, 5, 6, None]
+
+
+# Reconnects should back off rather than hammer the server.
+
+
+def test_sync_reconnect_backs_off(no_sleep):
+    _run_sync(_long_stream_steps(ListByteStream, drops=2))
+    assert len(no_sleep) == 2 and all(d > 0 for d in no_sleep)
+
+
+async def test_async_reconnect_backs_off(no_sleep):
+    await _run_async(_long_stream_steps(AsyncListByteStream, drops=2))
+    assert len(no_sleep) == 2 and all(d > 0 for d in no_sleep)
 
 
 # --- _sse_to_v2_dict conversion ---

libs/sdk-py/tests/test_runs_client.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import json
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Callable, Iterator, Sequence
+from typing import Any
 
 import httpx
 import pytest
@@ -24,7 +25,16 @@ class _BodyDropsMidRead(httpx.SyncByteStream, httpx.AsyncByteStream):
         raise httpx.ReadError("connection dropped")
 
 
-def _wait_handler(requests: list[httpx.Request]):
+def _done(_request: httpx.Request) -> httpx.Response:
+    return httpx.Response(200, json={"done": True})
+
+
+def _wait_handler(
+    requests: list[httpx.Request],
+    reconnects: Sequence[Callable[[httpx.Request], httpx.Response]] = (_done,),
+):
+    """POST drops mid-body; the reconnects answer from `reconnects` in order."""
+
     def handler(request: httpx.Request) -> httpx.Response:
         requests.append(request)
         if request.method == "POST":
@@ -33,7 +43,7 @@ def _wait_handler(requests: list[httpx.Request]):
                 headers={"Location": "/threads/t1/runs/r1/join"},
                 stream=_BodyDropsMidRead(),
             )
-        return httpx.Response(200, json={"done": True})
+        return reconnects[len(requests) - 2](request)
 
     return handler
 
@@ -46,6 +56,7 @@ def _assert_bodyless_get(request: httpx.Request) -> None:
     assert request.read() == b""
 
 
+@pytest.mark.usefixtures("no_sleep")
 def test_sync_wait_reconnects_with_a_bodyless_get():
     requests: list[httpx.Request] = []
     transport = httpx.MockTransport(_wait_handler(requests))
@@ -58,6 +69,7 @@ def test_sync_wait_reconnects_with_a_bodyless_get():
     _assert_bodyless_get(requests[1])
 
 
+@pytest.mark.usefixtures("no_sleep")
 async def test_async_wait_reconnects_with_a_bodyless_get():
     requests: list[httpx.Request] = []
     transport = httpx.MockTransport(_wait_handler(requests))
@@ -72,6 +84,89 @@ async def test_async_wait_reconnects_with_a_bodyless_get():
     _assert_bodyless_get(requests[1])
 
 
+def _connect_error(request: httpx.Request) -> httpx.Response:
+    raise httpx.ConnectError("[Errno 61] Connection refused", request=request)
+
+
+async def _wait(
+    handler: Callable[[httpx.Request], httpx.Response], *, asynchronous: bool
+) -> Any:
+    transport = httpx.MockTransport(handler)
+    if asynchronous:
+        async with httpx.AsyncClient(
+            transport=transport, base_url="https://example.com"
+        ) as client:
+            return await RunsClient(HttpClient(client)).wait("t1", "agent", input={})
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        return SyncRunsClient(SyncHttpClient(client)).wait("t1", "agent", input={})
+
+
+# What a server restart or a deploy looks like to the reconnect GET: each one is
+# retried after a backoff.
+RECONNECT_FAILURES = {
+    "429": lambda request: httpx.Response(429, request=request),
+    "503": lambda request: httpx.Response(503, request=request),
+    "connect_error": _connect_error,
+    "body_drops": lambda _request: httpx.Response(
+        200,
+        headers={"Location": "/threads/t1/runs/r1/join"},
+        stream=_BodyDropsMidRead(),
+    ),
+}
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize(
+    "failure", RECONNECT_FAILURES.values(), ids=RECONNECT_FAILURES.keys()
+)
+async def test_wait_retries_a_failed_reconnect_after_a_backoff(
+    failure, asynchronous, no_sleep
+):
+    requests: list[httpx.Request] = []
+    with pytest.warns(UserWarning, match="attempting reconnect"):
+        result = await _wait(
+            _wait_handler(requests, [failure, _done]), asynchronous=asynchronous
+        )
+
+    assert result == {"done": True}
+    assert len(requests) == 3
+    assert len(no_sleep) == 2 and min(no_sleep) >= 1, no_sleep
+
+
+@pytest.mark.usefixtures("no_sleep")
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_wait_raises_after_five_failed_reconnects(asynchronous):
+    requests: list[httpx.Request] = []
+    with (
+        pytest.warns(UserWarning, match="attempting reconnect"),
+        pytest.raises(httpx.HTTPStatusError, match="503"),
+    ):
+        await _wait(
+            _wait_handler(requests, [RECONNECT_FAILURES["503"]] * 5),
+            asynchronous=asynchronous,
+        )
+
+    assert len(requests) == 6
+
+
+@pytest.mark.usefixtures("no_sleep")
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_wait_does_not_retry_a_client_error(asynchronous):
+    requests: list[httpx.Request] = []
+    with (
+        pytest.warns(UserWarning, match="attempting reconnect"),
+        pytest.raises(httpx.HTTPStatusError, match="404"),
+    ):
+        await _wait(
+            _wait_handler(
+                requests, [lambda request: httpx.Response(404, request=request)]
+            ),
+            asynchronous=asynchronous,
+        )
+
+    assert len(requests) == 2
+
+
 def test_sync_wait_raises_the_run_error():
     requests: list[httpx.Request] = []
 

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