fix(cli): defer parent agent env validation

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.31"
+__version__ = "0.4.32"

libs/cli/langgraph_cli/deploy.py

@@ -12,6 +12,7 @@ from collections.abc import Callable, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
+from functools import partial
 from typing import Protocol, TypeVar
 
 import click
@@ -1690,11 +1691,18 @@ OPT_HOST_URL = click.option(
 )
 
 OPT_AGENT_ID = click.option(
-    "--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)."
+    "--agent-id",
+    envvar="LANGSMITH_AGENT_ID",
+    show_envvar=True,
+    help="Logical agent ID (requires agent mode enabled for the tenant).",
 )
 
-OPT_AGENT_ENVIRONMENT = click.option(
-    "--environment",
+OPT_AGENT_ENVIRONMENT = partial(
+    click.option,
+    "--agent-environment",
+    "environment",
+    envvar="LANGSMITH_AGENT_ENVIRONMENT",
+    show_envvar=True,
     type=click.Choice(["development", "staging", "production"]),
     help="Agent environment (requires agent mode enabled for the tenant).",
 )
@@ -1798,7 +1806,9 @@ def _deploy_base_options(
             OPT_HOST_API_KEY,
             OPT_HOST_DEPLOYMENT_NAME,
             OPT_AGENT_ID,
-            OPT_AGENT_ENVIRONMENT,
+            OPT_AGENT_ENVIRONMENT()
+            if include_docker_args
+            else OPT_AGENT_ENVIRONMENT(type=str),
             click.option(
                 "--deployment-id",
                 help=(
@@ -1930,6 +1940,12 @@ def deploy(ctx: click.Context, **_: object):
     # otherwise, we return None here and click will proceed to actually run the subcommand (list or delete)
     if ctx.invoked_subcommand is not None:
         return
+    environment_param = next(
+        param for param in _deploy_cmd.params if param.name == "environment"
+    )
+    ctx.params["environment"] = environment_param.type_cast_value(
+        ctx, ctx.params["environment"]
+    )
     if (
         ctx.params.get("agent_id") is not None
         or ctx.params.get("environment") is not None
@@ -1982,13 +1998,14 @@ 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 --environment are required together."
+                "--agent-id and --agent-environment are required together."
             )
         if name is not None or deployment_id is not None:
             raise click.UsageError(
-                "--agent-id and --environment cannot be combined with --name or --deployment-id."
+                "--agent-id and --agent-environment cannot be combined with --name or --deployment-id."
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2124,7 +2141,7 @@ def _deploy_cmd(
 @OPT_HOST_API_KEY
 @OPT_HOST_URL
 @OPT_AGENT_ID
-@OPT_AGENT_ENVIRONMENT
+@OPT_AGENT_ENVIRONMENT()
 @click.option(
     "--name-contains",
     default="",
@@ -2138,6 +2155,11 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
+    if agent_id is not None or environment is not None:
+        click.secho(
+            "Note: --agent-id and --agent-environment flags are in private beta",
+            fg="yellow",
+        )
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}

libs/cli/tests/unit_tests/test_deploy_agents.py

@@ -58,7 +58,7 @@ AGENT_ARGS = [
     "deploy",
     "--agent-id",
     "customer-support",
-    "--environment",
+    "--agent-environment",
     "staging",
     "--remote",
     "--no-wait",

libs/langgraph/langgraph/pregel/main.py

@@ -1146,8 +1146,6 @@ class Pregel(
         self,
         config: RunnableConfig,
         saved: CheckpointTuple | None,
-        *,
-        saver: BaseCheckpointSaver,
         recurse: BaseCheckpointSaver | None = None,
         apply_pending_writes: bool = False,
     ) -> StateSnapshot:
@@ -1171,7 +1169,9 @@ class Pregel(
         channels, managed = channels_from_checkpoint(
             self.channels,
             saved.checkpoint,
-            saver=saver,
+            saver=self.checkpointer
+            if isinstance(self.checkpointer, BaseCheckpointSaver)
+            else None,
             config=saved.config,
         )
         # tasks for this checkpoint
@@ -1186,7 +1186,11 @@ class Pregel(
             stop,
             for_execution=True,
             store=self.store,
-            checkpointer=saver,
+            checkpointer=(
+                self.checkpointer
+                if isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None
+            ),
             manager=None,
         )
         # get the subgraphs
@@ -1265,8 +1269,6 @@ class Pregel(
         self,
         config: RunnableConfig,
         saved: CheckpointTuple | None,
-        *,
-        saver: BaseCheckpointSaver,
         recurse: BaseCheckpointSaver | None = None,
         apply_pending_writes: bool = False,
     ) -> StateSnapshot:
@@ -1290,7 +1292,9 @@ class Pregel(
         channels, managed = await achannels_from_checkpoint(
             self.channels,
             saved.checkpoint,
-            saver=saver,
+            saver=self.checkpointer
+            if isinstance(self.checkpointer, BaseCheckpointSaver)
+            else None,
             config=saved.config,
         )
         # tasks for this checkpoint
@@ -1305,7 +1309,11 @@ class Pregel(
             stop,
             for_execution=True,
             store=self.store,
-            checkpointer=saver,
+            checkpointer=(
+                self.checkpointer
+                if isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None
+            ),
             manager=None,
         )
         # get the subgraphs
@@ -1421,7 +1429,6 @@ class Pregel(
         return self._prepare_state_snapshot(
             config,
             saved,
-            saver=checkpointer,
             recurse=checkpointer if subgraphs else None,
             apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
         )
@@ -1466,7 +1473,6 @@ class Pregel(
         return await self._aprepare_state_snapshot(
             config,
             saved,
-            saver=checkpointer,
             recurse=checkpointer if subgraphs else None,
             apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
         )
@@ -1521,7 +1527,7 @@ class Pregel(
             checkpointer.list(config, before=before, limit=limit, filter=filter)
         ):
             yield self._prepare_state_snapshot(
-                checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer
+                checkpoint_tuple.config, checkpoint_tuple
             )
 
     async def aget_state_history(
@@ -1578,7 +1584,7 @@ class Pregel(
             )
         ]:
             yield await self._aprepare_state_snapshot(
-                checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer
+                checkpoint_tuple.config, checkpoint_tuple
             )
 
     def bulk_update_state(
@@ -1660,7 +1666,10 @@ class Pregel(
             channels, managed = channels_from_checkpoint(
                 self.channels,
                 checkpoint,
-                saver=checkpointer if saved is not None else None,
+                saver=self.checkpointer
+                if saved is not None
+                and isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None,
                 config=saved.config if saved is not None else None,
             )
             values, as_node = updates[0][:2]
@@ -2123,7 +2132,10 @@ class Pregel(
             channels, managed = await achannels_from_checkpoint(
                 self.channels,
                 checkpoint,
-                saver=checkpointer if saved is not None else None,
+                saver=self.checkpointer
+                if saved is not None
+                and isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None,
                 config=saved.config if saved is not None else None,
             )
             values, as_node = updates[0][:2]

libs/langgraph/tests/test_delta_channel_subgraph.py

@@ -1,253 +0,0 @@
-import operator
-from typing import Annotated, Any, Literal
-
-import pytest
-from langgraph.checkpoint.base import BaseCheckpointSaver
-from typing_extensions import TypedDict
-
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph import END, START, StateGraph
-
-pytestmark = pytest.mark.anyio
-
-
-def _extend(state: list | None, writes: list[Any]) -> list:
-    out = list(state or [])
-    for write in writes:
-        out.extend(write if isinstance(write, list) else [write])
-    return out
-
-
-def _state_schema(snapshot_frequency: int = 1000) -> type:
-    class State(TypedDict, total=False):
-        delta: Annotated[
-            list, DeltaChannel(_extend, snapshot_frequency=snapshot_frequency)
-        ]
-        plain: Annotated[list, operator.add]
-
-    return State
-
-
-def _both(*items: str) -> dict:
-    return {"delta": list(items), "plain": list(items)}
-
-
-def _child_builder(*, snapshot_frequency: int = 1000) -> StateGraph:
-    builder = StateGraph(_state_schema(snapshot_frequency))
-    builder.add_node("a", lambda state: _both("a1"))
-    builder.add_node("b", lambda state: _both("b1", "b2"))
-    builder.add_edge(START, "a")
-    builder.add_edge("a", "b")
-    builder.add_edge("b", END)
-    return builder
-
-
-def _wrap(
-    inner: StateGraph,
-    *,
-    checkpointer: bool | None = None,
-    interrupt_before: list[str] | None = None,
-) -> StateGraph:
-    builder = StateGraph(inner.state_schema)
-    builder.add_node(
-        "child",
-        inner.compile(checkpointer=checkpointer, interrupt_before=interrupt_before),
-    )
-    builder.add_edge(START, "child")
-    builder.add_edge("child", END)
-    return builder
-
-
-def _nested_app(
-    checkpointer: BaseCheckpointSaver,
-    *,
-    depth: int = 1,
-    snapshot_frequency: int = 1000,
-    pause_before_b: bool = False,
-    subgraph_checkpointer: bool | None = None,
-) -> Any:
-    graph = _child_builder(snapshot_frequency=snapshot_frequency)
-    for _ in range(depth):
-        graph = _wrap(
-            graph,
-            checkpointer=subgraph_checkpointer,
-            interrupt_before=["b"] if pause_before_b else None,
-        )
-    return graph.compile(checkpointer=checkpointer)
-
-
-def _scoped(config: dict, namespace: str) -> dict:
-    return {"configurable": {**config["configurable"], "checkpoint_ns": namespace}}
-
-
-def _child_namespace(app: Any, config: dict, *, depth: int = 1) -> str:
-    namespace = ""
-    for level in range(depth):
-        scoped = _scoped(config, namespace) if namespace else config
-        namespace = next(
-            (
-                task.state["configurable"]["checkpoint_ns"]
-                for snapshot in app.get_state_history(scoped)
-                for task in snapshot.tasks
-                if task.name == "child" and isinstance(task.state, dict)
-            ),
-            "",
-        )
-        assert namespace, f"no `child` subgraph task at nesting level {level}"
-    return namespace
-
-
-async def _achild_namespace(app: Any, config: dict) -> str:
-    async for snapshot in app.aget_state_history(config):
-        for task in snapshot.tasks:
-            if task.name == "child" and isinstance(task.state, dict):
-                return task.state["configurable"]["checkpoint_ns"]
-    raise AssertionError("no `child` subgraph task")
-
-
-HISTORY = [_both("a1", "b1", "b2"), _both("a1"), _both(), _both()]
-
-
-def test_subgraph_get_state(sync_checkpointer: BaseCheckpointSaver) -> None:
-    app = _nested_app(sync_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    child = _scoped(config, _child_namespace(app, config))
-
-    assert app.get_state(config).values == _both("a1", "b1", "b2")
-    assert app.get_state(child).values == _both("a1", "b1", "b2")
-
-
-async def test_subgraph_aget_state(async_checkpointer: BaseCheckpointSaver) -> None:
-    app = _nested_app(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    await app.ainvoke({}, config)
-
-    child = _scoped(config, await _achild_namespace(app, config))
-
-    assert (await app.aget_state(child)).values == _both("a1", "b1", "b2")
-
-
-def test_subgraph_get_state_history(sync_checkpointer: BaseCheckpointSaver) -> None:
-    app = _nested_app(sync_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    child = _scoped(config, _child_namespace(app, config))
-
-    assert [s.values for s in app.get_state_history(child)] == HISTORY
-
-
-async def test_subgraph_aget_state_history(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    await app.ainvoke({}, config)
-
-    child = _scoped(config, await _achild_namespace(app, config))
-
-    assert [s.values async for s in app.aget_state_history(child)] == HISTORY
-
-
-def test_doubly_nested_subgraph_get_state(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(sync_checkpointer, depth=2)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    child = _scoped(config, _child_namespace(app, config, depth=2))
-
-    assert app.get_state(child).values == _both("a1", "b1", "b2")
-
-
-@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
-def test_interrupted_subgraph_task_state(
-    sync_checkpointer: BaseCheckpointSaver,
-    persistence: Literal["per-invocation", "per-thread"],
-) -> None:
-    app = _nested_app(
-        sync_checkpointer,
-        pause_before_b=True,
-        subgraph_checkpointer=True if persistence == "per-thread" else None,
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    (task,) = app.get_state(config, subgraphs=True).tasks
-
-    assert task.state.values == _both("a1")
-
-
-@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
-async def test_interrupted_subgraph_task_state_async(
-    async_checkpointer: BaseCheckpointSaver,
-    persistence: Literal["per-invocation", "per-thread"],
-) -> None:
-    app = _nested_app(
-        async_checkpointer,
-        pause_before_b=True,
-        subgraph_checkpointer=True if persistence == "per-thread" else None,
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    await app.ainvoke({}, config)
-
-    (task,) = (await app.aget_state(config, subgraphs=True)).tasks
-
-    assert task.state.values == _both("a1")
-
-
-def test_subgraph_update_state_keeps_history(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(sync_checkpointer, snapshot_frequency=2)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    child = _scoped(config, _child_namespace(app, config))
-    app.update_state(child, _both("manual"))
-
-    assert app.get_state(child).values == _both("a1", "b1", "b2", "manual")
-
-
-async def test_subgraph_aupdate_state_keeps_history(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(async_checkpointer, snapshot_frequency=2)
-    config = {"configurable": {"thread_id": "1"}}
-    await app.ainvoke({}, config)
-
-    child = _scoped(config, await _achild_namespace(app, config))
-    await app.aupdate_state(child, _both("manual"))
-
-    assert (await app.aget_state(child)).values == _both("a1", "b1", "b2", "manual")
-
-
-def test_stateless_subgraph_persists_nothing(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(sync_checkpointer, subgraph_checkpointer=False)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    child_tasks = [
-        task
-        for snapshot in app.get_state_history(config)
-        for task in snapshot.tasks
-        if task.name == "child" and isinstance(task.state, dict)
-    ]
-
-    assert child_tasks == []
-    assert app.get_state(config).values == _both("a1", "b1", "b2")
-
-
-def test_completed_subgraph_exposes_no_task_state(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    app = _nested_app(sync_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    app.invoke({}, config)
-
-    assert app.get_state(config, subgraphs=True).tasks == ()