fix(cli): include build environment hash in Docker cache keys

libs/checkpoint-conformance/uv.lock

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

libs/checkpoint-postgres/uv.lock

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

libs/checkpoint-sqlite/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.3"
+version = "3.1.2"
 description = "Library with a SQLite implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"
@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-    "langgraph-checkpoint>=4.3.1,<5.0.0",
+    "langgraph-checkpoint>=4.3.0,<5.0.0",
     "aiosqlite>=0.20",
     "sqlite-vec>=0.1.6",
 ]

libs/checkpoint-sqlite/uv.lock

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

libs/checkpoint/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint"
-version = "4.3.1"
+version = "4.3.0"
 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.1"
+version = "4.3.0"
 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.6.7"
-  resolved "https://registry.yarnpkg.com/zod/-/zod-4.6.7.tgz#e806cf64a7f4e37635ce9f2f52c04c3766f75e5b"
-  integrity sha512-g8W2D2kU6a8untc053fX1uVBKzM7yqAxW/3zwk83+6RyF2+SKOuV+c0wBDhUwzV1PaGapmZH+rmcUzPf+3CGwg==
+  version "4.3.6"
+  resolved "https://registry.yarnpkg.com/zod/-/zod-4.3.6.tgz#89c56e0aa7d2b05107d894412227087885ab112a"
+  integrity sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==

libs/cli/langgraph_cli/config.py

@@ -1589,6 +1589,8 @@ 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"]),
             "",
@@ -1679,6 +1681,8 @@ 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,6 +955,8 @@ 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,6 +148,7 @@ 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,6 +910,7 @@ 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
@@ -974,6 +975,7 @@ 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 && \\
@@ -1034,6 +1036,7 @@ 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
@@ -1097,6 +1100,7 @@ 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 && \\
@@ -1144,6 +1148,7 @@ 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 . --
@@ -1177,6 +1182,7 @@ 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
@@ -1222,6 +1228,7 @@ 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
@@ -1413,6 +1420,7 @@ 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
@@ -1524,6 +1532,7 @@ 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
@@ -1571,6 +1580,7 @@ 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
@@ -2843,6 +2853,7 @@ 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 && \\
@@ -2884,6 +2895,7 @@ 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 && \\
@@ -2929,6 +2941,7 @@ 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 && \\
@@ -2967,6 +2980,7 @@ 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 && \\
@@ -3014,6 +3028,7 @@ 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

@@ -0,0 +1,142 @@
+"""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/_internal/_runnable.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import copy
 import enum
 import inspect
 import logging
@@ -24,16 +23,9 @@ from typing import (
     Optional,
     Protocol,
     TypeGuard,
-    TypeVar,
     cast,
 )
 
-from langchain_core.callbacks.base import BaseCallbackHandler, BaseCallbackManager
-from langchain_core.callbacks.manager import (
-    AsyncCallbackManagerForChainRun,
-    BaseRunManager,
-    CallbackManagerForChainRun,
-)
 from langchain_core.runnables.base import (
     Runnable,
     RunnableConfig,
@@ -49,7 +41,6 @@ from langchain_core.runnables.config import (
     var_child_runnable_config,
 )
 from langchain_core.runnables.utils import Input, Output
-from langchain_core.tracers.base import BaseTracer
 from langchain_core.tracers.langchain import LangChainTracer
 from langgraph.store.base import BaseStore
 
@@ -66,7 +57,7 @@ from langgraph._internal._constants import (
 )
 from langgraph._internal._typing import MISSING
 from langgraph.errors import NodeError
-from langgraph.types import StreamWriter, TracePolicy
+from langgraph.types import StreamWriter
 
 try:
     from langchain_core.tracers._streaming import _StreamingCallbackHandler
@@ -94,29 +85,6 @@ def _trace_payload(value: Any, transform: Callable[[Any], Any] | None) -> Any:
         return value
 
 
-def _is_stream_handler(handler: BaseCallbackHandler) -> bool:
-    # Inheritance, not isinstance on the runtime-checkable protocol, which also
-    # matches any exporter that merely has the tap methods: stream handlers get a
-    # node's real payload and its streamed output.
-    return _StreamingCallbackHandler in type(handler).__mro__
-
-
-_M = TypeVar("_M", bound=BaseCallbackManager | BaseRunManager)
-
-
-def _split_handlers(manager: _M) -> tuple[_M, _M]:
-    # The recorded payload exists to keep tracer ingest small, so only tracers get it.
-    # `astream_log`'s handler is a tracer too, but it feeds the caller's own stream.
-    real, tracers = copy.copy(manager), copy.copy(manager)
-    tracers.handlers = [
-        h
-        for h in manager.handlers
-        if isinstance(h, BaseTracer) and not _is_stream_handler(h)
-    ]
-    real.handlers = [h for h in manager.handlers if h not in tracers.handlers]
-    return real, tracers
-
-
 def _set_config_context(
     config: RunnableConfig, run: Any = None
 ) -> Token[RunnableConfig | None]:
@@ -625,15 +593,13 @@ class RunnableSeq(Runnable):
         *steps: RunnableLike,
         name: str | None = None,
         trace_inputs: Callable[[Any], Any] | None = None,
-        trace_policy: TracePolicy | None = None,
+        trace_outputs: Callable[[Any], Any] | None = None,
     ) -> None:
         """Create a new RunnableSeq.
 
         Args:
             steps: The steps to include in the sequence.
             name: The name of the `Runnable`.
-            trace_inputs: Transforms the input every callback handler receives.
-            trace_policy: Processes what tracers record, after `trace_inputs`.
 
         Raises:
             ValueError: If the sequence has less than 2 steps.
@@ -653,67 +619,7 @@ class RunnableSeq(Runnable):
         self.steps = steps_flat
         self.name = name
         self.trace_inputs = trace_inputs
-        self.trace_policy = trace_policy or TracePolicy()
-
-    def _start(self, input: Any, config: RunnableConfig) -> CallbackManagerForChainRun:
-        callback_manager = get_callback_manager_for_config(config)
-        payload = _trace_payload(input, self.trace_inputs)
-        recorded = _trace_payload(payload, self.trace_policy.process_inputs)
-        name = config.get("run_name") or self.get_name()
-        run_id = config.pop("run_id", None)
-        if recorded is payload:
-            return callback_manager.on_chain_start(
-                None, payload, name=name, run_id=run_id
-            )
-        real, tracers = _split_handlers(callback_manager)
-        run_manager = real.on_chain_start(None, payload, name=name, run_id=run_id)
-        tracers.on_chain_start(None, recorded, name=name, run_id=run_manager.run_id)
-        run_manager.handlers = callback_manager.handlers
-        return run_manager
-
-    async def _astart(
-        self, input: Any, config: RunnableConfig
-    ) -> AsyncCallbackManagerForChainRun:
-        callback_manager = get_async_callback_manager_for_config(config)
-        payload = _trace_payload(input, self.trace_inputs)
-        recorded = _trace_payload(payload, self.trace_policy.process_inputs)
-        name = config.get("run_name") or self.get_name()
-        run_id = config.pop("run_id", None)
-        if recorded is payload:
-            return await callback_manager.on_chain_start(
-                None, payload, name=name, run_id=run_id
-            )
-        real, tracers = _split_handlers(callback_manager)
-        # Tracers start last, so a cancellation before then leaves no tracer run open.
-        run_manager = await real.on_chain_start(None, payload, name=name, run_id=run_id)
-        await tracers.on_chain_start(
-            None, recorded, name=name, run_id=run_manager.run_id
-        )
-        run_manager.handlers = callback_manager.handlers
-        return run_manager
-
-    def _end(self, run_manager: CallbackManagerForChainRun, output: Any) -> None:
-        recorded = _trace_payload(output, self.trace_policy.process_outputs)
-        if recorded is output:
-            run_manager.on_chain_end(output)
-        else:
-            real, tracers = _split_handlers(run_manager)
-            tracers.on_chain_end(recorded)
-            real.on_chain_end(output)
-
-    async def _aend(
-        self, run_manager: AsyncCallbackManagerForChainRun, output: Any
-    ) -> None:
-        recorded = _trace_payload(output, self.trace_policy.process_outputs)
-        if recorded is output:
-            await run_manager.on_chain_end(output)
-        else:
-            real, tracers = _split_handlers(run_manager)
-            # Tracers end first: on_chain_end is shielded once it starts, so a
-            # cancellation can only skip the other handlers' end, never leave a
-            # tracer run open.
-            await tracers.on_chain_end(recorded)
-            await real.on_chain_end(output)
+        self.trace_outputs = trace_outputs
 
     def __or__(
         self,
@@ -770,8 +676,15 @@ class RunnableSeq(Runnable):
     ) -> Any:
         if config is None:
             config = ensure_config()
+        # setup callbacks and context
+        callback_manager = get_callback_manager_for_config(config)
         # start the root run
-        run_manager = self._start(input, config)
+        run_manager = callback_manager.on_chain_start(
+            None,
+            _trace_payload(input, self.trace_inputs),
+            name=config.get("run_name") or self.get_name(),
+            run_id=config.pop("run_id", None),
+        )
         # invoke all steps in sequence
         try:
             for i, step in enumerate(self.steps):
@@ -799,7 +712,7 @@ class RunnableSeq(Runnable):
             run_manager.on_chain_error(e)
             raise
         else:
-            self._end(run_manager, input)
+            run_manager.on_chain_end(_trace_payload(input, self.trace_outputs))
             return input
 
     async def ainvoke(
@@ -810,8 +723,15 @@ class RunnableSeq(Runnable):
     ) -> Any:
         if config is None:
             config = ensure_config()
+        # setup callbacks
+        callback_manager = get_async_callback_manager_for_config(config)
         # start the root run
-        run_manager = await self._astart(input, config)
+        run_manager = await callback_manager.on_chain_start(
+            None,
+            _trace_payload(input, self.trace_inputs),
+            name=config.get("run_name") or self.get_name(),
+            run_id=config.pop("run_id", None),
+        )
 
         # invoke all steps in sequence
         try:
@@ -845,7 +765,7 @@ class RunnableSeq(Runnable):
             await run_manager.on_chain_error(e)
             raise
         else:
-            await self._aend(run_manager, input)
+            await run_manager.on_chain_end(_trace_payload(input, self.trace_outputs))
             return input
 
     def stream(
@@ -856,8 +776,15 @@ class RunnableSeq(Runnable):
     ) -> Iterator[Any]:
         if config is None:
             config = ensure_config()
+        # setup callbacks
+        callback_manager = get_callback_manager_for_config(config)
         # start the root run
-        run_manager = self._start(input, config)
+        run_manager = callback_manager.on_chain_start(
+            None,
+            _trace_payload(input, self.trace_inputs),
+            name=config.get("run_name") or self.get_name(),
+            run_id=config.pop("run_id", None),
+        )
         # get the run object
         for h in run_manager.handlers:
             if isinstance(h, LangChainTracer):
@@ -887,9 +814,10 @@ class RunnableSeq(Runnable):
                         )
                         iterator = step.transform(iterator, config)
                 # populates streamed_output in astream_log() output if needed
-                for h in run_manager.handlers:
-                    if _is_stream_handler(h):
-                        iterator = h.tap_output_iter(run_manager.run_id, iterator)
+                if _StreamingCallbackHandler is not None:
+                    for h in run_manager.handlers:
+                        if isinstance(h, _StreamingCallbackHandler):
+                            iterator = h.tap_output_iter(run_manager.run_id, iterator)
                 # consume into final output
                 output = context.run(_consume_iter, iterator)
                 # sequence doesn't emit output, yield to mark as generator
@@ -898,7 +826,7 @@ class RunnableSeq(Runnable):
                 run_manager.on_chain_error(e)
                 raise
             else:
-                self._end(run_manager, output)
+                run_manager.on_chain_end(_trace_payload(output, self.trace_outputs))
 
     async def astream(
         self,
@@ -908,8 +836,15 @@ class RunnableSeq(Runnable):
     ) -> AsyncIterator[Any]:
         if config is None:
             config = ensure_config()
+        # setup callbacks
+        callback_manager = get_async_callback_manager_for_config(config)
         # start the root run
-        run_manager = await self._astart(input, config)
+        run_manager = await callback_manager.on_chain_start(
+            None,
+            _trace_payload(input, self.trace_inputs),
+            name=config.get("run_name") or self.get_name(),
+            run_id=config.pop("run_id", None),
+        )
         # stream the last steps
         # transform the input stream of each step with the next
         # steps that don't natively support transforming an input stream will
@@ -945,11 +880,12 @@ class RunnableSeq(Runnable):
                             if hasattr(aiterator, "aclose"):
                                 stack.push_async_callback(aiterator.aclose)
                         # populates streamed_output in astream_log() output if needed
-                        for h in run_manager.handlers:
-                            if _is_stream_handler(h):
-                                aiterator = h.tap_output_aiter(
-                                    run_manager.run_id, aiterator
-                                )
+                        if _StreamingCallbackHandler is not None:
+                            for h in run_manager.handlers:
+                                if isinstance(h, _StreamingCallbackHandler):
+                                    aiterator = h.tap_output_aiter(
+                                        run_manager.run_id, aiterator
+                                    )
                         # consume into final output
                         output = await asyncio.create_task(
                             _consume_aiter(aiterator), context=context
@@ -960,7 +896,9 @@ class RunnableSeq(Runnable):
                     await run_manager.on_chain_error(e)
                     raise
                 else:
-                    await self._aend(run_manager, output)
+                    await run_manager.on_chain_end(
+                        _trace_payload(output, self.trace_outputs)
+                    )
         else:
             try:
                 async with AsyncExitStack() as stack:
@@ -976,11 +914,12 @@ class RunnableSeq(Runnable):
                         if hasattr(aiterator, "aclose"):
                             stack.push_async_callback(aiterator.aclose)
                     # populates streamed_output in astream_log() output if needed
-                    for h in run_manager.handlers:
-                        if _is_stream_handler(h):
-                            aiterator = h.tap_output_aiter(
-                                run_manager.run_id, aiterator
-                            )
+                    if _StreamingCallbackHandler is not None:
+                        for h in run_manager.handlers:
+                            if isinstance(h, _StreamingCallbackHandler):
+                                aiterator = h.tap_output_aiter(
+                                    run_manager.run_id, aiterator
+                                )
                     # consume into final output
                     output = await _consume_aiter(aiterator)
                     # sequence doesn't emit output, yield to mark as generator
@@ -989,7 +928,9 @@ class RunnableSeq(Runnable):
                 await run_manager.on_chain_error(e)
                 raise
             else:
-                await self._aend(run_manager, output)
+                await run_manager.on_chain_end(
+                    _trace_payload(output, self.trace_outputs)
+                )
 
 
 def _consume_iter(it: Iterator[Any]) -> Any:

libs/langgraph/langgraph/graph/message.py

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

libs/langgraph/langgraph/pregel/_draw.py

@@ -228,10 +228,6 @@ 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/_loop.py

@@ -132,7 +132,7 @@ from langgraph.pregel.debug import (
     map_debug_tasks,
 )
 from langgraph.pregel.protocol import StreamChunk, StreamProtocol
-from langgraph.runtime import RunControl, Runtime, _RunExits
+from langgraph.runtime import RunControl, Runtime
 from langgraph.types import (
     All,
     CachePolicy,
@@ -1497,10 +1497,6 @@ class PregelLoop:
             if self._delta_write_futs is not None:
                 self._delta_write_futs.append(fut)
 
-    def _run_exits(self) -> _RunExits | None:
-        runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
-        return runtime._run_exits if isinstance(runtime, Runtime) else None
-
     def _suppress_interrupt(
         self,
         exc_type: type[BaseException] | None,
@@ -1878,10 +1874,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )
-        # Bottom of the stack, so run-exit callbacks run last: after the
-        # executor has stopped every task and the final checkpoint is saved.
-        if (exits := self._run_exits()) is not None and exits.bind(is_async=False):
-            self.stack.push(exits.exit)
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs,
@@ -2138,10 +2130,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )
-        # Bottom of the stack, so run-exit callbacks run last: after the
-        # executor has stopped every task and the final checkpoint is saved.
-        if (exits := self._run_exits()) is not None and exits.bind(is_async=True):
-            self.stack.push_async_exit(exits.aexit)
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )

libs/langgraph/langgraph/pregel/_messages.py

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

libs/langgraph/langgraph/pregel/_read.py

@@ -227,14 +227,23 @@ class PregelNode:
     def node(self) -> Runnable[Any, Any] | None:
         """Get a runnable that combines `bound` and `writers`."""
         writers = self.flat_writers
+        trace_inputs = self.trace_policy.process_inputs if self.trace_policy else None
+        trace_outputs = self.trace_policy.process_outputs if self.trace_policy else None
         if self.bound is DEFAULT_BOUND and not writers:
             return None
         elif self.bound is DEFAULT_BOUND and len(writers) == 1:
             return writers[0]
         elif self.bound is DEFAULT_BOUND:
-            return RunnableSeq(*writers, trace_policy=self.trace_policy)
+            return RunnableSeq(
+                *writers, trace_inputs=trace_inputs, trace_outputs=trace_outputs
+            )
         elif writers:
-            return RunnableSeq(self.bound, *writers, trace_policy=self.trace_policy)
+            return RunnableSeq(
+                self.bound,
+                *writers,
+                trace_inputs=trace_inputs,
+                trace_outputs=trace_outputs,
+            )
         else:
             return self.bound
 

libs/langgraph/langgraph/pregel/_retry.py

@@ -34,7 +34,6 @@ 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
@@ -734,14 +733,15 @@ async def arun_with_retry(
                 if stream:
                     async for _ in task.proc.astream(task.input, config):
                         pass
-                    return _return_value(task)
+                    break
                 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:
-                return _return_value(task)
+                # if successful, end
+                break
             return result
         except ParentCommand as exc:
             ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
@@ -831,12 +831,6 @@ 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

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

@@ -97,7 +97,6 @@ from langgraph._internal._runnable import (
     Runnable,
     RunnableLike,
     RunnableSeq,
-    _is_stream_handler,
     coerce_to_runnable,
 )
 from langgraph._internal._timeout import coerce_timeout_policy
@@ -175,7 +174,6 @@ from langgraph.runtime import (
     RunControl,
     Runtime,
     ServerInfo,
-    _RunExits,
 )
 from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import StreamTransformer
@@ -205,6 +203,11 @@ from langgraph.types import (
 from langgraph.typing import ContextT, InputT, OutputT, StateT
 from langgraph.warnings import LangGraphDeprecatedSinceV10
 
+try:
+    from langchain_core.tracers._streaming import _StreamingCallbackHandler
+except ImportError:
+    _StreamingCallbackHandler = None  # type: ignore
+
 __all__ = ("NodeBuilder", "Pregel")
 
 _WriteValue = Callable[[Input], Output] | Any
@@ -2877,20 +2880,6 @@ class Pregel(
                     for h in callback_manager.inheritable_handlers
                     if not isinstance(h, StreamMessagesHandlerV2)
                 ]
-            if CONFIG_KEY_TASK_ID not in config[CONF]:
-                # Not run as a task (a child invoked with its own thread, say),
-                # this graph's namespace starts at the root, so a caller's
-                # handler would take this graph's events for its own graph's.
-                callback_manager.handlers = [
-                    h
-                    for h in callback_manager.handlers
-                    if not _is_stream_mode_handler(h)
-                ]
-                callback_manager.inheritable_handlers = [
-                    h
-                    for h in callback_manager.inheritable_handlers
-                    if not _is_stream_mode_handler(h)
-                ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
             run_manager = callback_manager.on_chain_start(
@@ -2909,9 +2898,9 @@ class Pregel(
                 )
             # set up subgraph checkpointing
             config = self._own_checkpoint_config(config)
-            stream_handlers: list[StreamMessagesHandler | StreamToolCallHandler] = []
             # set up messages stream mode
             if "messages" in stream_modes:
+                ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
                 use_stream_messages_v2 = bool(
                     version == "v2" and config[CONF].get(CONFIG_KEY_STREAM_MESSAGES_V2)
                 )
@@ -2920,12 +2909,24 @@ class Pregel(
                     if use_stream_messages_v2
                     else StreamMessagesHandler
                 )
-                stream_handlers.append(messages_handler_cls(stream.put, subgraphs))
+                run_manager.inheritable_handlers.append(
+                    messages_handler_cls(
+                        stream.put,
+                        subgraphs,
+                        parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
+                    )
+                )
 
             # set up tools stream mode
             if "tools" in stream_modes:
-                stream_handlers.append(StreamToolCallHandler(stream.put, subgraphs))
-            run_manager.inheritable_handlers.extend(stream_handlers)
+                ns_tools = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
+                run_manager.inheritable_handlers.append(
+                    StreamToolCallHandler(
+                        stream.put,
+                        subgraphs,
+                        parent_ns=tuple(ns_tools.split(NS_SEP)) if ns_tools else None,
+                    )
+                )
 
             # set up custom stream mode
             if "custom" in stream_modes:
@@ -2967,8 +2968,6 @@ class Pregel(
                 execution_info=None,
                 server_info=server_info,
                 control=control or parent_runtime.control or RunControl(),
-                # a subgraph shares the root run's exit callbacks
-                _run_exits=parent_runtime._run_exits or _RunExits(),
             )
             runtime = parent_runtime.merge(runtime)
             config[CONF][CONFIG_KEY_RUNTIME] = runtime
@@ -3010,10 +3009,6 @@ class Pregel(
                 cache_policy=self.cache_policy,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
-                # Known only now: the loop numbers the namespace of every
-                # subgraph a task runs after its first.
-                for handler in stream_handlers:
-                    handler.parent_ns = loop.checkpoint_ns
                 emit_graph_lifecycle_events(loop)
                 # create runner
                 runner = PregelRunner(
@@ -3296,20 +3291,6 @@ class Pregel(
                     for h in callback_manager.inheritable_handlers
                     if not isinstance(h, StreamMessagesHandlerV2)
                 ]
-            if CONFIG_KEY_TASK_ID not in config[CONF]:
-                # Not run as a task (a child invoked with its own thread, say),
-                # this graph's namespace starts at the root, so a caller's
-                # handler would take this graph's events for its own graph's.
-                callback_manager.handlers = [
-                    h
-                    for h in callback_manager.handlers
-                    if not _is_stream_mode_handler(h)
-                ]
-                callback_manager.inheritable_handlers = [
-                    h
-                    for h in callback_manager.inheritable_handlers
-                    if not _is_stream_mode_handler(h)
-                ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
             run_manager = await callback_manager.on_chain_start(
@@ -3323,10 +3304,18 @@ class Pregel(
                 run_id=run_manager.run_id,
             )
             # if running from astream_log() run each proc with streaming
-            do_stream = any(
-                _is_stream_handler(h)
-                and not isinstance(h, (StreamMessagesHandler, StreamToolCallHandler))
-                for h in run_manager.handlers
+            do_stream = (
+                next(
+                    (
+                        True
+                        for h in run_manager.handlers
+                        if isinstance(h, _StreamingCallbackHandler)
+                        and not isinstance(h, StreamMessagesHandler)
+                    ),
+                    False,
+                )
+                if _StreamingCallbackHandler is not None
+                else False
             )
             if checkpointer is None and durability is not None:
                 warnings.warn(
@@ -3334,9 +3323,10 @@ class Pregel(
                 )
             # set up subgraph checkpointing
             config = self._own_checkpoint_config(config)
-            stream_handlers: list[StreamMessagesHandler | StreamToolCallHandler] = []
             # set up messages stream mode
             if "messages" in stream_modes:
+                # namespace can be None in a root level graph?
+                ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
                 use_stream_messages_v2 = bool(
                     version == "v2" and config[CONF].get(CONFIG_KEY_STREAM_MESSAGES_V2)
                 )
@@ -3345,12 +3335,24 @@ class Pregel(
                     if use_stream_messages_v2
                     else StreamMessagesHandler
                 )
-                stream_handlers.append(messages_handler_cls(stream_put, subgraphs))
+                run_manager.inheritable_handlers.append(
+                    messages_handler_cls(
+                        stream_put,
+                        subgraphs,
+                        parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
+                    )
+                )
 
             # set up tools stream mode
             if "tools" in stream_modes:
-                stream_handlers.append(StreamToolCallHandler(stream_put, subgraphs))
-            run_manager.inheritable_handlers.extend(stream_handlers)
+                ns_tools = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
+                run_manager.inheritable_handlers.append(
+                    StreamToolCallHandler(
+                        stream_put,
+                        subgraphs,
+                        parent_ns=tuple(ns_tools.split(NS_SEP)) if ns_tools else None,
+                    )
+                )
 
             # set up custom stream mode
             def stream_writer(c: Any) -> None:
@@ -3407,8 +3409,6 @@ class Pregel(
                 execution_info=None,
                 server_info=server_info,
                 control=control or parent_runtime.control or RunControl(),
-                # a subgraph shares the root run's exit callbacks
-                _run_exits=parent_runtime._run_exits or _RunExits(),
             )
             runtime = parent_runtime.merge(runtime)
             config[CONF][CONFIG_KEY_RUNTIME] = runtime
@@ -3461,10 +3461,6 @@ class Pregel(
                 cache_policy=self.cache_policy,
                 has_graph_lifecycle_callbacks=bool(graph_callback_manager.handlers),
             ) as loop:
-                # Known only now: the loop numbers the namespace of every
-                # subgraph a task runs after its first.
-                for handler in stream_handlers:
-                    handler.parent_ns = loop.checkpoint_ns
                 await aemit_graph_lifecycle_events(loop)
                 # create runner
                 runner = PregelRunner(
@@ -4462,10 +4458,6 @@ def _coerce_checkpoint_values(payload: Any, mapper: Callable[[Any], Any]) -> Non
         payload["values"] = mapper(payload["values"])
 
 
-def _is_stream_mode_handler(handler: object) -> bool:
-    return isinstance(handler, (StreamMessagesHandler, StreamToolCallHandler))
-
-
 def _resolve_parent_ns(
     graph_config: RunnableConfig | None, call_config: RunnableConfig | None
 ) -> tuple[str, ...]:

libs/langgraph/langgraph/pregel/remote.py

@@ -847,11 +847,6 @@ 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))
@@ -1002,11 +997,6 @@ 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/langgraph/runtime.py

@@ -1,11 +1,7 @@
 from __future__ import annotations
 
-import inspect
-import logging
-import threading
-from collections.abc import Awaitable, Callable
+from collections.abc import Callable
 from dataclasses import dataclass, field, replace
-from types import TracebackType
 from typing import Any, Generic, cast
 
 from langgraph.store.base import BaseStore
@@ -108,129 +104,6 @@ class RunControl:
         return self._drain_reason
 
 
-logger = logging.getLogger(__name__)
-
-ExitCallback = Callable[[], Awaitable[None] | None]
-"""A cleanup callback registered with `Runtime.on_exit`."""
-
-
-class _RunExits:
-    """Cleanup callbacks for one root graph invocation.
-
-    Created by the root `stream`/`astream` call and shared by every subgraph
-    it runs, through `Runtime`. The root Pregel loop binds it at the bottom of
-    its exit stack, so callbacks run after the loop has stopped every task,
-    persisted its final checkpoint and handled any interrupt, whatever the
-    reason the run ended.
-    """
-
-    __slots__ = ("_callbacks", "_is_async", "_lock", "_state")
-
-    def __init__(self) -> None:
-        self._callbacks: list[ExitCallback] = []
-        self._lock = threading.Lock()
-        self._is_async = False
-        # "new" -> "bound" (owned by a loop) -> "closed" (callbacks ran)
-        self._state = "new"
-
-    def bind(self, *, is_async: bool) -> bool:
-        """Claim these exits for a loop. Only the first (root) loop wins."""
-        with self._lock:
-            if self._state != "new":
-                return False
-            self._state = "bound"
-            self._is_async = is_async
-            return True
-
-    def register(self, callback: ExitCallback) -> None:
-        if not callable(callback):
-            msg = "on_exit() expects a callable"
-            raise TypeError(msg)
-        with self._lock:
-            if self._state == "closed":
-                msg = "on_exit() called after the graph run has already exited"
-                raise RuntimeError(msg)
-            if not self._is_async and _is_async_callable(callback):
-                msg = (
-                    "on_exit() got an async callback in a synchronous graph run "
-                    "(invoke/stream). Register a sync callback, or run the graph "
-                    "with ainvoke/astream."
-                )
-                raise TypeError(msg)
-            self._callbacks.append(callback)
-
-    def _drain(self) -> list[ExitCallback]:
-        with self._lock:
-            self._state = "closed"
-            callbacks = self._callbacks[::-1]  # LIFO
-            self._callbacks = []
-        return callbacks
-
-    def exit(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_value: BaseException | None,
-        traceback: TracebackType | None,
-    ) -> bool:
-        errors: list[Exception] = []
-        for callback in self._drain():
-            try:
-                result = callback()
-                if inspect.isawaitable(result):
-                    if inspect.iscoroutine(result):
-                        result.close()
-                    msg = (
-                        f"on_exit() callback {callback!r} returned an awaitable "
-                        "in a synchronous graph run; it was not awaited."
-                    )
-                    raise TypeError(msg)
-            except Exception as e:
-                errors.append(e)
-        return self._finish(errors, exc_value)
-
-    async def aexit(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_value: BaseException | None,
-        traceback: TracebackType | None,
-    ) -> bool:
-        errors: list[Exception] = []
-        for callback in self._drain():
-            try:
-                result = callback()
-                if inspect.isawaitable(result):
-                    await result
-            except Exception as e:
-                errors.append(e)
-        return self._finish(errors, exc_value)
-
-    @staticmethod
-    def _finish(errors: list[Exception], exc_value: BaseException | None) -> bool:
-        # Every callback runs even if an earlier one failed. A failing callback
-        # never replaces the run's own exception (error, cancellation, stream
-        # close): it is logged instead. If the run otherwise ended cleanly, the
-        # first callback error is raised so cleanup bugs aren't silent.
-        if not errors:
-            return False
-        if exc_value is None:
-            for e in errors[1:]:
-                logger.error("on_exit() callback failed", exc_info=e)
-            raise errors[0]
-        for e in errors:
-            logger.error(
-                "on_exit() callback failed while the graph run was exiting with %r",
-                exc_value,
-                exc_info=e,
-            )
-        return False
-
-
-def _is_async_callable(callback: Any) -> bool:
-    return inspect.iscoroutinefunction(callback) or inspect.iscoroutinefunction(
-        getattr(callback, "__call__", None)
-    )
-
-
 def _no_op_stream_writer(_: Any) -> None: ...
 
 
@@ -364,40 +237,6 @@ class Runtime(Generic[ContextT]):
     graph runtime.
     """
 
-    _run_exits: _RunExits | None = field(default=None, repr=False, compare=False)
-    """Exit callbacks of the root graph run. Use `on_exit` to register one."""
-
-    def on_exit(self, callback: ExitCallback) -> None:
-        """Register cleanup to run when the current graph run ends.
-
-        The callback runs once the root invocation (the `invoke`/`stream`/
-        `ainvoke`/`astream` call the caller made) exits for any reason: it
-        completed, raised, was cancelled, paused on an interrupt, or the caller
-        closed the stream or broke out of it. Calls from subgraphs register on
-        that same root run.
-
-        Callbacks take no arguments and run in reverse registration order,
-        after every task of the run has stopped and the final checkpoint has
-        been written. In `ainvoke`/`astream` a callback may be sync or async; in
-        `invoke`/`stream` it must be sync. A failing callback never hides the
-        run's own exception: it is logged, and raised only if the run otherwise
-        ended without one.
-
-        A process that is killed does not run its callbacks, so use them to
-        free resources early, not as the only guarantee they are freed.
-
-        Args:
-            callback: A zero-argument sync or async callable.
-
-        Raises:
-            RuntimeError: When called outside a graph run, or after it exited.
-            TypeError: When given an async callback in a synchronous run.
-        """
-        if self._run_exits is None:
-            msg = "on_exit() can only be called during a graph run"
-            raise RuntimeError(msg)
-        self._run_exits.register(callback)
-
     def merge(self, other: Runtime[ContextT]) -> Runtime[ContextT]:
         """Merge two runtimes together.
 
@@ -416,7 +255,6 @@ class Runtime(Generic[ContextT]):
             execution_info=other.execution_info or self.execution_info,
             server_info=other.server_info or self.server_info,
             control=other.control or self.control,
-            _run_exits=other._run_exits or self._run_exits,
         )
 
     def override(

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/__snapshots__/test_large_cases.ambr

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

libs/langgraph/tests/test_pregel.py

@@ -8950,39 +8950,6 @@ 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,31 +6834,6 @@ 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."""
@@ -7623,40 +7598,6 @@ 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:
@@ -9503,44 +9444,6 @@ 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,11 +16,9 @@ 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
@@ -865,87 +863,6 @@ 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_run_exit.py

@@ -1,339 +0,0 @@
-"""Tests for `Runtime.on_exit`: cleanup that runs however a graph run ends."""
-
-from __future__ import annotations
-
-import asyncio
-import sys
-
-import pytest
-from langgraph.checkpoint.memory import InMemorySaver
-from typing_extensions import TypedDict
-
-from langgraph.graph import END, START, StateGraph
-from langgraph.runtime import Runtime, get_runtime
-from langgraph.types import Command, interrupt
-
-pytestmark = pytest.mark.anyio
-
-NEEDS_CONTEXTVARS = pytest.mark.skipif(
-    sys.version_info < (3, 11),
-    reason="Python 3.11+ is required for async contextvars support",
-)
-
-
-class State(TypedDict, total=False):
-    value: str
-
-
-def _two_node_graph(first, second=None, *, checkpointer=None):
-    builder = StateGraph(State)
-    builder.add_node("first", first)
-    builder.add_edge(START, "first")
-    if second is None:
-        builder.add_edge("first", END)
-    else:
-        builder.add_node("second", second)
-        builder.add_edge("first", "second")
-        builder.add_edge("second", END)
-    return builder.compile(checkpointer=checkpointer)
-
-
-# sync
-
-
-def test_runs_on_success_in_lifo_order() -> None:
-    events: list[str] = []
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup-a"))
-        runtime.on_exit(lambda: events.append("cleanup-b"))
-        events.append("first")
-        return {"value": "x"}
-
-    def second(state: State) -> State:
-        events.append("second")
-        return {}
-
-    graph = _two_node_graph(first, second)
-    assert graph.invoke({}) == {"value": "x"}
-    assert events == ["first", "second", "cleanup-b", "cleanup-a"]
-
-
-def test_runs_on_error_and_keeps_the_run_exception() -> None:
-    events: list[str] = []
-
-    def failing_cleanup() -> None:
-        events.append("failing")
-        raise KeyError("cleanup failed")
-
-    def first(state: State) -> State:
-        runtime = get_runtime()
-        runtime.on_exit(lambda: events.append("cleanup"))
-        runtime.on_exit(failing_cleanup)
-        raise ValueError("node failed")
-
-    graph = _two_node_graph(first)
-    with pytest.raises(ValueError, match="node failed"):
-        graph.invoke({})
-    # the failing callback did not stop the earlier one from running
-    assert events == ["failing", "cleanup"]
-
-
-def test_callback_error_is_raised_when_the_run_succeeded() -> None:
-    def failing_cleanup() -> None:
-        raise KeyError("cleanup failed")
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(failing_cleanup)
-        return {}
-
-    graph = _two_node_graph(first)
-    with pytest.raises(KeyError, match="cleanup failed"):
-        graph.invoke({})
-
-
-def test_runs_on_interrupt() -> None:
-    events: list[str] = []
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        return {"value": interrupt("approve?")}
-
-    graph = _two_node_graph(first, checkpointer=InMemorySaver())
-    config = {"configurable": {"thread_id": "1"}}
-    result = graph.invoke({}, config)
-    assert "__interrupt__" in result
-    assert events == ["cleanup"]
-
-    # each resumed run is its own invocation with its own callbacks
-    graph.invoke(Command(resume="ok"), config)
-    assert events == ["cleanup", "cleanup"]
-
-
-def test_runs_when_the_stream_is_closed_early() -> None:
-    events: list[str] = []
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        return {"value": "x"}
-
-    def second(state: State) -> State:
-        events.append("second")
-        return {}
-
-    graph = _two_node_graph(first, second)
-    stream = graph.stream({}, stream_mode="updates")
-    assert next(stream) == {"first": {"value": "x"}}
-    assert events == []
-    stream.close()
-    assert events == ["cleanup"]
-
-
-def test_runs_when_breaking_out_of_stream() -> None:
-    events: list[str] = []
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        return {"value": "x"}
-
-    graph = _two_node_graph(first, lambda s: {})
-    for _ in graph.stream({}, stream_mode="updates"):
-        break
-    # the dropped generator is closed by refcounting on CPython
-    assert events == ["cleanup"]
-
-
-def test_subgraph_registers_on_the_root_run() -> None:
-    events: list[str] = []
-
-    def inner(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        events.append("inner")
-        return {}
-
-    sub = StateGraph(State)
-    sub.add_node("inner", inner)
-    sub.add_edge(START, "inner")
-    subgraph = sub.compile()
-
-    def call_subgraph(state: State) -> State:
-        subgraph.invoke(state)
-        events.append("subgraph returned")
-        return {}
-
-    def after(state: State) -> State:
-        events.append("after")
-        return {}
-
-    builder = StateGraph(State)
-    builder.add_node("as_node", subgraph)
-    builder.add_node("call_subgraph", call_subgraph)
-    builder.add_node("after", after)
-    builder.add_edge(START, "as_node")
-    builder.add_edge("as_node", "call_subgraph")
-    builder.add_edge("call_subgraph", "after")
-    graph = builder.compile()
-
-    graph.invoke({})
-    # nothing runs when either subgraph ends; both run when the root ends
-    assert events == [
-        "inner",
-        "inner",
-        "subgraph returned",
-        "after",
-        "cleanup",
-        "cleanup",
-    ]
-
-
-def test_async_callback_rejected_in_sync_run() -> None:
-    async def acleanup() -> None:
-        pass
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(acleanup)
-        return {}
-
-    graph = _two_node_graph(first)
-    with pytest.raises(TypeError, match="async callback in a synchronous"):
-        graph.invoke({})
-
-
-def test_outside_a_run_raises() -> None:
-    with pytest.raises(RuntimeError, match="during a graph run"):
-        Runtime().on_exit(lambda: None)
-
-
-def test_each_invocation_gets_its_own_callbacks() -> None:
-    calls: list[int] = []
-
-    def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: calls.append(1))
-        return {}
-
-    graph = _two_node_graph(first)
-    graph.invoke({})
-    graph.invoke({})
-    assert calls == [1, 1]
-
-
-# async
-
-
-async def test_async_runs_on_success_sync_and_async_callbacks() -> None:
-    events: list[str] = []
-
-    async def acleanup() -> None:
-        await asyncio.sleep(0)
-        events.append("async-cleanup")
-
-    async def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("sync-cleanup"))
-        runtime.on_exit(acleanup)
-        events.append("first")
-        return {"value": "x"}
-
-    graph = _two_node_graph(first)
-    assert await graph.ainvoke({}) == {"value": "x"}
-    assert events == ["first", "async-cleanup", "sync-cleanup"]
-
-
-async def test_async_runs_on_error_and_keeps_the_run_exception() -> None:
-    events: list[str] = []
-
-    async def failing_cleanup() -> None:
-        raise KeyError("cleanup failed")
-
-    async def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        runtime.on_exit(failing_cleanup)
-        raise ValueError("node failed")
-
-    graph = _two_node_graph(first)
-    with pytest.raises(ValueError, match="node failed"):
-        await graph.ainvoke({})
-    assert events == ["cleanup"]
-
-
-async def test_async_runs_on_interrupt() -> None:
-    events: list[str] = []
-
-    async def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(lambda: events.append("cleanup"))
-        return {"value": interrupt("approve?")}
-
-    graph = _two_node_graph(first, checkpointer=InMemorySaver())
-    result = await graph.ainvoke({}, {"configurable": {"thread_id": "1"}})
-    assert "__interrupt__" in result
-    assert events == ["cleanup"]
-
-
-async def test_async_runs_on_cancel_after_the_node_unwinds() -> None:
-    events: list[str] = []
-    started = asyncio.Event()
-
-    async def acleanup() -> None:
-        events.append("cleanup")
-
-    async def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(acleanup)
-        started.set()
-        try:
-            await asyncio.sleep(60)
-        finally:
-            events.append("node finally")
-        return {}
-
-    graph = _two_node_graph(first)
-    task = asyncio.create_task(graph.ainvoke({}))
-    await started.wait()
-    task.cancel()
-    with pytest.raises(asyncio.CancelledError):
-        await task
-    # cleanup ran after the node had stopped
-    assert events == ["node finally", "cleanup"]
-
-
-async def test_async_runs_when_the_stream_is_closed_early() -> None:
-    events: list[str] = []
-
-    async def acleanup() -> None:
-        events.append("cleanup")
-
-    async def first(state: State, runtime: Runtime) -> State:
-        runtime.on_exit(acleanup)
-        return {"value": "x"}
-
-    async def second(state: State) -> State:
-        events.append("second")
-        return {}
-
-    graph = _two_node_graph(first, second)
-    stream = graph.astream({}, stream_mode="updates")
-    assert await stream.__anext__() == {"first": {"value": "x"}}
-    assert events == []
-    await stream.aclose()
-    assert events == ["cleanup"]
-
-
-@NEEDS_CONTEXTVARS
-async def test_async_subgraph_registers_on_the_root_run() -> None:
-    events: list[str] = []
-
-    async def inner(state: State) -> State:
-        get_runtime().on_exit(lambda: events.append("cleanup"))
-        return {}
-
-    sub = StateGraph(State)
-    sub.add_node("inner", inner)
-    sub.add_edge(START, "inner")
-    subgraph = sub.compile()
-
-    async def call_subgraph(state: State) -> State:
-        await subgraph.ainvoke(state)
-        events.append("subgraph returned")
-        return {}
-
-    graph = _two_node_graph(call_subgraph, lambda s: events.append("after") or {})
-    await graph.ainvoke({})
-    assert events == ["subgraph returned", "after", "cleanup"]

libs/langgraph/tests/test_stream_namespaces.py

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

libs/langgraph/tests/test_tool_stream_handler.py

@@ -13,17 +13,13 @@ 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):
@@ -58,14 +54,14 @@ def _caller_async(tool_name: str, tool_args: dict[str, Any], tc_id: str = "tc1")
     return caller
 
 
-def _build_graph(caller, tools, checkpointer=None) -> Any:
+def _build_graph(caller, tools) -> 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(checkpointer=checkpointer)
+    return sg.compile()
 
 
 def _tool_events(stream) -> list[tuple[tuple[str, ...], dict]]:
@@ -150,105 +146,6 @@ 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.
@@ -299,25 +196,6 @@ 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/tests/test_trace_policy.py

@@ -1,24 +1,11 @@
 """End-to-end tests for `TracePolicy` input processing on node trace runs."""
 
-import asyncio
-from collections.abc import AsyncIterator, Iterator
 from typing import Any
-from uuid import UUID
 
 import pytest
-from langchain_core.callbacks import BaseCallbackHandler
-from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
-from langchain_core.messages import AIMessage, HumanMessage
-from langchain_core.runnables import RunnableConfig
 from typing_extensions import TypedDict
 
-from langgraph._internal._runnable import (
-    ASYNCIO_ACCEPTS_CONTEXT,
-    RunnableCallable,
-    RunnableSeq,
-)
-from langgraph.graph import END, START, MessagesState, StateGraph
-from langgraph.graph.state import CompiledStateGraph
+from langgraph.graph import END, START, StateGraph
 from langgraph.types import TracePolicy
 from tests.fake_tracer import FakeTracer, Run
 
@@ -160,179 +147,3 @@ async def test_trace_policy_transforms_recorded_inputs_async() -> None:
     run = _node_run(tracer, "n")
     assert run.inputs == {"scrubbed_in": True}
     assert run.outputs == {"value": 6}
-
-
-_RECORDING_POLICY = TracePolicy(
-    process_inputs=lambda _: {"recorded": "inputs"},
-    process_outputs=lambda _: {"recorded": "outputs"},
-)
-
-
-def _chat_graph() -> CompiledStateGraph:
-    model = FakeMessagesListChatModel(
-        responses=[AIMessage("model reply", id="ai-model")]
-    )
-
-    def chat(state: MessagesState) -> dict[str, Any]:
-        reply = model.invoke(state["messages"])
-        return {
-            "messages": [
-                *state["messages"],
-                reply,
-                AIMessage("from node", id="ai-node"),
-            ]
-        }
-
-    return (
-        StateGraph(MessagesState)
-        .add_node("chat", chat, trace_policy=_RECORDING_POLICY)
-        .add_edge(START, "chat")
-        .compile()
-    )
-
-
-def _assert_scrubbed_node_run(tracer: FakeTracer, *, model_is_child: bool) -> None:
-    run = _node_run(tracer, "chat")
-    assert run.inputs == {"recorded": "inputs"}
-    assert run.outputs == {"recorded": "outputs"}
-    if model_is_child:
-        assert [c.name for c in run.child_runs] == ["FakeMessagesListChatModel"]
-
-
-def test_trace_policy_scrubs_tracer_not_messages_stream() -> None:
-    tracer = FakeTracer()
-    streamed = [
-        message.content
-        for message, _ in _chat_graph().stream(
-            {"messages": [HumanMessage("hello", id="h-1")]},
-            {"callbacks": [tracer]},
-            stream_mode="messages",
-        )
-    ]
-
-    assert streamed == ["model reply", "from node"]
-    _assert_scrubbed_node_run(tracer, model_is_child=True)
-
-
-@pytest.mark.anyio
-async def test_trace_policy_scrubs_tracer_not_messages_stream_async() -> None:
-    tracer = FakeTracer()
-    streamed = [
-        message.content
-        async for message, _ in _chat_graph().astream(
-            {"messages": [HumanMessage("hello", id="h-1")]},
-            {"callbacks": [tracer]},
-            stream_mode="messages",
-        )
-    ]
-
-    assert streamed == ["model reply", "from node"]
-    _assert_scrubbed_node_run(tracer, model_is_child=ASYNCIO_ACCEPTS_CONTEXT)
-
-
-@pytest.mark.anyio
-async def test_trace_policy_scrubs_tracer_not_astream_events() -> None:
-    tracer = FakeTracer()
-    chat_events = [
-        event
-        async for event in _chat_graph().astream_events(
-            {"messages": [HumanMessage("hello", id="h-1")]},
-            {"callbacks": [tracer]},
-            version="v2",
-        )
-        if event["name"] == "chat"
-        and event["event"] in ("on_chain_start", "on_chain_end")
-    ]
-
-    assert [e["event"] for e in chat_events] == ["on_chain_start", "on_chain_end"]
-    start, end = (e["data"] for e in chat_events)
-    assert [m.content for m in start["input"]["messages"]] == ["hello"]
-    assert [m.content for m in end["output"]["messages"]] == [
-        "hello",
-        "model reply",
-        "from node",
-    ]
-    _assert_scrubbed_node_run(tracer, model_is_child=ASYNCIO_ACCEPTS_CONTEXT)
-
-
-class _HandlerWithTapMethods(BaseCallbackHandler):
-    def __init__(self) -> None:
-        self.node_payloads: list[Any] = []
-        self.node_run_ids: set[UUID] = set()
-
-    def _record(self, run_id: UUID, payload: Any) -> None:
-        if run_id in self.node_run_ids:
-            self.node_payloads.append(payload)
-
-    def tap_output_iter(self, run_id: UUID, output: Iterator[Any]) -> Iterator[Any]:
-        for chunk in output:
-            self._record(run_id, chunk)
-            yield chunk
-
-    async def tap_output_aiter(
-        self, run_id: UUID, output: AsyncIterator[Any]
-    ) -> AsyncIterator[Any]:
-        async for chunk in output:
-            self._record(run_id, chunk)
-            yield chunk
-
-    def on_chain_start(
-        self, serialized: Any, inputs: Any, *, run_id: UUID, **kwargs: Any
-    ) -> None:
-        if kwargs.get("name") == "n":
-            self.node_run_ids.add(run_id)
-            self._record(run_id, inputs)
-
-    def on_chain_end(self, outputs: Any, *, run_id: UUID, **kwargs: Any) -> None:
-        self._record(run_id, outputs)
-
-
-@pytest.mark.anyio
-@pytest.mark.parametrize("via_events", [False, True], ids=["ainvoke", "astream_events"])
-async def test_trace_policy_reaches_tracers_but_not_custom_handlers(
-    via_events: bool,
-) -> None:
-    graph = (
-        StateGraph(State)
-        .add_node("n", _incr, trace_policy=_RECORDING_POLICY)
-        .add_edge(START, "n")
-        .add_edge("n", END)
-        .compile()
-    )
-    handler = _HandlerWithTapMethods()
-    tracer = FakeTracer()
-    config: RunnableConfig = {"callbacks": [handler, tracer]}
-
-    if via_events:
-        async for _ in graph.astream_events({"value": 1}, config, version="v2"):
-            pass
-    else:
-        assert await graph.ainvoke({"value": 1}, config) == {"value": 2}
-    assert handler.node_payloads == [{"value": 1}, {"value": 2}]
-    run = _node_run(tracer, "n")
-    assert (run.inputs, run.outputs) == (
-        {"recorded": "inputs"},
-        {"recorded": "outputs"},
-    )
-
-
-@pytest.mark.anyio
-async def test_trace_policy_cancel_at_node_end_still_closes_tracer_run() -> None:
-    def cancel_own_task(value: Any) -> Any:
-        asyncio.current_task().cancel()
-        return value
-
-    tracer = FakeTracer()
-    seq = RunnableSeq(
-        RunnableCallable(lambda x: x, trace=False),
-        RunnableCallable(cancel_own_task, trace=False),
-        trace_policy=_RECORDING_POLICY,
-    )
-    with pytest.raises(asyncio.CancelledError):
-        await asyncio.create_task(seq.ainvoke({}, {"callbacks": [tracer]}))
-    for _ in range(100):
-        if tracer.runs:
-            break
-        await asyncio.sleep(0.01)
-
-    assert [r.outputs for r in tracer.runs] == [{"recorded": "outputs"}]

libs/langgraph/uv.lock

@@ -1439,7 +1439,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.19"
+version = "1.2.17"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1625,7 +1625,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.1"
+version = "4.3.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1722,7 +1722,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.3"
+version = "3.1.2"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
     { name = "aiosqlite" },
@@ -3419,11 +3419,11 @@ wheels = [
 
 [[package]]
 name = "setuptools"
-version = "84.1.0"
+version = "83.0.0"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]

libs/prebuilt/uv.lock

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

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.11"
+__version__ = "0.4.9"
 
 __all__ = [
     "Auth",

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

@@ -14,14 +14,9 @@ import orjson
 
 from langgraph_sdk._shared.utilities import (
     _orjson_default,
-    _reconnect_delay,
     _validate_reconnect_location,
 )
-from langgraph_sdk.errors import (
-    InternalServerError,
-    RateLimitError,
-    _araise_for_status_typed,
-)
+from langgraph_sdk.errors import _araise_for_status_typed
 from langgraph_sdk.schema import QueryParamTypes, StreamPart
 from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw
 
@@ -151,8 +146,7 @@ class HttpClient:
         on_response: Callable[[httpx.Response], None] | None = None,
         reconnect_limit: int = 5,
     ) -> Any:
-        """Send a request that reconnects to its Location header if the response is
-        cut off, with backoff between attempts."""
+        """Send a request that automatically reconnects to Location header."""
         request_headers, content = await _aencode_json(json)
         if headers:
             request_headers.update(headers)
@@ -182,27 +176,13 @@ class HttpClient:
                     stacklevel=2,
                 )
                 await r.aclose()
-                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
+                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,
+                )
 
     async def stream(
         self,
@@ -237,104 +217,72 @@ 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
 
-            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}"
-                        )
+            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_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(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""):
+                # 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:
                             if decoder.last_event_id is not None:
                                 last_event_id = decoder.last_event_id
-                            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.
+                            if sse.event or sse.data is not None:
                                 yield sse
-            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:
+                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:
                 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,6 +16,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import logging
+import random
 from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
 from dataclasses import dataclass, field
 from typing import Any, Literal, TypedDict, cast
@@ -24,7 +25,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, _reconnect_delay
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.schema import LangSmithTracing, QueryParamTypes
 from langgraph_sdk.stream.controller import _SeenEventIds
 from langgraph_sdk.stream.decoders import (
@@ -1690,13 +1691,11 @@ class AsyncThreadStream:
 
     async def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff and jitter for reconnect attempt `attempt`."""
-        await asyncio.sleep(
-            _reconnect_delay(
-                attempt,
-                base=self._shared_reconnect_backoff_base,
-                cap=self._shared_reconnect_backoff_cap,
-            )
-        )
+        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)
 
     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,7 +4,6 @@ from __future__ import annotations
 
 import functools
 import os
-import random
 import re
 from collections.abc import Mapping
 from datetime import tzinfo
@@ -195,15 +194,6 @@ 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,7 +4,6 @@ from __future__ import annotations
 
 import logging
 import sys
-import time
 import warnings
 from collections.abc import Callable, Iterator, Mapping
 from typing import Any, cast
@@ -14,14 +13,9 @@ import orjson
 
 from langgraph_sdk._shared.utilities import (
     _orjson_default,
-    _reconnect_delay,
     _validate_reconnect_location,
 )
-from langgraph_sdk.errors import (
-    InternalServerError,
-    RateLimitError,
-    _raise_for_status_typed,
-)
+from langgraph_sdk.errors import _raise_for_status_typed
 from langgraph_sdk.schema import QueryParamTypes, StreamPart
 from langgraph_sdk.sse import SSEDecoder, iter_lines_raw
 
@@ -152,8 +146,7 @@ class SyncHttpClient:
         on_response: Callable[[httpx.Response], None] | None = None,
         reconnect_limit: int = 5,
     ) -> Any:
-        """Send a request that reconnects to its Location header if the response is
-        cut off, with backoff between attempts."""
+        """Send a request that automatically reconnects to Location header."""
         request_headers, content = _encode_json(json)
         if headers:
             request_headers.update(headers)
@@ -183,27 +176,13 @@ class SyncHttpClient:
                     stacklevel=2,
                 )
                 r.close()
-                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
+                return self.request_reconnect(
+                    loc,
+                    "GET",
+                    headers=headers,
+                    # don't pass on_response so it's only called once
+                    reconnect_limit=reconnect_limit - 1,
+                )
 
     def stream(
         self,
@@ -240,99 +219,69 @@ 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
 
-            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}"
-                        )
+            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_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(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""):
+                decoder = SSEDecoder()
+                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:
-                                # See async stream implementation for rationale on
-                                # skipping empty flush events.
+                            if sse.event or sse.data is not None:
                                 yield sse
-            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:
+                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:
                 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] | None = None,
+        resources: str | Sequence[str],
         actions: str | Sequence[str] | None = None,
     ) -> Callable[[AHO], AHO]: ...
 

libs/sdk-py/langgraph_sdk/sse.py

@@ -30,14 +30,18 @@ class BytesLineDecoder:
         # See https://docs.python.org/3/glossary.html#term-universal-newlines
         NEWLINE_CHARS = b"\n\r"
 
-        if not text:
-            return []
-        if self.trailing_cr and text.startswith(b"\n"):
-            text = text[1:]
-        self.trailing_cr = text.endswith(b"\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 []
+            # 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
 
         trailing_newline = text[-1] in NEWLINE_CHARS
         lines = cast(list[BytesLike], text.splitlines())
@@ -62,7 +66,7 @@ class BytesLineDecoder:
         return lines
 
     def flush(self) -> list[BytesLike]:
-        if not self.buffer:
+        if not self.buffer and not self.trailing_cr:
             return []
 
         lines: list[BytesLike] = [self.buffer]
@@ -72,25 +76,18 @@ class BytesLineDecoder:
 
 
 class SSEDecoder:
-    def __init__(self, *, last_event_id: str | None = None) -> None:
+    def __init__(self) -> None:
         self._event = ""
         self._data = bytearray()
-        self._last_event_id = last_event_id
+        self._last_event_id: str | None = None
         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 an ID is supplied or received, and `""` after an empty
+        `None` until the stream sends an `id` field, and `""` after an empty
         one, which resets the reconnect cursor.
         """
 
@@ -104,26 +101,17 @@ 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._id_changed
+                and not self._last_event_id
                 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=data,  # ty: ignore[invalid-argument-type]
+                data=orjson.loads(self._data) if self._data else None,  # ty: ignore[invalid-argument-type]
                 id=self._last_event_id or None,
             )
 
@@ -131,7 +119,6 @@ class SSEDecoder:
             self._event = ""
             self._data = bytearray()
             self._retry = None
-            self._id_changed = False
 
             return sse
 
@@ -144,16 +131,16 @@ class SSEDecoder:
             value = value[1:]
 
         if fieldname == b"event":
-            self._event = value.decode(errors="replace")
+            self._event = value.decode()
         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(errors="replace")
-                self._id_changed = True
+                self._last_event_id = value.decode()
         elif fieldname == b"retry":
             with contextlib.suppress(TypeError, ValueError):
                 self._retry = int(value)

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

@@ -15,6 +15,7 @@ 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
@@ -22,7 +23,6 @@ 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,13 +275,11 @@ class StreamController:
 
     async def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff and jitter for reconnect attempt *attempt*."""
-        await asyncio.sleep(
-            _reconnect_delay(
-                attempt,
-                base=self._reconnect_backoff_base,
-                cap=self._reconnect_backoff_cap,
-            )
-        )
+        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)
 
     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,6 +4,7 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import random
 import threading
 import time
 from dataclasses import dataclass, field
@@ -12,7 +13,6 @@ 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,13 +282,11 @@ class SyncStreamController:
 
     def _reconnect_sleep(self, attempt: int) -> None:
         """Sleep with exponential backoff + jitter before a reconnect attempt."""
-        time.sleep(
-            _reconnect_delay(
-                attempt,
-                base=self._reconnect_backoff_base,
-                cap=self._reconnect_backoff_cap,
-            )
-        )
+        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)
 
     def _reconnect_shared_stream(self) -> bool:
         """Attempt to reopen the shared stream after a transport drop.

libs/sdk-py/tests/conftest.py

@@ -1,17 +0,0 @@
-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,49 +1,8 @@
-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 Callable, Iterator, Sequence
+from collections.abc import Iterator, Sequence
 from pathlib import Path
 from typing import Any
 
@@ -13,7 +13,6 @@ 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,
@@ -131,110 +130,6 @@ 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 ---
 
 
@@ -281,7 +176,6 @@ 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 = [
@@ -347,7 +241,6 @@ 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 = [
@@ -419,14 +312,18 @@ async def test_http_client_stream_recovers_after_disconnect():
     ]
 
 
-def _drop_after(chunks: list[bytes], stream_cls: type, reconnects: list[httpx.Request]):
+def _drop_after_an_empty_id(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(
-                    chunks, httpx.RemoteProtocolError("incomplete chunked read")
+                    [
+                        b"id: 1\nevent: values\ndata: 1\n\n",
+                        b"id:\nevent: values\ndata: 2\n\n",
+                    ],
+                    httpx.RemoteProtocolError("incomplete chunked read"),
                 ),
             )
         reconnects.append(request)
@@ -439,409 +336,26 @@ def _drop_after(chunks: list[bytes], stream_cls: type, reconnects: list[httpx.Re
     return handler
 
 
-def _sync_reconnect(
-    chunks: list[bytes], headers: dict[str, str] | None
-) -> httpx.Request:
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
     reconnects: list[httpx.Request] = []
-    transport = httpx.MockTransport(_drop_after(chunks, ListByteStream, reconnects))
+    transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream, reconnects))
     with httpx.Client(transport=transport, base_url="https://example.com") as client:
-        list(SyncHttpClient(client).stream("/stream", "POST", json={}, headers=headers))
-    return reconnects[0]
+        list(SyncHttpClient(client).stream("/stream", "POST", json={}))
+
+    assert "last-event-id" not in reconnects[0].headers
 
 
-async def _async_reconnect(
-    chunks: list[bytes], headers: dict[str, str] | None
-) -> httpx.Request:
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
     reconnects: list[httpx.Request] = []
     transport = httpx.MockTransport(
-        _drop_after(chunks, AsyncListByteStream, reconnects)
+        _drop_after_an_empty_id(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={}, headers=headers
-            )
-        ]
-    return reconnects[0]
+        [part async for part in HttpClient(client).stream("/stream", "POST", json={})]
 
-
-_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]
-
-
-# Each reconnect resumes after the newest event the stream got, so a reconnect
-# loop on a resumable run replays only what it missed. A failed reconnect keeps
-# the cursor for the next one.
-RECONNECT_LOOPS = {
-    "progress_between_drops": (
-        lambda stream_cls: _long_stream_steps(stream_cls, drops=3),
-        [None, "0", "1", "2"],
-    ),
-    "failed_reconnect": (
-        lambda stream_cls: _transient_steps(stream_cls, _status(502)),
-        [None, "1", "1"],
-    ),
-}
-
-
-@pytest.mark.usefixtures("no_sleep")
-@pytest.mark.parametrize(
-    ("steps", "cursors"), RECONNECT_LOOPS.values(), ids=RECONNECT_LOOPS.keys()
-)
-def test_sync_reconnect_loop_resumes_after_the_newest_event(steps, cursors):
-    _, seen = _run_sync(steps(ListByteStream))
-    assert [r.headers.get("last-event-id") for r in seen] == cursors
-
-
-@pytest.mark.usefixtures("no_sleep")
-@pytest.mark.parametrize(
-    ("steps", "cursors"), RECONNECT_LOOPS.values(), ids=RECONNECT_LOOPS.keys()
-)
-async def test_async_reconnect_loop_resumes_after_the_newest_event(steps, cursors):
-    _, seen = await _run_async(steps(AsyncListByteStream))
-    assert [r.headers.get("last-event-id") for r in seen] == cursors
-
-
-# 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)
+    assert "last-event-id" not in reconnects[0].headers
 
 
 # --- _sse_to_v2_dict conversion ---

libs/sdk-py/tests/test_runs_client.py

@@ -1,8 +1,7 @@
 from __future__ import annotations
 
 import json
-from collections.abc import AsyncIterator, Callable, Iterator, Sequence
-from typing import Any
+from collections.abc import AsyncIterator, Iterator
 
 import httpx
 import pytest
@@ -25,16 +24,7 @@ class _BodyDropsMidRead(httpx.SyncByteStream, httpx.AsyncByteStream):
         raise httpx.ReadError("connection dropped")
 
 
-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 _wait_handler(requests: list[httpx.Request]):
     def handler(request: httpx.Request) -> httpx.Response:
         requests.append(request)
         if request.method == "POST":
@@ -43,7 +33,7 @@ def _wait_handler(
                 headers={"Location": "/threads/t1/runs/r1/join"},
                 stream=_BodyDropsMidRead(),
             )
-        return reconnects[len(requests) - 2](request)
+        return httpx.Response(200, json={"done": True})
 
     return handler
 
@@ -56,7 +46,6 @@ 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))
@@ -69,7 +58,6 @@ 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))
@@ -84,89 +72,6 @@ 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.19"
+version = "1.2.17"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -385,7 +385,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.3.1"
+version = "4.3.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },