fix(cli): remove outdated deploy private-beta notice (#9506)

libs/cli/langgraph_cli/deploy.py

@@ -2431,7 +2431,6 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
-        em.note("Note: --agent-id and --agent-environment flags are in private beta")
         if not agent_id or not agent_id.strip() or not environment:
             raise click.UsageError(
                 "--agent-id and --agent-environment are required together."

libs/cli/tests/unit_tests/test_deploy_agents.py

@@ -85,6 +85,8 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch):
     assert "name" not in payload
     assert build.call_args.kwargs["deployment_id"] == "runtime-id"
     assert "server-name" in result.output
+    assert "'langgraph deploy' is in beta" in result.output
+    assert "private beta" not in result.output
     assert (tmp_path / ".env").read_text() == "LANGSMITH_DEPLOYMENT_NAME=legacy\n"
 
 

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

@@ -175,7 +175,6 @@ from langgraph.runtime import (
     RunControl,
     Runtime,
     ServerInfo,
-    _RunExits,
 )
 from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import StreamTransformer
@@ -2967,8 +2966,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
@@ -3407,8 +3404,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

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