feat(cli): log in to LangSmith without an API key

Reuse the `langsmith login` OAuth profile (~/.langsmith/config.json) for the
deploy commands, refreshing the 5-minute access token per request, and add
`langgraph login` (device flow). With no credentials, an interactive deploy
starts the login flow instead of prompting for an API key.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

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

@@ -360,7 +360,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "3.1.3"
+version = "3.1.2"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },

libs/cli/langgraph_cli/cli.py

@@ -11,6 +11,7 @@ import click.exceptions
 
 import langgraph_cli.config
 import langgraph_cli.docker
+from langgraph_cli import langsmith_auth
 from langgraph_cli.analytics import log_command
 from langgraph_cli.config import Config
 from langgraph_cli.constants import DEFAULT_CONFIG, DEFAULT_PORT
@@ -233,6 +234,13 @@ def cli():
     pass
 
 
+@cli.command(help="Log in to LangSmith in the browser (no API key needed).")
+@click.option("--no-browser", is_flag=True, help="Print the URL instead of opening it.")
+def login(no_browser: bool) -> None:
+    api_url = langsmith_auth.resolve_api_url(os.environ.get("LANGSMITH_ENDPOINT"))
+    langsmith_auth.login(api_url, open_browser=not no_browser)
+
+
 # Wire the deploy group (defined in deploy.py) into the top-level CLI.
 cli.add_command(deploy)
 

libs/cli/langgraph_cli/deploy.py

@@ -1,6 +1,7 @@
 """Deploy command and subcommands for the LangGraph CLI."""
 
 import base64
+import functools
 import json as json_mod
 import os
 import pathlib
@@ -19,6 +20,7 @@ import click.exceptions
 from dotenv import dotenv_values, set_key
 
 import langgraph_cli.config
+from langgraph_cli import langsmith_auth
 from langgraph_cli.analytics import log_command
 from langgraph_cli.config import Config
 from langgraph_cli.constants import DEFAULT_CONFIG
@@ -1982,17 +1984,6 @@ def _create_host_backend_client(
             if val:
                 resolved_api_key = val
                 break
-    if not resolved_api_key:
-        if effective_no_input:
-            raise click.ClickException(
-                "No LangSmith API key found. Set LANGSMITH_API_KEY in the "
-                "environment or .env file."
-            )
-        click.secho(
-            "No LangSmith API key found. Create one at Settings > API Keys in LangSmith.",
-            fg="yellow",
-        )
-        resolved_api_key = click.prompt("Enter LangSmith API key", hide_input=True)
     tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get(
         "LANGSMITH_TENANT_ID"
     )
@@ -2000,8 +1991,31 @@ def _create_host_backend_client(
         "LANGSMITH_ENDPOINT"
     )
     endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
+
+    access_token: Callable[[], str] | None = None
+    if not resolved_api_key:
+        profile = langsmith_auth.load_profile()
+        api_url = langsmith_auth.resolve_api_url(langsmith_endpoint)
+        if profile.get("api_key"):
+            resolved_api_key = profile["api_key"]
+        else:
+            if not any((profile.get("oauth") or {}).values()):
+                if effective_no_input:
+                    raise click.ClickException(
+                        "No LangSmith credentials found. Set LANGSMITH_API_KEY "
+                        "in the environment or .env file (an API key), or run "
+                        "`langgraph login`."
+                    )
+                click.secho("You are not logged in to LangSmith.", fg="yellow")
+                langsmith_auth.login(api_url)
+                profile = langsmith_auth.load_profile()
+            access_token = functools.partial(langsmith_auth.get_access_token, api_url)
+        tenant_id = tenant_id or profile.get("workspace_id")
     return HostBackendClient(
-        endpoints.control_plane_url, resolved_api_key, tenant_id=tenant_id
+        endpoints.control_plane_url,
+        resolved_api_key,
+        tenant_id=tenant_id,
+        **({"access_token": access_token} if access_token else {}),
     )
 
 

libs/cli/langgraph_cli/host_backend.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from typing import Any, Literal
 from urllib.parse import urlparse
@@ -130,17 +130,18 @@ class HostBackendClient:
     def __init__(
         self,
         base_url: str,
-        api_key: str,
+        api_key: str | None,
         tenant_id: str | None = None,
         *,
+        access_token: Callable[[], str] | None = None,
         transport: httpx.BaseTransport | None = None,
     ):
         if not base_url:
             raise click.UsageError("Host backend URL is required")
-        headers: dict[str, str] = {
-            "X-Api-Key": api_key,
-            "Accept": "application/json",
-        }
+        self._access_token = access_token
+        headers: dict[str, str] = {"Accept": "application/json"}
+        if api_key:
+            headers["X-Api-Key"] = api_key
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
         self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
@@ -170,8 +171,15 @@ class HostBackendClient:
         payload: dict[str, Any] | None = None,
         params: dict[str, Any] | None = None,
     ) -> Any:
+        headers = (
+            {"Authorization": f"Bearer {self._access_token()}"}
+            if self._access_token
+            else None
+        )
         try:
-            resp = self._client.request(method, path, json=payload, params=params)
+            resp = self._client.request(
+                method, path, json=payload, params=params, headers=headers
+            )
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
             detail = _error_detail(err.response)

libs/cli/langgraph_cli/langsmith_auth.py

@@ -0,0 +1,225 @@
+"""LangSmith OAuth login, shared with the ``langsmith`` CLI via ``~/.langsmith/config.json``."""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import pathlib
+import time
+import webbrowser
+from typing import Any
+from urllib.parse import urlparse
+
+import click
+import httpx
+
+CLIENT_ID = "langsmith-cli"
+DEFAULT_API_URL = "https://api.smith.langchain.com"
+_WELL_KNOWN = "/.well-known/oauth-authorization-server"
+_REFRESH_LEEWAY = datetime.timedelta(seconds=60)
+_DEVICE_GRANT = "urn:ietf:params:oauth:grant-type:device_code"
+
+
+def _config_path() -> pathlib.Path:
+    override = os.environ.get("LANGSMITH_CONFIG_FILE")
+    if override:
+        return pathlib.Path(override)
+    return pathlib.Path.home() / ".langsmith" / "config.json"
+
+
+def _read_config() -> dict[str, Any]:
+    try:
+        config = json.loads(_config_path().read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+    return config if isinstance(config, dict) else {}
+
+
+def _write_config(config: dict[str, Any]) -> None:
+    path = _config_path()
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
+    os.chmod(tmp, 0o600)
+    os.replace(tmp, path)
+
+
+def _profile_name(config: dict[str, Any]) -> str:
+    return (
+        os.environ.get("LANGSMITH_PROFILE")
+        or config.get("current_profile")
+        or "default"
+    )
+
+
+def load_profile() -> dict[str, Any]:
+    """Return the active profile from the LangSmith config file, or ``{}``."""
+    config = _read_config()
+    profile = (config.get("profiles") or {}).get(_profile_name(config))
+    return profile if isinstance(profile, dict) else {}
+
+
+def resolve_api_url(endpoint: str | None = None) -> str:
+    """Pick the LangSmith API URL: explicit endpoint, then profile, then cloud."""
+    return endpoint or load_profile().get("api_url") or DEFAULT_API_URL
+
+
+def _save_tokens(token: dict[str, Any], api_url: str | None = None) -> None:
+    config = _read_config()
+    name = _profile_name(config)
+    profile = config.setdefault("profiles", {}).setdefault(name, {})
+    if api_url:
+        profile["api_url"] = api_url
+    oauth = profile.setdefault("oauth", {})
+    oauth["access_token"] = token["access_token"]
+    if token.get("refresh_token"):
+        oauth["refresh_token"] = token["refresh_token"]
+    if token.get("expires_in"):
+        expires_at = datetime.datetime.now(datetime.timezone.utc) + (
+            datetime.timedelta(seconds=int(token["expires_in"]))
+        )
+        oauth["expires_at"] = expires_at.strftime("%Y-%m-%dT%H:%M:%SZ")
+    config.setdefault("current_profile", name)
+    _write_config(config)
+
+
+def _oauth_endpoints(api_url: str) -> dict[str, str]:
+    """Discover the OAuth endpoints, falling back to ``<root>/oauth/*``."""
+    root = api_url.rstrip("/")
+    for suffix in ("/api/v1", "/api"):
+        if root.endswith(suffix):
+            root = root[: -len(suffix)]
+            break
+    for base in (root, f"{root}/api"):
+        try:
+            resp = httpx.get(base + _WELL_KNOWN, timeout=10)
+            doc = resp.json() if resp.is_success else None
+        except (httpx.HTTPError, ValueError):
+            continue
+        if (
+            isinstance(doc, dict)
+            and str(doc.get("issuer", "")).rstrip("/") == base
+            and all(
+                urlparse(str(doc.get(key, ""))).netloc == urlparse(base).netloc
+                for key in ("token_endpoint", "device_authorization_endpoint")
+            )
+        ):
+            return {
+                "issuer": base,
+                "token": doc["token_endpoint"],
+                "device": doc["device_authorization_endpoint"],
+            }
+    return {
+        "issuer": root,
+        "token": f"{root}/oauth/token",
+        "device": f"{root}/oauth/device/code",
+    }
+
+
+def _expiring(oauth: dict[str, Any]) -> bool:
+    if not oauth.get("access_token"):
+        return True
+    expires_at = oauth.get("expires_at")
+    if not expires_at:
+        return False
+    try:
+        expiry = datetime.datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
+    except ValueError:
+        return True
+    if expiry.tzinfo is None:
+        expiry = expiry.replace(tzinfo=datetime.timezone.utc)
+    return expiry <= datetime.datetime.now(datetime.timezone.utc) + _REFRESH_LEEWAY
+
+
+def get_access_token(api_url: str) -> str:
+    """Return a valid access token from the profile, refreshing it if needed."""
+    oauth = load_profile().get("oauth") or {}
+    if not _expiring(oauth):
+        return oauth["access_token"]
+    refresh_token = oauth.get("refresh_token")
+    if not refresh_token:
+        raise click.ClickException(
+            "LangSmith session expired. Run `langgraph login` to sign in again."
+        )
+    endpoints = _oauth_endpoints(api_url)
+    try:
+        resp = httpx.post(
+            endpoints["token"],
+            data={
+                "grant_type": "refresh_token",
+                "client_id": CLIENT_ID,
+                "refresh_token": refresh_token,
+            },
+            timeout=10,
+        )
+        resp.raise_for_status()
+        token = resp.json()
+    except (httpx.HTTPError, ValueError):
+        raise click.ClickException(
+            "Could not refresh your LangSmith session. "
+            "Run `langgraph login` to sign in again."
+        ) from None
+    if not isinstance(token, dict) or not token.get("access_token"):
+        raise click.ClickException(
+            "Could not refresh your LangSmith session. "
+            "Run `langgraph login` to sign in again."
+        )
+    _save_tokens(token)
+    return token["access_token"]
+
+
+def login(api_url: str, *, open_browser: bool = True) -> None:
+    """Sign in with the OAuth device flow and store tokens in the profile."""
+    endpoints = _oauth_endpoints(api_url)
+    try:
+        resp = httpx.post(
+            endpoints["device"],
+            data={"client_id": CLIENT_ID, "resource": endpoints["issuer"]},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        device = resp.json()
+        device_code, user_code = device["device_code"], device["user_code"]
+        verification_url = (
+            device.get("verification_uri_complete") or device["verification_uri"]
+        )
+    except (httpx.HTTPError, ValueError, KeyError):
+        raise click.ClickException(
+            f"Could not start LangSmith login against {api_url}."
+        ) from None
+
+    click.echo(f"Open this URL to authorize the CLI:\n{verification_url}\n")
+    click.echo(f"Code: {user_code}\n")
+    if open_browser:
+        webbrowser.open(verification_url)
+    click.echo("Waiting for authorization...")
+
+    interval = max(int(device.get("interval") or 5), 5)
+    deadline = time.monotonic() + int(device.get("expires_in") or 600)
+    while time.monotonic() < deadline:
+        time.sleep(interval)
+        try:
+            resp = httpx.post(
+                endpoints["token"],
+                data={
+                    "grant_type": _DEVICE_GRANT,
+                    "client_id": CLIENT_ID,
+                    "device_code": device_code,
+                    "resource": endpoints["issuer"],
+                },
+                timeout=10,
+            )
+            body = resp.json()
+        except (httpx.HTTPError, ValueError):
+            continue
+        if resp.is_success and body.get("access_token"):
+            _save_tokens(body, api_url)
+            click.secho("Logged in to LangSmith.", fg="green")
+            return
+        error = body.get("error") if isinstance(body, dict) else None
+        if error == "slow_down":
+            interval += 5
+        elif error != "authorization_pending":
+            raise click.ClickException(f"LangSmith login failed: {error or 'error'}.")
+    raise click.ClickException("Timed out waiting for LangSmith login.")

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -578,12 +578,13 @@ def test_a_deployed_result_prints_the_deployment_url(
 class TestCreateHostBackendClientNoInput:
     def test_raises_when_no_api_key_and_no_input(self, monkeypatch, tmp_path):
 
+        monkeypatch.setenv("LANGSMITH_CONFIG_FILE", str(tmp_path / "missing.json"))
         monkeypatch.setattr(deploy_mod, "_no_input", True)
         monkeypatch.delenv("LANGSMITH_API_KEY", raising=False)
         monkeypatch.delenv("LANGCHAIN_API_KEY", raising=False)
         monkeypatch.delenv("LANGGRAPH_HOST_API_KEY", raising=False)
 
-        with pytest.raises(click.ClickException, match="API key"):
+        with pytest.raises(click.ClickException, match="credentials"):
             _create_host_backend_client(
                 host_url="https://api.example.com",
                 api_key=None,

libs/cli/tests/unit_tests/test_langsmith_auth.py

@@ -0,0 +1,191 @@
+import json
+from datetime import datetime, timedelta, timezone
+
+import click
+import httpx
+import pytest
+
+from langgraph_cli import deploy as deploy_mod
+from langgraph_cli import langsmith_auth
+from langgraph_cli.deploy import _create_host_backend_client
+from langgraph_cli.host_backend import HostBackendClient
+
+API_URL = "https://api.smith.langchain.com"
+
+
+def _iso(delta: timedelta) -> str:
+    return (datetime.now(timezone.utc) + delta).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+@pytest.fixture
+def config_file(tmp_path, monkeypatch):
+    path = tmp_path / "config.json"
+    monkeypatch.setenv("LANGSMITH_CONFIG_FILE", str(path))
+    monkeypatch.delenv("LANGSMITH_PROFILE", raising=False)
+    for name in ("LANGGRAPH_HOST_API_KEY", "LANGSMITH_API_KEY", "LANGCHAIN_API_KEY"):
+        monkeypatch.delenv(name, raising=False)
+    return path
+
+
+def _write_profile(path, **oauth):
+    path.write_text(
+        json.dumps(
+            {
+                "current_profile": "default",
+                "profiles": {
+                    "default": {
+                        "api_url": API_URL,
+                        "workspace_id": "ws-1",
+                        "oauth": oauth,
+                    }
+                },
+            }
+        )
+    )
+
+
+def _no_discovery(monkeypatch):
+    def get(*_a, **_k):
+        raise httpx.ConnectError("no discovery")
+
+    monkeypatch.setattr(httpx, "get", get)
+
+
+def test_valid_token_is_returned_without_network(config_file, monkeypatch):
+    _write_profile(
+        config_file, access_token="tok", expires_at=_iso(timedelta(minutes=5))
+    )
+
+    def boom(*_a, **_k):
+        raise AssertionError("no network expected")
+
+    monkeypatch.setattr(httpx, "get", boom)
+    monkeypatch.setattr(httpx, "post", boom)
+
+    assert langsmith_auth.get_access_token(API_URL) == "tok"
+
+
+def test_expired_token_is_refreshed_and_saved(config_file, monkeypatch):
+    _write_profile(
+        config_file,
+        access_token="old",
+        refresh_token="refresh-1",
+        expires_at=_iso(timedelta(seconds=-5)),
+    )
+    _no_discovery(monkeypatch)
+    seen = {}
+
+    def post(url, data, timeout):
+        seen.update(url=url, data=data)
+        return httpx.Response(
+            200,
+            json={
+                "access_token": "new",
+                "refresh_token": "refresh-2",
+                "expires_in": 300,
+            },
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx, "post", post)
+
+    assert langsmith_auth.get_access_token(API_URL) == "new"
+    assert seen["url"] == f"{API_URL}/oauth/token"
+    assert seen["data"]["refresh_token"] == "refresh-1"
+    saved = json.loads(config_file.read_text())["profiles"]["default"]["oauth"]
+    assert saved["access_token"] == "new"
+    assert saved["refresh_token"] == "refresh-2"
+
+
+def test_expired_token_without_refresh_token_asks_to_log_in(config_file):
+    _write_profile(
+        config_file, access_token="old", expires_at=_iso(timedelta(seconds=-5))
+    )
+    with pytest.raises(click.ClickException, match="langgraph login"):
+        langsmith_auth.get_access_token(API_URL)
+
+
+def test_login_stores_tokens_after_pending(config_file, monkeypatch):
+    _no_discovery(monkeypatch)
+    monkeypatch.setattr(langsmith_auth.time, "sleep", lambda _s: None)
+    monkeypatch.setattr(langsmith_auth.webbrowser, "open", lambda _u: True)
+    token_responses = iter(
+        [
+            {"error": "authorization_pending"},
+            {"access_token": "tok", "refresh_token": "r", "expires_in": 300},
+        ]
+    )
+
+    def post(url, data, timeout):
+        request = httpx.Request("POST", url)
+        if url.endswith("/oauth/device/code"):
+            return httpx.Response(
+                200,
+                json={
+                    "device_code": "dc",
+                    "user_code": "ABCD",
+                    "verification_uri": "https://example.com/activate",
+                    "expires_in": 600,
+                    "interval": 5,
+                },
+                request=request,
+            )
+        body = next(token_responses)
+        return httpx.Response(
+            400 if "error" in body else 200, json=body, request=request
+        )
+
+    monkeypatch.setattr(httpx, "post", post)
+
+    langsmith_auth.login(API_URL)
+
+    profile = json.loads(config_file.read_text())["profiles"]["default"]
+    assert profile["api_url"] == API_URL
+    assert profile["oauth"]["access_token"] == "tok"
+
+
+def test_client_sends_bearer_token_per_request():
+    seen = []
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen.append(dict(req.headers))
+        return httpx.Response(200, json={"resources": []})
+
+    tokens = iter(["t1", "t2"])
+    client = HostBackendClient(
+        "https://api.example.com",
+        None,
+        access_token=lambda: next(tokens),
+        transport=httpx.MockTransport(handler),
+    )
+    client.list_deployments()
+    client.list_deployments()
+
+    assert [h["authorization"] for h in seen] == ["Bearer t1", "Bearer t2"]
+    assert "x-api-key" not in seen[0]
+
+
+def test_factory_uses_logged_in_profile(config_file, monkeypatch):
+    _write_profile(
+        config_file, access_token="tok", expires_at=_iso(timedelta(minutes=5))
+    )
+    monkeypatch.setattr(deploy_mod, "_no_input", True)
+    monkeypatch.delenv("LANGSMITH_TENANT_ID", raising=False)
+    monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
+
+    client = _create_host_backend_client(
+        host_url="https://api.example.com", api_key=None, env_vars={}
+    )
+
+    assert client._access_token is not None
+    assert client._access_token() == "tok"
+    assert client._client.headers["x-tenant-id"] == "ws-1"
+    assert "x-api-key" not in client._client.headers
+
+
+def test_factory_without_credentials_and_no_input_fails(config_file, monkeypatch):
+    monkeypatch.setattr(deploy_mod, "_no_input", True)
+    with pytest.raises(click.ClickException, match="langgraph login"):
+        _create_host_backend_client(
+            host_url="https://api.example.com", api_key=None, env_vars={}
+        )

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/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/_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/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
@@ -2882,14 +2885,12 @@ class Pregel(
                 # 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)
+                    h for h in callback_manager.handlers if not _is_stream_handler(h)
                 ]
                 callback_manager.inheritable_handlers = [
                     h
                     for h in callback_manager.inheritable_handlers
-                    if not _is_stream_mode_handler(h)
+                    if not _is_stream_handler(h)
                 ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
@@ -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
@@ -3301,14 +3300,12 @@ class Pregel(
                 # 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)
+                    h for h in callback_manager.handlers if not _is_stream_handler(h)
                 ]
                 callback_manager.inheritable_handlers = [
                     h
                     for h in callback_manager.inheritable_handlers
-                    if not _is_stream_mode_handler(h)
+                    if not _is_stream_handler(h)
                 ]
             if "ls_integration" not in callback_manager.metadata:
                 callback_manager.add_metadata({"ls_integration": "langgraph"})
@@ -3323,10 +3320,20 @@ 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, StreamToolCallHandler)
+                        )
+                    ),
+                    False,
+                )
+                if _StreamingCallbackHandler is not None
+                else False
             )
             if checkpointer is None and durability is not None:
                 warnings.warn(
@@ -3407,8 +3414,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
@@ -4462,7 +4467,7 @@ def _coerce_checkpoint_values(payload: Any, mapper: Callable[[Any], Any]) -> Non
         payload["values"] = mapper(payload["values"])
 
 
-def _is_stream_mode_handler(handler: object) -> bool:
+def _is_stream_handler(handler: object) -> bool:
     return isinstance(handler, (StreamMessagesHandler, StreamToolCallHandler))
 
 

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.18"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

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_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.18"
 source = { editable = "." }
 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" },

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.19"
+version = "1.2.18"
 source = { editable = "../langgraph" }
 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.10"
 
 __all__ = [
     "Auth",

libs/sdk-py/uv.lock

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