fix(cli): defer parent agent env validation

Follow-up to #8482. `langgraph deploy --push-to` can now create a
deployment in a workspace that
deploys through a listener in the customer's own cluster, which is the
hybrid case. Before this,
creation in such a workspace was impossible from the CLI: the control
plane rejected it and the CLI
told the user to go and create the deployment in the UI first.

## Changes
- Smart Auto-Placement: The CLI now proactively checks your workspace.
If you only have one listener and one Kubernetes namespace configured
(and are using the managed cloud control plane), it automatically routes
your deployment there. No extra flags needed.
- New Disambiguation Flags: If your workspace has multiple listeners or
namespaces, the CLI will ask you to choose. You can now pass
--listener-id and --k8s-namespace to tell it exactly where to deploy.
- Failing Fast: The CLI now validates your listener and namespace
choices before it starts building and pushing the heavy Docker image. If
you provide an invalid ID, it stops immediately instead of wasting your
time and bandwidth.
- Fixed a Duplication Bug: Previously, if you had many deployments with
similar names, a pagination issue could hide your existing deployment
from the CLI, causing it to accidentally create a duplicate. The CLI now
queries the server for the exact deployment name to guarantee this
doesn't happen.
- Cleaner Errors: Error messages from the control plane are now stripped
of their clunky HTTP envelopes so you get clear, readable sentences when
something goes wrong.

## Testing

Deployment on 3 paths, hybrid, self-hosted, nominal

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -267,61 +267,6 @@ async def test_history_seed_ancestor_own_writes_are_replayed(
     )
 
 
-# Every uuid4 `build_delta_chain` tags its own writes with sorts between these
-# two, so task_id order is fixed and always disagrees with task_path order.
-TASK_ID_SORTS_FIRST = "00000000-0000-0000-0000-000000000000"
-TASK_ID_SORTS_LAST = "zzzzzzzz-0000-0000-0000-000000000000"
-
-
-async def test_history_orders_parallel_writes_by_task_path(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Writes from parallel tasks replay in task_path order, not task_id order."""
-    configs = await build_delta_chain(
-        saver,
-        thread_id=str(uuid4()),
-        channel="ch",
-        snapshots_at_steps=[0],
-        total_steps=3,
-    )
-    step_1, head = configs[1], configs[2]
-    await saver.aput_writes(
-        step_1, [("ch", "second")], TASK_ID_SORTS_FIRST, "~pull, 02"
-    )
-    await saver.aput_writes(step_1, [("ch", "first")], TASK_ID_SORTS_LAST, "~pull, 01")
-
-    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    values = [w[2] for w in result["ch"]["writes"]]
-    assert values == [1, "first", "second"], (
-        f"Expected task_path order [1, 'first', 'second'], got {values}. "
-        "Ordering by (task_id, idx) alone yields [1, 'second', 'first']."
-    )
-
-
-async def test_history_orders_pathless_writes_first(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Writes stored without a task_path (graph input) replay before task writes."""
-    configs = await build_delta_chain(
-        saver,
-        thread_id=str(uuid4()),
-        channel="ch",
-        snapshots_at_steps=[0],
-        total_steps=3,
-    )
-    step_1, head = configs[1], configs[2]
-    await saver.aput_writes(
-        step_1, [("ch", "from_node")], TASK_ID_SORTS_FIRST, "~pull, a"
-    )
-    await saver.aput_writes(step_1, [("ch", "from_input")], TASK_ID_SORTS_LAST)
-
-    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    values = [w[2] for w in result["ch"]["writes"]]
-    assert values == [1, "from_input", "from_node"], (
-        f"Expected pathless writes first, got {values}"
-    )
-
-
 ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_returns_writes_oldest_first,
     test_history_seed_is_nearest_snapshot,
@@ -331,8 +276,6 @@ ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_walk_to_root_no_seed,
     test_history_migration_plain_value_as_seed,
     test_history_seed_ancestor_own_writes_are_replayed,
-    test_history_orders_parallel_writes_by_task_path,
-    test_history_orders_pathless_writes_first,
 ]
 
 

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -168,7 +168,6 @@ class _DeltaStage2Row(TypedDict, total=False):
     type: str | None
     blob: bytes | None
     task_id: str | None  # "w" rows only
-    task_path: str | None  # "w" rows only
     idx: int | None  # "w" rows only
     version: str | None  # "b" rows only
 
@@ -320,7 +319,7 @@ def _build_delta_stage2_sql(
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
-            "type, blob, task_id, task_path, idx, NULL::text AS version "
+            "type, blob, task_id, idx, NULL::text AS version "
             "FROM checkpoint_writes "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND checkpoint_id = ANY(%s)"
@@ -328,8 +327,7 @@ def _build_delta_stage2_sql(
     for _ in channels_with_seed:
         branches.append(
             "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
-            "type, blob, NULL::text AS task_id, NULL::text AS task_path, "
+            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
-            "NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"
@@ -494,11 +492,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         stored value, or when the seed blob is sentinel "empty" — in both cases
         the consumer treats absence as "start empty".
         """
-        # writes_by_ch_by_cid[channel][cid] = list of
+        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
-        # (type, blob, task_id, idx, task_path)
+        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-        writes_by_ch_by_cid: dict[
+            ch: {} for ch in channels
-            str, dict[str, list[tuple[str, bytes, str, int, str]]]
+        }
-        ] = {ch: {} for ch in channels}
         # seed_blob_by_ver[(channel, version)] = (type, blob)
         seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
@@ -509,14 +506,8 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
-                        "tuple[str, bytes, str, int, str]",
+                        "tuple[str, bytes, str, int]",
-                        (
+                        (r["type"], r["blob"], r["task_id"], r["idx"]),
-                            r["type"],
-                            r["blob"],
-                            r["task_id"],
-                            r["idx"],
-                            r["task_path"],
-                        ),
                     )
                 )
             else:  # kind == "b"
@@ -525,10 +516,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # Sort writes per (channel, cid) newest-first by (task_path, task_id, idx)
+        # Sort writes per (channel, cid) newest-first by (task_id, idx)
         for cid_map in writes_by_ch_by_cid.values():
             for ws in cid_map.values():
-                ws.sort(key=lambda w: (w[4], w[2], w[3]), reverse=True)
+                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
@@ -538,9 +529,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             collected: list[PendingWrite] = []
             cid_writes = writes_by_ch_by_cid.get(ch, {})
             for cid in chain_cids:
-                for type_tag, write_blob, task_id, _idx, _path in cid_writes.get(
+                for type_tag, write_blob, task_id, _idx in cid_writes.get(cid, []):
-                    cid, []
-                ):
                     val = self.serde.loads_typed((type_tag, write_blob))
                     collected.append((task_id, ch, val))
             collected.reverse()

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -154,7 +154,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                 checkpoint_ns TEXT NOT NULL DEFAULT '',
                 checkpoint_id TEXT NOT NULL,
                 task_id TEXT NOT NULL,
-                task_path TEXT NOT NULL DEFAULT '',
                 idx INTEGER NOT NULL,
                 channel TEXT NOT NULL,
                 type TEXT,
@@ -163,15 +162,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             );
             """
         )
-        # sqlite has no ADD COLUMN IF NOT EXISTS; this migrates databases
-        # created before `task_path` existed and is a no-op on the rest.
-        try:
-            self.conn.execute(
-                "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
-            )
-        except sqlite3.OperationalError as e:
-            if "duplicate column name" not in str(e):
-                raise
 
         self.is_setup = True
 
@@ -470,9 +460,9 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         with self.cursor() as cur:
             cur.executemany(
@@ -483,7 +473,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
-                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -579,7 +568,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                     )
                 cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]", cur.fetchall()
+                    "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
                 )
             else:
                 stage2_rows = []

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_delta.py

@@ -57,7 +57,7 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
     for n in chain_lens:
         cid_placeholders = ",".join("?" * n)
         branches.append(
-            "SELECT checkpoint_id, channel, task_id, idx, type, value, task_path "
+            "SELECT checkpoint_id, channel, task_id, idx, type, value "
             "FROM writes "
             "WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ? "
             f"AND checkpoint_id IN ({cid_placeholders})"
@@ -130,31 +130,29 @@ def build_delta_channels_writes_history(
     chain_by_ch: Mapping[str, list[str]],
     seed_val_by_ch: Mapping[str, Any],
     seeded: set[str],
-    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes, str]],
+    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes]],
     serde: Any,
 ) -> dict[str, DeltaChannelHistory]:
     """Demux stage-2 rows per channel; produce per-channel histories.
 
-    Stage-2 rows are
+    Stage-2 rows are `(checkpoint_id, channel, task_id, idx, type, value)`.
-    `(checkpoint_id, channel, task_id, idx, type, value, task_path)`.
+    Final write order is oldest→newest globally and `(task_id, idx)` within
-    Final write order is oldest→newest globally and
+    a checkpoint, matching the contract on `DeltaChannelHistory.writes`.
-    `(task_path, task_id, idx)` within a checkpoint, matching the contract
-    on `DeltaChannelHistory.writes`.
 
     `seed` is omitted when the walk reached a true root with no snapshot
     found (channel never entered `seeded`); consumers treat absence as
     "start empty".
     """
-    writes_by_ch_by_cid: dict[
+    writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-        str, dict[str, list[tuple[str, bytes, str, int, str]]]
+        ch: {} for ch in channels
-    ] = {ch: {} for ch in channels}
+    }
-    for cid, ch, task_id, idx, type_tag, value_blob, task_path in stage2_rows:
+    for cid, ch, task_id, idx, type_tag, value_blob in stage2_rows:
         writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
-            (type_tag, value_blob, task_id, idx, task_path)
+            (type_tag, value_blob, task_id, idx)
         )
     for cid_map in writes_by_ch_by_cid.values():
         for ws in cid_map.values():
-            ws.sort(key=lambda w: (w[4], w[2], w[3]))
+            ws.sort(key=lambda w: (w[2], w[3]))
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
@@ -163,7 +161,7 @@ def build_delta_channels_writes_history(
         collected: list[PendingWrite] = []
         # Chain is newest-first; iterate oldest-first for the public order.
         for cid in reversed(chain_cids):
-            for type_tag, value_blob, task_id, _idx, _path in cid_writes.get(cid, []):
+            for type_tag, value_blob, task_id, _idx in cid_writes.get(cid, []):
                 collected.append(
                     (task_id, ch, serde.loads_typed((type_tag, value_blob)))
                 )

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@@ -331,7 +331,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     checkpoint_ns TEXT NOT NULL DEFAULT '',
                     checkpoint_id TEXT NOT NULL,
                     task_id TEXT NOT NULL,
-                    task_path TEXT NOT NULL DEFAULT '',
                     idx INTEGER NOT NULL,
                     channel TEXT NOT NULL,
                     type TEXT,
@@ -342,17 +341,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             ):
                 await self.conn.commit()
 
-            # sqlite has no ADD COLUMN IF NOT EXISTS; this migrates databases
-            # created before `task_path` existed and is a no-op on the rest.
-            try:
-                await self.conn.execute(
-                    "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
-                )
-                await self.conn.commit()
-            except aiosqlite.OperationalError as e:
-                if "duplicate column name" not in str(e):
-                    raise
-
             self.is_setup = True
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
@@ -588,9 +576,9 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         await self.setup()
         async with self.lock, self.conn.cursor() as cur:
@@ -602,7 +590,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
-                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -694,7 +681,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     )
                 await cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]",
+                    "list[tuple[str, str, str, int, str, bytes]]",
                     await cur.fetchall(),
                 )
             else:

libs/checkpoint-sqlite/tests/test_writes_task_path_migration.py

@@ -1,87 +0,0 @@
-import sqlite3
-from pathlib import Path
-
-import aiosqlite
-import pytest
-from langgraph.checkpoint.base import empty_checkpoint
-
-from langgraph.checkpoint.sqlite import SqliteSaver
-from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-
-WRITES_BEFORE_TASK_PATH = """
-CREATE TABLE writes (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    checkpoint_id TEXT NOT NULL,
-    task_id TEXT NOT NULL,
-    idx INTEGER NOT NULL,
-    channel TEXT NOT NULL,
-    type TEXT,
-    value BLOB,
-    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
-);
-INSERT INTO writes VALUES ('t', '', 'c', 'old-task', 0, 'ch', 'null', X'');
-"""
-
-
-@pytest.fixture
-def legacy_db(tmp_path: Path) -> Path:
-    db = tmp_path / "legacy.sqlite"
-    with sqlite3.connect(db) as conn:
-        conn.executescript(WRITES_BEFORE_TASK_PATH)
-    return db
-
-
-def test_setup_migrates_legacy_writes_table_repeatably(legacy_db: Path) -> None:
-    for _ in range(2):
-        with SqliteSaver.from_conn_string(str(legacy_db)) as saver:
-            saver.setup()
-            rows = saver.conn.execute(
-                "SELECT task_id, task_path FROM writes"
-            ).fetchall()
-            assert rows == [("old-task", "")]
-
-
-@pytest.mark.parametrize("fresh", [True, False], ids=["fresh", "legacy"])
-def test_put_writes_persists_task_path(
-    tmp_path: Path, legacy_db: Path, fresh: bool
-) -> None:
-    db = tmp_path / "fresh.sqlite" if fresh else legacy_db
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        config = saver.put(
-            {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
-            empty_checkpoint(),
-            {},
-            {},
-        )
-        saver.put_writes(config, [("ch", "v")], "task-1", "~__pregel_pull, node")
-        stored = saver.conn.execute(
-            "SELECT task_path FROM writes WHERE task_id = 'task-1'"
-        ).fetchall()
-    assert stored == [("~__pregel_pull, node",)]
-
-
-async def test_async_setup_migrates_legacy_writes_table_repeatably(
-    legacy_db: Path,
-) -> None:
-    for _ in range(2):
-        async with AsyncSqliteSaver.from_conn_string(str(legacy_db)) as saver:
-            await saver.setup()
-            config = await saver.aput(
-                {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
-                empty_checkpoint(),
-                {},
-                {},
-            )
-            await saver.aput_writes(
-                config, [("ch", "v")], "task-1", "~__pregel_pull, node"
-            )
-
-    async with aiosqlite.connect(legacy_db) as conn:
-        async with conn.execute(
-            "SELECT DISTINCT task_id, task_path FROM writes ORDER BY task_id"
-        ) as cur:
-            assert await cur.fetchall() == [
-                ("old-task", ""),
-                ("task-1", "~__pregel_pull, node"),
-            ]

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -162,13 +162,6 @@ class DeltaChannelHistory(TypedDict):
       Always present; possibly empty. Already filtered to one channel.
       Writes stored at the target checkpoint itself are pending for the
       next super-step and are excluded.
-
-      Within a single checkpoint, writes are ordered by
-      `(task_path, task_id, idx)`: the order `apply_writes` applied them in
-      live. `task_id` is a hash of the path, so ordering by it permutes
-      parallel tasks writing one channel, and reducers need not be
-      order-invariant. Writes stored without a `task_path` (graph input, or
-      rows predating the column) sort first.
     * `seed` — the stored value at the nearest ancestor whose
       `channel_values[ch]` is populated. Omitted if the walk reached the
       root without finding any stored value (consumer treats absence as
@@ -618,11 +611,6 @@ class BaseCheckpointSaver(Generic[V]):
         `PostgresSaver`) override for performance; the return contract is
         fixed here.
 
-        `PendingWrite` carries no `task_path`, so this default replays each
-        checkpoint's writes in `get_tuple`'s `pending_writes` order. Savers
-        that do not return `pending_writes` ordered by
-        `(task_path, task_id, idx)` must override it.
-
         Args:
             config: Configuration identifying the target checkpoint.
             channels: Channel names to walk for. Empty → empty mapping.

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -199,8 +199,8 @@ class InMemorySaver(
                     terminated_here.add(ch)
 
             step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-            for _, (tid, ch, serialized, _) in sorted(
+            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
-                step_writes.items(), key=lambda kv: (kv[1][3], kv[0]), reverse=True
+                step_writes.items(), reverse=True
             ):
                 if ch not in remaining:
                     continue

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
@@ -26,6 +27,7 @@ from langgraph_cli.dependency_tracking import find_tracked_packages
 from langgraph_cli.docker import build_docker_image, can_build_locally
 from langgraph_cli.exec import CommandRunner, Runner, subp_exec
 from langgraph_cli.host_backend import (
+    MAX_PAGE_SIZE,
     ControlPlaneEndpoints,
     HostBackendClient,
     HostBackendError,
@@ -101,15 +103,15 @@ _NATIVE_AMD64_MACHINE = "x86_64"
 _PUSH_ATTEMPTS = 3
 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
 _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
-_LISTENER_REQUIRED_MARKER = "listener_id' is required"
-_HYBRID_LISTENER_GUIDANCE = (
-    "This workspace deploys through a listener in your own cluster, and the "
-    "control plane needs a listener ID to create a deployment. Create the "
-    "deployment once in the LangSmith UI, choosing the listener and namespace, "
-    "then re-run with --deployment-id <id>."
-)
-
 _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
+_LISTENER_REQUIRED_MARKER = "listener_id' is required"
+_LISTENERS_SHOWN = 10
+_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422})
+_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners"
+_NO_LISTENERS = (
+    "This workspace has no listeners, so --listener-id and --k8s-namespace "
+    "do not apply."
+)
 
 
 _TERMINAL_STATUSES = frozenset(
@@ -161,6 +163,134 @@ class ByAgent:
 DeploymentSelector = ById | ByName | ByAgent
 
 
+@dataclass(frozen=True, slots=True)
+class Listener:
+    id: str
+    compute_id: str
+    namespaces: tuple[str, ...]
+
+    @classmethod
+    def from_resource(cls, resource: Mapping[str, object]) -> "Listener":
+        identifier = str(resource.get("id") or "")
+        if not identifier:
+            raise HostBackendError(
+                "The control plane returned a listener without an id."
+            )
+        compute_config = resource.get("compute_config")
+        namespaces = (
+            compute_config.get("k8s_namespaces")
+            if isinstance(compute_config, Mapping)
+            else None
+        )
+        return cls(
+            identifier,
+            str(resource.get("compute_id", "")),
+            tuple(str(namespace) for namespace in namespaces)
+            if isinstance(namespaces, list)
+            else (),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class Unplaced:
+    @property
+    def summary(self) -> str:
+        return ""
+
+    def source_config(self) -> dict[str, object]:
+        return {}
+
+
+@dataclass(frozen=True, slots=True)
+class OnListener:
+    listener_id: str
+    k8s_namespace: str
+
+    @property
+    def summary(self) -> str:
+        return (
+            f"Deploying through listener {self.listener_id} "
+            f"in namespace {self.k8s_namespace}"
+        )
+
+    def source_config(self) -> dict[str, object]:
+        return {
+            "listener_id": self.listener_id,
+            "listener_config": {"k8s_namespace": self.k8s_namespace},
+        }
+
+
+Placement = Unplaced | OnListener
+
+
+@dataclass(frozen=True, slots=True)
+class RequestedPlacement:
+    listener_id: str | None = None
+    k8s_namespace: str | None = None
+
+    @property
+    def requested(self) -> bool:
+        return self.listener_id is not None or self.k8s_namespace is not None
+
+    def ensure_not_requested(self, deployment_id: str) -> None:
+        if self.requested:
+            raise click.UsageError(
+                "Listener and namespace are fixed when a deployment is created. "
+                f"Deployment {deployment_id} already exists, so drop --listener-id "
+                "and --k8s-namespace, or create a new deployment with a different "
+                "--name."
+            )
+
+    def on(self, listener: Listener) -> Placement:
+        return OnListener(listener.id, self._namespace(listener))
+
+    def among(self, listeners: Sequence[Listener]) -> Placement:
+        if not listeners:
+            if self.requested:
+                raise click.UsageError(_NO_LISTENERS)
+            return Unplaced()
+        if len(listeners) > 1:
+            raise click.UsageError(
+                "This workspace has several listeners. Choose one with "
+                f"--listener-id:\n{_describe_listeners(listeners)}"
+            )
+        return self.on(listeners[0])
+
+    def _namespace(self, listener: Listener) -> str:
+        if not listener.namespaces:
+            raise click.UsageError(
+                f"Listener {listener.id} serves no namespaces. Check its configuration."
+            )
+        if self.k8s_namespace is None:
+            if len(listener.namespaces) == 1:
+                return listener.namespaces[0]
+            raise click.UsageError(
+                f"Listener {listener.id} serves several namespaces. Choose one with "
+                f"--k8s-namespace: {', '.join(listener.namespaces)}"
+            )
+        if self.k8s_namespace not in listener.namespaces:
+            raise click.UsageError(
+                f"Listener {listener.id} does not serve namespace "
+                f"'{self.k8s_namespace}'. Choose one of: "
+                f"{', '.join(listener.namespaces)}"
+            )
+        return self.k8s_namespace
+
+
+def _describe_listeners(listeners: Sequence[Listener]) -> str:
+    shown = listeners[:_LISTENERS_SHOWN]
+    lines = [
+        f"  {listener.id}  cluster {listener.compute_id}  "
+        f"namespaces: {', '.join(listener.namespaces)}"
+        for listener in shown
+    ]
+    if len(listeners) > len(shown):
+        lines.append(f"  ... and {len(listeners) - len(shown)} more")
+    if len(listeners) == MAX_PAGE_SIZE:
+        lines.append(f"  (only the first {MAX_PAGE_SIZE} listeners were read)")
+    return "\n".join(lines)
+
+
 @dataclass(frozen=True, slots=True)
 class ExistingDeployment:
     id: str
@@ -379,15 +509,16 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
     client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    listed = client.list_deployments(name_contains=name)
+    listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
-    resources = listed.get("resources", []) if isinstance(listed, dict) else []
+    for resource in listed:
-    for resource in resources:
+        if resource.get("name") == name and resource.get("id"):
-        if (
-            isinstance(resource, dict)
-            and resource.get("name") == name
-            and resource.get("id")
-        ):
             return ExistingDeployment(str(resource["id"]), _source_of(resource))
+    if len(listed) >= MAX_PAGE_SIZE:
+        raise click.ClickException(
+            "This workspace has more deployments than the CLI can search, so it "
+            f"cannot tell whether '{name}' already exists. Pass --deployment-id to "
+            "update an existing deployment."
+        )
     return None
 
 
@@ -683,14 +814,22 @@ def _find_deployment(
         existing = _call_host_backend_with_optional_tenant(
             client,
             lambda c: c.list_deployments(
-                agent_id=selector.agent_id, agent_environment=selector.environment
+                agent_id=selector.agent_id,
+                agent_environment=selector.environment,
+                limit=MAX_PAGE_SIZE,
             ),
         )
+        if len(existing) > 1:
+            raise click.ClickException(
+                "This control plane does not filter deployments by agent, so the "
+                f"CLI cannot tell which one belongs to '{selector.agent_id}' in "
+                f"{selector.environment}. Deploy by --name instead."
+            )
         found = next(
             (
                 ExistingDeployment(str(dep["id"]), _source_of(dep))
-                for dep in existing.get("resources", [])
+                for dep in existing
-                if not dep.get("is_preview")
+                if dep.get("id") and not dep.get("is_preview")
             ),
             None,
         )
@@ -758,21 +897,18 @@ def _create_deployment(
 
 
 def _get_deployment_status_url(
-    updated: object, deployment_id: str, host_url: str
+    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
 ) -> str | None:
-    """Compute the LangSmith dashboard URL for a deployment, if possible."""
     tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
     if not tenant_id:
         return None
-    base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
+    return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
-    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
 
 
 def _emit_deployment_status_url(
-    updated: object, deployment_id: str, host_url: str
+    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
 ) -> str | None:
-    """Emit the deployment status URL and return it."""
+    url = _get_deployment_status_url(updated, deployment_id, endpoints)
-    url = _get_deployment_status_url(updated, deployment_id, host_url)
     if url:
         _get_emitter().status_url(url)
     return url
@@ -790,14 +926,11 @@ def _poll_revision_status(
 ) -> tuple[str, str | None]:
     """Poll latest revision status until terminal status or timeout."""
     em = _get_emitter()
-    revisions_resp = client.list_revisions(deployment_id, limit=1)
+    revisions = client.list_revisions(deployment_id, limit=1)
-    resources = (
+    if not revisions:
-        revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else []
-    )
-    if not resources:
         return "", None
 
-    revision_id = str(resources[0]["id"])
+    revision_id = str(revisions[0]["id"])
     last_status = ""
     deadline = time.time() + timeout_seconds
     start_time = time.monotonic()
@@ -1318,6 +1451,7 @@ def _run_remote_build(
 @dataclass(frozen=True, slots=True)
 class DeployContext:
     client: HostBackendClient
+    endpoints: ControlPlaneEndpoints
     spec: BuildSpec
     verbose: bool
     selector: DeploymentSelector
@@ -1347,19 +1481,66 @@ def _resolve_or_create(
     )
     if found is not None:
         return found.id, step
-    created, step = _create_deployment(
+    try:
-        ctx.client,
+        created, step = _create_deployment(
-        step,
+            ctx.client,
-        name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
+            step,
-        agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
+            name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
-        source=source,
+            agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
-        source_config={"deployment_type": ctx.deployment_type},
+            source=source,
-        source_revision_config={},
+            source_config={"deployment_type": ctx.deployment_type},
-        secrets=ctx.secrets,
+            source_revision_config={},
-    )
+            secrets=ctx.secrets,
+        )
+    except HostBackendError as err:
+        if _needs_a_listener(err):
+            raise ListenerRequiredError(
+                "The image has to come from a registry you manage, so re-run with "
+                "--push-to <registry>/<repository>."
+            ) from None
+        raise
     return created.id, step
 
 
+class ListenerRequiredError(click.UsageError):
+    def __init__(self, remedy: str) -> None:
+        super().__init__(
+            "This workspace deploys through a listener in your own cluster. "
+            f"{remedy}\nLearn about listeners: {_LISTENERS_DOCS_URL}"
+        )
+
+
+def _needs_a_listener(err: HostBackendError) -> bool:
+    return err.status_code == 400 and _LISTENER_REQUIRED_MARKER in (
+        err.detail or err.message
+    )
+
+
+def _requested_listener(client: HostBackendClient, listener_id: str) -> Listener:
+    try:
+        resource = _call_host_backend_with_optional_tenant(
+            client, lambda c: c.get_listener(listener_id)
+        )
+    except HostBackendError as err:
+        if err.status_code not in _LISTENER_NOT_FOUND_STATUSES:
+            raise
+        available = _available_listeners(client)
+        if not available:
+            raise click.UsageError(_NO_LISTENERS) from None
+        raise click.UsageError(
+            f"Listener {listener_id} was not found in this workspace. "
+            f"Available listeners:\n{_describe_listeners(available)}"
+        ) from None
+    return Listener.from_resource(resource)
+
+
+def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]:
+    resources = _call_host_backend_with_optional_tenant(
+        client, lambda c: c.list_listeners()
+    )
+    return tuple(Listener.from_resource(resource) for resource in resources)
+
+
 def _ensure_customer_registry_source(existing: ExistingDeployment) -> None:
     if existing.source != _CUSTOMER_REGISTRY_SOURCE:
         raise click.UsageError(
@@ -1422,6 +1603,7 @@ class RemoteBuildSource:
 class CustomerRegistrySource:
     reference: ImageReference
     prebuilt_image: str | None
+    requested_placement: RequestedPlacement
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1443,6 +1625,7 @@ class CustomerRegistrySource:
         self, ctx: DeployContext, existing: ExistingDeployment, step: int
     ) -> DeployOutcome:
         _ensure_customer_registry_source(existing)
+        self.requested_placement.ensure_not_requested(existing.id)
         image_uri, step = self._publish(ctx, step)
         _log_deploy_step(step, f"Updating deployment {existing.id}")
         updated = ctx.client.update_deployment(
@@ -1456,7 +1639,25 @@ class CustomerRegistrySource:
             existing.id, _image_revision_result(updated, "Deployment updated")
         )
 
+    def _resolve_placement(self, ctx: DeployContext) -> Placement:
+        requested = self.requested_placement
+        if requested.listener_id is not None:
+            return requested.on(_requested_listener(ctx.client, requested.listener_id))
+        if not (ctx.endpoints.is_cloud or requested.requested):
+            return Unplaced()
+        return requested.among(_available_listeners(ctx.client))
+
+    def _announce(self, placement: Placement) -> None:
+        if isinstance(placement, OnListener):
+            _get_emitter().info(
+                placement.summary,
+                listener_id=placement.listener_id,
+                k8s_namespace=placement.k8s_namespace,
+            )
+
     def _create(self, ctx: DeployContext, name: str | None, step: int) -> DeployOutcome:
+        placement = self._resolve_placement(ctx)
+        self._announce(placement)
         image_uri, step = self._publish(ctx, step)
         try:
             created, _ = _create_deployment(
@@ -1467,13 +1668,19 @@ class CustomerRegistrySource:
                 if isinstance(ctx.selector, ByAgent)
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
-                source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC},
+                source_config={
+                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
+                    **placement.source_config(),
+                },
                 source_revision_config={"image_uri": image_uri},
                 secrets=ctx.secrets,
             )
         except HostBackendError as err:
-            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
+            if _needs_a_listener(err):
-                raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None
+                raise ListenerRequiredError(
+                    "Re-run with --listener-id and --k8s-namespace.\n"
+                    f"{err.detail or err.message}"
+                ) from None
             raise
         return DeployOutcome(
             created.id, _image_revision_result(created.resource, "Deployment created")
@@ -1534,14 +1741,31 @@ def _select_source(
     image_name: str | None,
     tag: str | None,
     remote_build_flag: bool | None,
+    placement: RequestedPlacement,
+    selector: DeploymentSelector,
 ) -> DeploymentSource:
+    if push_to is None and placement.requested:
+        raise click.UsageError(
+            "--listener-id and --k8s-namespace only apply when creating a "
+            "deployment with --push-to."
+        )
+    if placement.requested and isinstance(selector, ById):
+        raise click.UsageError(
+            "Listener and namespace are fixed when a deployment is created, so "
+            "they cannot be set for an existing --deployment-id. Drop them, or "
+            "create a new deployment with --name."
+        )
     if push_to is not None:
         if remote_build_flag is True:
             raise click.UsageError("--push-to cannot be combined with --remote.")
         reference = _push_reference(push_to, tag)
         if image is None:
             _require_local_docker()
-        return CustomerRegistrySource(reference, prebuilt_image=image)
+        return CustomerRegistrySource(
+            reference=reference,
+            prebuilt_image=image,
+            requested_placement=placement,
+        )
     if image and remote_build_flag is True:
         raise click.UsageError("--image cannot be combined with --remote builds.")
     use_remote_build, local_build_error = _resolve_build_mode(
@@ -1647,9 +1871,7 @@ def _call_host_backend_with_optional_tenant(
                 prompted_for_tenant = True
                 continue
             if err.status_code == 403 and "not enabled" in err.message.lower():
-                smith_base = ControlPlaneEndpoints.from_control_plane_url(
+                smith_base = client.endpoints.dashboard_url
-                    client.base_url
-                ).dashboard_url
                 raise HostBackendError(
                     "LangSmith Deployment is not enabled for this organization. "
                     f"Enable it at {smith_base}/host/deployments"
@@ -1690,11 +1912,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(
+OPT_AGENT_ENVIRONMENT = partial(
-    "--environment",
+    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 +2027,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=(
@@ -1848,6 +2079,21 @@ def _deploy_base_options(
                     "Give the tag here or with --tag (default: latest)."
                 ),
             ),
+            click.option(
+                "--listener-id",
+                help=(
+                    "Listener that will run the deployment, for workspaces that "
+                    "deploy through a listener in your own cluster. Only used when "
+                    "creating a deployment with --push-to."
+                ),
+            ),
+            click.option(
+                "--k8s-namespace",
+                help=(
+                    "Kubernetes namespace the listener deploys into. Only used when "
+                    "creating a deployment with --push-to."
+                ),
+            ),
             click.option(
                 "--config",
                 "-c",
@@ -1930,6 +2176,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
@@ -1958,6 +2210,8 @@ def _deploy_cmd(
     image_name: str | None,
     image: str | None,
     push_to: str | None,
+    listener_id: str | None,
+    k8s_namespace: str | None,
     tag: str | None,
     base_image: str | None,
     install_command: str | None,
@@ -1982,13 +2236,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():
@@ -2024,12 +2279,15 @@ def _deploy_cmd(
 
     secrets = _secrets_from_env(_env_without_deployment_name(env_vars))
 
+    selector = ByAgent(**agent) if agent else deployment_selector(deployment_id, name)
     source = _select_source(
         push_to=push_to,
         image=image,
         image_name=image_name,
         tag=tag,
         remote_build_flag=remote_build_flag,
+        placement=RequestedPlacement(listener_id, k8s_namespace),
+        selector=selector,
     )
 
     client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
@@ -2042,6 +2300,7 @@ def _deploy_cmd(
     outcome = source.run(
         DeployContext(
             client=client,
+            endpoints=client.endpoints,
             spec=BuildSpec(
                 config=config,
                 config_json=config_json,
@@ -2053,9 +2312,7 @@ def _deploy_cmd(
                 build_command=build_command,
             ),
             verbose=verbose,
-            selector=ByAgent(**agent)
+            selector=selector,
-            if agent
-            else deployment_selector(deployment_id, name),
             deployment_type=deployment_type,
             secrets=secrets,
             tracked_packages=tracked_packages,
@@ -2064,7 +2321,7 @@ def _deploy_cmd(
     dep_status_url = _emit_deployment_status_url(
         outcome.build_result.updated,
         outcome.deployment_id,
-        client.base_url,
+        client.endpoints,
     )
 
     if no_wait:
@@ -2124,7 +2381,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 +2395,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 = {}
@@ -2146,16 +2408,10 @@ def deploy_list(
     if environment is not None:
         filters["agent_environment"] = environment
     client = _create_host_backend_client(host_url, api_key)
-    response = _call_host_backend_with_optional_tenant(
+    deployments = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_deployments(name_contains=name_contains, **filters),
     )
-    resources = response.get("resources") if isinstance(response, dict) else None
-    deployments = (
-        [item for item in resources if isinstance(item, dict)]
-        if isinstance(resources, list)
-        else []
-    )
     if not deployments:
         click.echo("No deployments found.")
         return
@@ -2195,16 +2451,10 @@ def deploy_revisions_list(
     api_key: str | None, host_url: str | None, limit: int, deployment_id: str
 ) -> None:
     client = _create_host_backend_client(host_url, api_key)
-    response = _call_host_backend_with_optional_tenant(
+    revisions = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_revisions(deployment_id, limit=limit),
     )
-    resources = response.get("resources") if isinstance(response, dict) else None
-    revisions = (
-        [item for item in resources if isinstance(item, dict)]
-        if isinstance(resources, list)
-        else []
-    )
     if not revisions:
         click.echo(f"No revisions found for deployment {deployment_id}.")
         return
@@ -2354,17 +2604,12 @@ def deploy_logs(
         dep_id = found.id
 
     if log_type == "build" and not revision_id:
-        revisions_resp = client.list_revisions(dep_id, limit=1)
+        revisions = client.list_revisions(dep_id, limit=1)
-        resources = (
+        if not revisions:
-            revisions_resp.get("resources", [])
-            if isinstance(revisions_resp, dict)
-            else []
-        )
-        if not resources:
             raise click.ClickException(
                 "No revisions found for this deployment. Cannot fetch build logs."
             )
-        revision_id = str(resources[0]["id"])
+        revision_id = str(revisions[0]["id"])
         click.secho(f"Using latest revision: {revision_id}", fg="cyan")
 
     payload: dict = {"limit": limit, "order": "desc"}

libs/cli/langgraph_cli/host_backend.py

@@ -18,6 +18,7 @@ CLOUD_DASHBOARD_HOST = "smith.langchain.com"
 CONTROL_PLANE_PATH = "/api-host"
 LANGSMITH_API_PATHS = ("/api/v1", "/api")
 LOCAL_HOSTNAMES = ("localhost", "127.0.0.1")
+MAX_PAGE_SIZE = 100
 SourceName = Literal["internal_docker", "internal_source", "external_docker"]
 
 
@@ -36,6 +37,13 @@ class ControlPlaneEndpoints:
             return cls.from_langsmith_endpoint(langsmith_endpoint)
         return cls(CLOUD_CONTROL_PLANE_URL, CLOUD_DASHBOARD_URL)
 
+    @property
+    def is_cloud(self) -> bool:
+        hostname = urlparse(self.control_plane_url).hostname or ""
+        return hostname == CLOUD_CONTROL_PLANE_HOST or hostname.endswith(
+            f".{CLOUD_CONTROL_PLANE_HOST}"
+        )
+
     @classmethod
     def from_control_plane_url(cls, url: str) -> ControlPlaneEndpoints:
         control_plane_url = url.rstrip("/")
@@ -83,12 +91,36 @@ def _without_api_path(path: str) -> str:
     return path
 
 
+def _resources(payload: object) -> list[dict[str, Any]]:
+    if not isinstance(payload, dict):
+        return []
+    resources = payload.get("resources")
+    if not isinstance(resources, list):
+        return []
+    return [item for item in resources if isinstance(item, dict)]
+
+
 class HostBackendError(click.ClickException):
     """Raised when the host backend returns an error response."""
 
-    def __init__(self, message: str, status_code: int | None = None):
+    def __init__(
+        self,
+        message: str,
+        status_code: int | None = None,
+        detail: str | None = None,
+    ):
         super().__init__(message)
         self.status_code = status_code
+        self.detail = detail
+
+
+def _error_detail(response: httpx.Response) -> str | None:
+    try:
+        body = response.json()
+    except ValueError:
+        return None
+    detail = body.get("detail") if isinstance(body, dict) else None
+    return detail if isinstance(detail, str) else None
 
 
 class HostBackendClient:
@@ -110,7 +142,8 @@ class HostBackendClient:
         }
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
-        self._base_url = base_url.rstrip("/")
+        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
+        self._base_url = self._endpoints.control_plane_url
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
@@ -122,6 +155,10 @@ class HostBackendClient:
     def base_url(self) -> str:
         return self._base_url
 
+    @property
+    def endpoints(self) -> ControlPlaneEndpoints:
+        return self._endpoints
+
     def set_tenant(self, tenant_id: str) -> None:
         self._client.headers["X-Tenant-ID"] = tenant_id
 
@@ -136,10 +173,12 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = err.response.text or str(err.response.status_code)
+            detail = _error_detail(err.response)
+            reason = detail or err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                f"{method} {path} failed with status {err.response.status_code}: {reason}",
                 status_code=err.response.status_code,
+                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -178,20 +217,29 @@ class HostBackendClient:
 
     def list_deployments(
         self,
-        name_contains: str = "",
         *,
+        name: str | None = None,
+        name_contains: str | None = None,
+        limit: int | None = None,
         agent_id: str | None = None,
         agent_environment: str | None = None,
-    ) -> dict[str, Any]:
+    ) -> list[dict[str, Any]]:
-        params = {"name_contains": name_contains}
+        given = (
-        if agent_id is not None:
+            ("name", name),
-            params["agent_id"] = agent_id
+            ("name_contains", name_contains),
-        if agent_environment is not None:
+            ("limit", limit),
-            params["agent_environment"] = agent_environment
+            ("agent_id", agent_id),
-        return self._request(
+            ("agent_environment", agent_environment),
-            "GET",
+        )
-            "/v2/deployments",
+        params = {key: value for key, value in given if value is not None}
-            params=params,
+        return _resources(self._request("GET", "/v2/deployments", params=params))
+
+    def get_listener(self, listener_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/v2/listeners/{listener_id}")
+
+    def list_listeners(self) -> list[dict[str, Any]]:
+        return _resources(
+            self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE})
         )
 
     def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -266,10 +314,15 @@ class HostBackendClient:
             payload["secrets"] = secrets
         return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
-    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
+    def list_revisions(
-        return self._request(
+        self, deployment_id: str, limit: int = 1
-            "GET",
+    ) -> list[dict[str, Any]]:
-            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
+        return _resources(
+            self._request(
+                "GET",
+                f"/v2/deployments/{deployment_id}/revisions",
+                params={"limit": limit},
+            )
         )
 
     def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:

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

@@ -382,20 +382,18 @@ def test_deploy_list_command(monkeypatch) -> None:
 
         def list_deployments(self, name_contains: str = ""):
             captured["name_contains"] = name_contains
-            return {
+            return [
-                "resources": [
+                {
-                    {
+                    "id": "dep-123",
-                        "id": "dep-123",
+                    "name": "alpha",
-                        "name": "alpha",
+                    "source_config": {"custom_url": "https://alpha.example.com"},
-                        "source_config": {"custom_url": "https://alpha.example.com"},
+                },
-                    },
+                {
-                    {
+                    "id": "dep-456",
-                        "id": "dep-456",
+                    "name": "beta",
-                        "name": "beta",
+                    "source_config": {"custom_url": "https://beta.example.com"},
-                        "source_config": {"custom_url": "https://beta.example.com"},
+                },
-                    },
+            ]
-                ]
-            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -435,7 +433,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_deployments(self, name_contains: str = ""):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -468,20 +466,18 @@ def test_deploy_revisions_list_command(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return {
+            return [
-                "resources": [
+                {
-                    {
+                    "id": "rev-123",
-                        "id": "rev-123",
+                    "status": "CREATING",
-                        "status": "CREATING",
+                    "created_at": "2023-11-07T05:31:56Z",
-                        "created_at": "2023-11-07T05:31:56Z",
+                },
-                    },
+                {
-                    {
+                    "id": "rev-456",
-                        "id": "rev-456",
+                    "status": "DEPLOYED",
-                        "status": "DEPLOYED",
+                    "created_at": "2023-11-08T10:00:00Z",
-                        "created_at": "2023-11-08T10:00:00Z",
+                },
-                    },
+            ]
-                ]
-            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -522,7 +518,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_revisions(self, deployment_id: str, limit: int = 1):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -555,7 +551,7 @@ def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import uuid
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
@@ -17,6 +18,7 @@ from langgraph_cli.host_backend import HostBackendClient
 from langgraph_cli.image_reference import ImageReference
 
 CONTROL_PLANE_URL = "https://control-plane.example.com"
+CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com"
 REGISTRY_URL = "https://registry.example.com/team"
 PUSH_TOKEN = "push-token"
 PUSHED_IMAGE = "registry.example.com/team/my-app:latest"
@@ -24,10 +26,25 @@ PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123"
 PUSH_REPOSITORY = "registry.example.com/team/agent"
 EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest"
 EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123"
-LISTENER_REQUIRED = (
+LISTENER_ID = "11111111-1111-4111-8111-111111111111"
-    "Source configuration error: 'source_config.listener_id' is required for "
+OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222"
-    "workspace with available listener IDs: ['listener-1']"
+PAGE_TWO_LISTENER_ID = "33333333-3333-4333-8333-333333333333"
-)
+UNKNOWN_LISTENER_ID = "99999999-9999-4999-8999-999999999999"
+LISTENER = {
+    "id": LISTENER_ID,
+    "compute_id": "prod-cluster",
+    "compute_config": {"k8s_namespaces": ["agents"]},
+}
+OTHER_LISTENER = {
+    "id": OTHER_LISTENER_ID,
+    "compute_id": "other-cluster",
+    "compute_config": {"k8s_namespaces": ["agents"]},
+}
+TWO_NAMESPACE_LISTENER = {
+    "id": LISTENER_ID,
+    "compute_id": "prod-cluster",
+    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+}
 CREATED_ID = "dep-created"
 TRACKED_PACKAGES = ["langgraph:1.0.0"]
 SIGNED_UPLOAD_URL = "https://storage.example.com/signed"
@@ -38,7 +55,12 @@ DIGESTS_FORMAT = "{{json .RepoDigests}}"
 NOT_A_CLI_DEPLOYMENT = (
     "push token is only available for 'internal_docker' source deployments"
 )
+LISTENER_REQUIRED = (
+    "Source configuration error: 'source_config.listener_id' is required "
+    f"for workspace with available listener IDs: ['{LISTENER_ID}']"
+)
 LIST_DEPLOYMENTS = "GET /v2/deployments"
+LIST_LISTENERS = "GET /v2/listeners"
 CREATE_DEPLOYMENT = "POST /v2/deployments"
 
 
@@ -58,12 +80,22 @@ def _get(deployment_id: str) -> str:
     return f"GET /v2/deployments/{deployment_id}"
 
 
+def _looks_like_a_uuid(value: str) -> bool:
+    try:
+        uuid.UUID(value)
+    except ValueError:
+        return False
+    return True
+
+
 @dataclass
 class ControlPlaneDouble:
     timeline: list[str]
     existing_deployments: list[dict] = field(default_factory=list)
     push_token_status: int = 200
     create_error: str | None = None
+    listeners: list[dict] = field(default_factory=list)
+    listeners_by_id: dict[str, dict] = field(default_factory=dict)
     bodies: dict[str, dict] = field(default_factory=dict)
 
     def handle(self, request: httpx.Request) -> httpx.Response:
@@ -71,14 +103,45 @@ class ControlPlaneDouble:
         self.timeline.append(route)
         if request.content:
             self.bodies[route] = json.loads(request.content)
-        return self._respond(request.method, request.url.path)
+        return self._respond(request)
 
-    def _respond(self, method: str, path: str) -> httpx.Response:
+    def _respond(self, request: httpx.Request) -> httpx.Response:
+        method, path = request.method, request.url.path
+        if (method, path) == ("GET", "/v2/listeners"):
+            return httpx.Response(200, json={"resources": self.listeners})
+        if method == "GET" and path.startswith("/v2/listeners/"):
+            listener_id = path.rsplit("/", 1)[-1]
+            if not _looks_like_a_uuid(listener_id):
+                return httpx.Response(
+                    422,
+                    json={
+                        "detail": [
+                            {"type": "uuid_parsing", "loc": ["path", "listener_id"]}
+                        ]
+                    },
+                )
+            known = {listener["id"]: listener for listener in self.listeners}
+            known.update(self.listeners_by_id)
+            if listener_id not in known:
+                return httpx.Response(
+                    404, json={"detail": f"Listener ID {listener_id} not found."}
+                )
+            return httpx.Response(200, json=known[listener_id])
         if (method, path) == ("GET", "/v2/deployments"):
-            return httpx.Response(200, json={"resources": self.existing_deployments})
+            name = request.url.params.get("name")
+            return httpx.Response(
+                200,
+                json={
+                    "resources": [
+                        deployment
+                        for deployment in self.existing_deployments
+                        if name is None or deployment.get("name") == name
+                    ]
+                },
+            )
         if (method, path) == ("POST", "/v2/deployments"):
             if self.create_error is not None:
-                return httpx.Response(400, text=self.create_error)
+                return httpx.Response(400, json={"detail": self.create_error})
             return httpx.Response(201, json={"id": CREATED_ID, "tenant_id": "tenant-1"})
         if path.endswith("/push-token"):
             if self.push_token_status != 200:
@@ -199,7 +262,7 @@ class DeployProject:
     timeline: list[str]
     uploads: list[tuple[str, str, int]]
 
-    def run(self, *args: str) -> Result:
+    def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
         return CliRunner().invoke(
             cli,
             [
@@ -207,7 +270,7 @@ class DeployProject:
                 "--api-key",
                 "test-key",
                 "--host-url",
-                CONTROL_PLANE_URL,
+                host_url,
                 "--name",
                 "my-app",
                 "--no-input",
@@ -611,18 +674,6 @@ def test_push_to_rejects_a_non_external_deployment_before_any_docker_work(
     assert deploy_project.docker.verbs() == []
 
 
-def test_push_to_explains_the_listener_requirement_of_hybrid_workspaces(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.create_error = LISTENER_REQUIRED
-
-    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
-
-    assert result.exit_code != 0
-    assert "listener" in result.output
-    assert "--deployment-id" in result.output
-
-
 def test_push_to_with_deployment_id_fetches_the_deployment_once(
     deploy_project: DeployProject,
 ) -> None:
@@ -652,3 +703,414 @@ def test_invalid_tag_fails_before_any_control_plane_call(
     assert result.exit_code != 0
     assert "Image tag may only contain" in result.output
     assert deploy_project.timeline == []
+
+
+def test_push_to_places_a_new_deployment_on_the_only_listener(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        LIST_DEPLOYMENTS,
+        LIST_LISTENERS,
+        "docker build",
+        "docker push",
+        "docker inspect-digest",
+        CREATE_DEPLOYMENT,
+    ]
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {},
+        "listener_id": LISTENER_ID,
+        "listener_config": {"k8s_namespace": "agents"},
+    }
+    assert f"Deploying through listener {LISTENER_ID} in namespace agents" in (
+        result.output
+    )
+
+
+def test_push_to_places_a_new_deployment_on_the_chosen_listener(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        OTHER_LISTENER_ID,
+        "--k8s-namespace",
+        "agents",
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {},
+        "listener_id": OTHER_LISTENER_ID,
+        "listener_config": {"k8s_namespace": "agents"},
+    }
+
+
+@pytest.mark.parametrize(
+    ("listeners", "args", "message"),
+    [
+        pytest.param(
+            [LISTENER, OTHER_LISTENER], (), "--listener-id", id="two_listeners"
+        ),
+        pytest.param(
+            [TWO_NAMESPACE_LISTENER], (), "--k8s-namespace", id="two_namespaces"
+        ),
+        pytest.param(
+            [LISTENER],
+            ("--k8s-namespace", "nope"),
+            "does not serve namespace",
+            id="unknown_namespace",
+        ),
+    ],
+)
+def test_push_to_refuses_an_unresolved_placement_before_any_docker_work(
+    deploy_project: DeployProject, listeners, args, message
+) -> None:
+    deploy_project.control_plane.listeners = listeners
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, *args, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code != 0
+    assert message in result.output
+    assert deploy_project.docker.verbs() == []
+    assert CREATE_DEPLOYMENT not in deploy_project.timeline
+
+
+def test_self_hosted_control_plane_keeps_its_default_placement(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {}
+    }
+
+
+def test_self_hosted_control_plane_places_when_asked(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, "--listener-id", LISTENER_ID
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {},
+        "listener_id": LISTENER_ID,
+        "listener_config": {"k8s_namespace": "agents"},
+    }
+
+
+def test_updating_a_deployment_never_looks_up_listeners(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code == 0, result.output
+    assert LIST_LISTENERS not in deploy_project.timeline
+
+
+def test_listener_flags_are_refused_for_a_deployment_id_without_any_call(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--deployment-id",
+        "dep-ext",
+        "--k8s-namespace",
+        "agents",
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code != 0
+    assert "fixed when a deployment is created" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_listener_flags_are_refused_on_an_existing_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        LISTENER_ID,
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code != 0
+    assert "fixed when a deployment is created" in result.output
+    assert deploy_project.docker.verbs() == []
+
+
+def test_a_deployment_without_a_listener_announces_nothing(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert "listener" not in result.output
+
+
+def test_a_self_hosted_create_without_flags_never_looks_up_listeners(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert LIST_LISTENERS not in deploy_project.timeline
+
+
+def test_a_control_plane_that_demands_a_listener_names_the_flags(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.create_error = LISTENER_REQUIRED
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code != 0
+    assert "--listener-id" in result.output
+    assert "--k8s-namespace" in result.output
+    assert LISTENER_ID in result.output
+    assert "{" not in result.output
+    assert "POST /v2/deployments failed" not in result.output
+
+
+def test_listener_flags_without_push_to_make_no_call_at_all(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--listener-id", LISTENER_ID)
+
+    assert result.exit_code != 0
+    assert "--push-to" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_a_truncated_listener_page_says_so(deploy_project: DeployProject) -> None:
+    deploy_project.control_plane.listeners = [
+        {
+            "id": str(uuid.UUID(int=index)),
+            "compute_id": "cluster",
+            "compute_config": {"k8s_namespaces": ["agents"]},
+        }
+        for index in range(100)
+    ]
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code != 0
+    assert "first 100" in result.output
+
+
+def test_a_managed_build_in_a_listener_workspace_points_at_push_to(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.create_error = LISTENER_REQUIRED
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code != 0
+    assert "--push-to" in result.output
+    assert deploy_project.docker.verbs() == []
+
+
+@pytest.mark.parametrize(
+    "args",
+    [
+        pytest.param(("--no-remote",), id="managed_build"),
+        pytest.param(("--push-to", PUSH_REPOSITORY), id="push_to"),
+    ],
+)
+def test_a_listener_requirement_links_the_listener_docs(
+    deploy_project: DeployProject, args: tuple[str, ...]
+) -> None:
+    deploy_project.control_plane.create_error = LISTENER_REQUIRED
+
+    result = deploy_project.run(*args)
+
+    assert result.exit_code != 0
+    assert "https://docs.langchain.com/langsmith/control-plane#listeners" in (
+        result.output
+    )
+
+
+def test_a_managed_control_plane_without_listeners_creates_as_before(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {}
+    }
+    assert deploy_project.timeline.count(LIST_LISTENERS) == 1
+
+
+def test_a_listener_without_an_id_is_reported_rather_than_ignored(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [
+        {"compute_id": "broken", "compute_config": {"k8s_namespaces": ["agents"]}},
+        LISTENER,
+    ]
+
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
+    )
+
+    assert result.exit_code != 0
+    assert "without an id" in result.output
+    assert deploy_project.docker.verbs() == []
+
+
+def _listener_route(listener_id: str) -> str:
+    return f"GET /v2/listeners/{listener_id}"
+
+
+def test_an_explicit_listener_is_fetched_by_id_not_searched(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        OTHER_LISTENER_ID,
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code == 0, result.output
+    assert _listener_route(OTHER_LISTENER_ID) in deploy_project.timeline
+    assert LIST_LISTENERS not in deploy_project.timeline
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {},
+        "listener_id": OTHER_LISTENER_ID,
+        "listener_config": {"k8s_namespace": "agents"},
+    }
+
+
+def test_an_explicit_listener_beyond_the_first_page_still_works(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [
+        {
+            "id": str(uuid.UUID(int=index)),
+            "compute_id": "cluster",
+            "compute_config": {"k8s_namespaces": ["agents"]},
+        }
+        for index in range(100)
+    ]
+    deploy_project.control_plane.listeners_by_id = {
+        PAGE_TWO_LISTENER_ID: {
+            "id": PAGE_TWO_LISTENER_ID,
+            "compute_id": "far-cluster",
+            "compute_config": {"k8s_namespaces": ["agents"]},
+        }
+    }
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        PAGE_TWO_LISTENER_ID,
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {},
+        "listener_id": PAGE_TWO_LISTENER_ID,
+        "listener_config": {"k8s_namespace": "agents"},
+    }
+
+
+def test_an_unknown_listener_names_the_ones_that_exist(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        UNKNOWN_LISTENER_ID,
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code != 0
+    assert "was not found" in result.output
+    assert LISTENER_ID in result.output
+    assert "prod-cluster" in result.output
+    assert deploy_project.docker.verbs() == []
+
+
+def test_an_explicit_listener_in_a_workspace_without_any_is_refused(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        LISTENER_ID,
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code != 0
+    assert "no listeners" in result.output
+    assert deploy_project.docker.verbs() == []
+
+
+def test_a_listener_id_that_is_not_an_identifier_still_names_the_real_ones(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.listeners = [LISTENER]
+
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--listener-id",
+        "not-a-listener",
+        host_url=CLOUD_CONTROL_PLANE_URL,
+    )
+
+    assert result.exit_code != 0
+    assert "was not found" in result.output
+    assert LISTENER_ID in result.output
+    assert "uuid_parsing" not in result.output

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",
@@ -72,9 +72,9 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch):
     result = CliRunner().invoke(cli, AGENT_ARGS)
     assert result.exit_code == 0, result.output
     assert dict(requests[0].url.params) == {
-        "name_contains": "",
         "agent_id": "customer-support",
         "agent_environment": "staging",
+        "limit": "100",
     }
     payload = json.loads(requests[1].content)
     assert payload["agent"] == {
@@ -103,3 +103,17 @@ def test_agent_rejects_explicit_name(deployment_api, monkeypatch):
     assert result.exit_code == 2
     assert "cannot be combined" in result.output
     assert not requests
+
+
+def test_agent_lookup_refuses_a_control_plane_that_ignores_the_filter(deployment_api):
+    state, requests, _ = deployment_api
+    state["resources"] = [
+        {"id": "someone-elses", "is_preview": False},
+        {"id": "another", "is_preview": False},
+    ]
+
+    result = CliRunner().invoke(cli, AGENT_ARGS)
+
+    assert result.exit_code != 0
+    assert "does not filter deployments by agent" in result.output
+    assert len(requests) == 1

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -13,10 +13,17 @@ import pytest
 
 import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
+    ById,
+    ByName,
     CustomerRegistrySource,
     DockerBuildCommand,
+    ExistingDeployment,
+    Listener,
     ManagedRegistrySource,
+    OnListener,
     RemoteBuildSource,
+    RequestedPlacement,
+    Unplaced,
     _call_host_backend_with_optional_tenant,
     _create_host_backend_client,
     _docker_config_for_token,
@@ -27,6 +34,7 @@ from langgraph_cli.deploy import (
     _resolve_pushed_image_digest,
     _select_source,
     _validate_prebuilt_image,
+    find_deployment_by_name,
     normalize_image_tag,
     normalize_name,
 )
@@ -280,11 +288,13 @@ class TestCallHostBackendWithOptionalTenant:
         return c
 
     def test_success_passes_through(self):
-        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
+        client = self._make_client(
+            lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]})
+        )
         result = _call_host_backend_with_optional_tenant(
             client, lambda c: c.list_deployments()
         )
-        assert result == {"ok": True}
+        assert result == [{"id": "dep-1"}]
 
     def test_403_not_enabled_gives_actionable_error(self):
         detail = (
@@ -607,6 +617,8 @@ class TestSelectSource:
         "image_name": None,
         "tag": None,
         "remote_build_flag": None,
+        "placement": RequestedPlacement(),
+        "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
 
@@ -617,7 +629,9 @@ class TestSelectSource:
                 {"push_to": REPOSITORY},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -625,7 +639,9 @@ class TestSelectSource:
                 {"push_to": f"{REPOSITORY}:v2"},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "v2"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -633,7 +649,9 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "tag": "v3"},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "v3"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -641,10 +659,25 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "image": "app:dev"},
                 False,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image="app:dev",
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
             ),
+            pytest.param(
+                {
+                    "push_to": REPOSITORY,
+                    "placement": RequestedPlacement("listener-1", "agents"),
+                },
+                True,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                ),
+                id="push_to_carries_the_requested_placement",
+            ),
             pytest.param(
                 {"remote_build_flag": True},
                 True,
@@ -720,6 +753,16 @@ class TestSelectSource:
                 "--image cannot be combined with --remote builds.",
                 id="image_with_remote",
             ),
+            pytest.param(
+                {"placement": RequestedPlacement(listener_id="listener-1")},
+                "only apply when creating a deployment with --push-to",
+                id="listener_without_push_to",
+            ),
+            pytest.param(
+                {"placement": RequestedPlacement(k8s_namespace="agents")},
+                "only apply when creating a deployment with --push-to",
+                id="namespace_without_push_to",
+            ),
         ],
     )
     def test_conflicting_flags_are_rejected(self, monkeypatch, flags, message):
@@ -890,3 +933,289 @@ class TestResolvePushedImageDigest:
         frame_locals = captured["coro"].cr_frame.f_locals
         assert "--config" not in frame_locals["args"]
         captured["coro"].close()
+
+
+class TestListener:
+    @pytest.mark.parametrize(
+        ("resource", "expected"),
+        [
+            pytest.param(
+                {
+                    "id": "listener-1",
+                    "compute_id": "prod-cluster",
+                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+                },
+                Listener("listener-1", "prod-cluster", ("agents", "agents-staging")),
+                id="reads_id_cluster_and_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": {}},
+                Listener("listener-1", "c", ()),
+                id="missing_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": None},
+                Listener("listener-1", "c", ()),
+                id="null_compute_config",
+            ),
+            pytest.param(
+                {"id": "listener-1"},
+                Listener("listener-1", "", ()),
+                id="only_an_id",
+            ),
+        ],
+    )
+    def test_from_resource_reads_the_control_plane_shape(self, resource, expected):
+        assert Listener.from_resource(resource) == expected
+
+
+ONE_NAMESPACE = Listener("listener-1", "prod-cluster", ("agents",))
+TWO_NAMESPACES = Listener("listener-2", "multi-cluster", ("agents", "agents-staging"))
+NO_NAMESPACE = Listener("listener-3", "broken-cluster", ())
+
+
+class TestRequestedPlacement:
+    @pytest.mark.parametrize(
+        ("request_", "listeners", "expected"),
+        [
+            pytest.param(
+                RequestedPlacement(), (), Unplaced(), id="no_listeners_no_request"
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (ONE_NAMESPACE,),
+                OnListener("listener-1", "agents"),
+                id="uses_the_only_possible_answer",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents-staging"),
+                (TWO_NAMESPACES,),
+                OnListener("listener-2", "agents-staging"),
+                id="namespace_alone_picks_the_only_listener",
+            ),
+        ],
+    )
+    def test_resolves_to_a_placement(self, request_, listeners, expected):
+        assert request_.among(listeners) == expected
+
+    @pytest.mark.parametrize(
+        ("request_", "listeners", "message"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-1"),
+                (),
+                "no listeners",
+                id="workspace_has_no_listeners",
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (ONE_NAMESPACE, TWO_NAMESPACES),
+                "--listener-id",
+                id="several_listeners_need_a_choice",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents"),
+                (ONE_NAMESPACE, TWO_NAMESPACES),
+                "--listener-id",
+                id="namespace_alone_is_ambiguous_with_several_listeners",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents"),
+                (),
+                "no listeners",
+                id="namespace_without_any_listener",
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (TWO_NAMESPACES,),
+                "--k8s-namespace",
+                id="several_namespaces_need_a_choice",
+            ),
+        ],
+    )
+    def test_refuses_and_names_the_choices(self, request_, listeners, message):
+        with pytest.raises(click.UsageError, match=message):
+            request_.among(listeners)
+
+    def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self):
+        with pytest.raises(click.UsageError) as error:
+            RequestedPlacement().among((ONE_NAMESPACE, TWO_NAMESPACES))
+
+        assert "listener-1" in error.value.message
+        assert "prod-cluster" in error.value.message
+        assert "agents-staging" in error.value.message
+
+    @pytest.mark.parametrize(
+        ("placement", "expected"),
+        [
+            pytest.param(Unplaced(), {}, id="unplaced_adds_nothing"),
+            pytest.param(
+                OnListener("listener-1", "agents"),
+                {
+                    "listener_id": "listener-1",
+                    "listener_config": {"k8s_namespace": "agents"},
+                },
+                id="placed_carries_listener_and_namespace",
+            ),
+        ],
+    )
+    def test_source_config_matches_the_control_plane_shape(self, placement, expected):
+        assert placement.source_config() == expected
+
+
+def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_version():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["params"] = dict(req.url.params)
+        return httpx.Response(
+            200,
+            json={"resources": [{"id": "dep-1", "name": "agent", "source": "github"}]},
+        )
+
+    client = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    found = find_deployment_by_name(client, "agent")
+
+    assert seen["params"] == {
+        "name": "agent",
+        "name_contains": "agent",
+        "limit": "100",
+    }
+    assert found == ExistingDeployment("dep-1", "github")
+
+
+def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deployment():
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(
+                200,
+                json={
+                    "resources": [
+                        {
+                            "id": "dep-other",
+                            "name": "another-teams-agent",
+                            "source": "external_docker",
+                        }
+                    ]
+                },
+            )
+        ),
+    )
+
+    assert find_deployment_by_name(client, "brand-new-agent") is None
+
+
+def test_a_full_page_without_a_match_refuses_to_claim_the_name_is_free():
+    page = [
+        {"id": f"dep-{index}", "name": f"other-agent-{index}"} for index in range(100)
+    ]
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(200, json={"resources": page})
+        ),
+    )
+
+    with pytest.raises(click.ClickException, match="--deployment-id"):
+        find_deployment_by_name(client, "brand-new-agent")
+
+
+def test_a_partial_page_without_a_match_means_the_name_is_free():
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(
+                200, json={"resources": [{"id": "dep-1", "name": "other"}]}
+            )
+        ),
+    )
+
+    assert find_deployment_by_name(client, "brand-new-agent") is None
+
+
+@pytest.mark.parametrize(
+    "resource",
+    [
+        pytest.param({"compute_id": "c"}, id="no_id"),
+        pytest.param({"id": ""}, id="empty_id"),
+    ],
+)
+def test_a_listener_without_an_id_is_refused(resource):
+    with pytest.raises(HostBackendError, match="without an id"):
+        Listener.from_resource(resource)
+
+
+def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
+    monkeypatch,
+):
+    def explode() -> tuple[bool, str | None]:
+        raise AssertionError("docker must not be probed for an argv-only conflict")
+
+    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
+
+    with pytest.raises(click.UsageError, match="--deployment-id"):
+        _select_source(
+            push_to="registry.example.com/app",
+            image=None,
+            image_name=None,
+            tag=None,
+            remote_build_flag=None,
+            placement=RequestedPlacement(listener_id="listener-1"),
+            selector=ById("dep-1"),
+        )
+
+
+class TestPlacementOnAKnownListener:
+    @pytest.mark.parametrize(
+        ("request_", "listener", "expected"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-1"),
+                ONE_NAMESPACE,
+                OnListener("listener-1", "agents"),
+                id="the_only_namespace_is_used",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"),
+                TWO_NAMESPACES,
+                OnListener("listener-2", "agents"),
+                id="the_chosen_namespace_is_used",
+            ),
+        ],
+    )
+    def test_places_on_the_listener(self, request_, listener, expected):
+        assert request_.on(listener) == expected
+
+    @pytest.mark.parametrize(
+        ("request_", "listener", "message"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2"),
+                TWO_NAMESPACES,
+                "--k8s-namespace",
+                id="several_namespaces_need_a_choice",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2", k8s_namespace="nope"),
+                TWO_NAMESPACES,
+                "does not serve namespace",
+                id="unknown_namespace",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-3"),
+                NO_NAMESPACE,
+                "serves no namespaces",
+                id="listener_without_namespaces",
+            ),
+        ],
+    )
+    def test_refuses_and_names_the_namespaces(self, request_, listener, message):
+        with pytest.raises(click.UsageError, match=message):
+            request_.on(listener)

libs/cli/tests/unit_tests/test_host_backend.py

@@ -79,19 +79,6 @@ def test_request_transport_error_raises():
         c._request("GET", "/test")
 
 
-def test_list_deployments_sends_query_params():
-    def handler(req: httpx.Request) -> httpx.Response:
-        assert req.url.path == "/v2/deployments"
-        assert req.url.params["name_contains"] == "my app"
-        return httpx.Response(200, json={"ok": True})
-
-    c = HostBackendClient(
-        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
-    )
-    result = c.list_deployments("my app")
-    assert result == {"ok": True}
-
-
 def _capturing_client(captured: dict) -> HostBackendClient:
     def handler(req: httpx.Request) -> httpx.Response:
         captured["body"] = req.read()
@@ -421,7 +408,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url():
         transport=httpx.MockTransport(handler),
     )
 
-    assert c.list_revisions("dep-1", limit=2) == {"ok": True}
+    assert c.list_revisions("dep-1", limit=2) == []
     assert seen == {
         "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
         "api_key": "key",
@@ -546,3 +533,144 @@ def test_control_plane_endpoints_resolve(host_url, langsmith_endpoint, expected)
     endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
 
     assert (endpoints.control_plane_url, endpoints.dashboard_url) == expected
+
+
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        pytest.param(
+            {"resources": [{"id": "a"}, {"id": "b"}]},
+            [{"id": "a"}, {"id": "b"}],
+            id="list_returns_the_resources",
+        ),
+        pytest.param({"resources": []}, [], id="empty_list"),
+        pytest.param({}, [], id="missing_key"),
+        pytest.param({"resources": None}, [], id="null_resources"),
+        pytest.param(
+            {"resources": ["nope", {"id": "a"}]}, [{"id": "a"}], id="skips_non_objects"
+        ),
+        pytest.param([], [], id="unexpected_envelope"),
+    ],
+)
+def test_list_endpoints_return_resource_objects(payload, expected):
+    def handler(req: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=payload)
+
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    assert c.list_deployments() == expected
+
+
+def test_list_listeners_asks_for_a_full_page():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["url"] = str(req.url)
+        return httpx.Response(200, json={"resources": [{"id": "listener-1"}]})
+
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    assert c.list_listeners() == [{"id": "listener-1"}]
+    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"
+
+
+@pytest.mark.parametrize(
+    ("control_plane_url", "expected"),
+    [
+        pytest.param("https://api.host.langchain.com", True, id="cloud"),
+        pytest.param("https://eu.api.host.langchain.com", True, id="cloud_region"),
+        pytest.param("https://dev.api.host.langchain.com", True, id="cloud_dev"),
+        pytest.param("https://smith.example.com/api-host", False, id="self_hosted"),
+        pytest.param(
+            "https://corp.example.com/langsmith/api-host",
+            False,
+            id="self_hosted_prefix",
+        ),
+        pytest.param("http://localhost:8080/api-host", False, id="local"),
+        pytest.param(
+            "https://evil-api.host.langchain.com", False, id="lookalike_needs_a_dot"
+        ),
+    ],
+)
+def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expected):
+    endpoints = ControlPlaneEndpoints.from_control_plane_url(control_plane_url)
+
+    assert endpoints.is_cloud is expected
+
+
+@pytest.mark.parametrize(
+    ("call", "expected_params"),
+    [
+        pytest.param(
+            lambda c: c.list_deployments(name="agent"),
+            {"name": "agent"},
+            id="exact_name_filters_server_side",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(name_contains="age"),
+            {"name_contains": "age"},
+            id="substring_search_keeps_its_own_parameter",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(),
+            {},
+            id="no_filter_sends_no_parameters",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(
+                name="agent", name_contains="agent", limit=100
+            ),
+            {"name": "agent", "name_contains": "agent", "limit": "100"},
+            id="both_filters_travel_together_for_older_servers",
+        ),
+    ],
+)
+def test_list_deployments_sends_one_name_filter(call, expected_params):
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen.update(dict(req.url.params))
+        return httpx.Response(200, json={"resources": []})
+
+    call(
+        HostBackendClient(
+            "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+        )
+    )
+
+    assert seen == expected_params
+
+
+@pytest.mark.parametrize(
+    ("body", "expected"),
+    [
+        pytest.param(
+            {"detail": "Source configuration error: bad listener"},
+            "Source configuration error: bad listener",
+            id="fastapi_detail_is_unwrapped",
+        ),
+        pytest.param(
+            {"detail": {"loc": ["body"], "msg": "nope"}},
+            None,
+            id="a_structured_detail_is_left_alone",
+        ),
+        pytest.param({"other": "shape"}, None, id="an_unknown_shape_is_left_alone"),
+    ],
+)
+def test_error_detail_is_readable(body, expected):
+    c = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(lambda req: httpx.Response(400, json=body)),
+    )
+
+    with pytest.raises(HostBackendError) as error:
+        c.get_deployment("dep-1")
+
+    assert error.value.detail == expected
+    if expected is not None:
+        assert error.value.message.endswith(expected)

libs/langgraph/tests/test_delta_channel_parallel_order.py

@@ -1,83 +0,0 @@
-"""`DeltaChannel` replay must apply parallel writes in the order `invoke` did."""
-
-from typing import Annotated, Any
-
-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
-
-# Sorted, because live execution applies PULL tasks in node-name order.
-FAN_OUT_NAMES = ["a", "b", "c", "d", "e", "f", "g", "h"]
-
-
-def _append_reducer(current: list, updates: list) -> list:
-    return [*current, *(x for u in updates for x in u)]
-
-
-def _build_fan_out_graph(checkpointer: BaseCheckpointSaver) -> Any:
-    class State(TypedDict):
-        items: Annotated[
-            list, DeltaChannel(_append_reducer, list, snapshot_frequency=10_000)
-        ]
-
-    def make_node(label: str) -> Any:
-        return lambda state: {"items": [label]}
-
-    builder = StateGraph(State)
-    for name in FAN_OUT_NAMES:
-        builder.add_node(name, make_node(name))
-        builder.add_edge(START, name)
-        builder.add_edge(name, END)
-    return builder.compile(checkpointer=checkpointer)
-
-
-async def test_get_state_matches_live_invoke_order(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_fan_out_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    live = (await graph.ainvoke({"items": []}, config))["items"]
-    replayed = (await graph.aget_state(config)).values["items"]
-
-    assert live == FAN_OUT_NAMES
-    assert replayed == live
-
-
-async def test_continuing_thread_preserves_committed_prefix(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_fan_out_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    first = (await graph.ainvoke({"items": []}, config))["items"]
-    second = (await graph.ainvoke({"items": []}, config))["items"]
-
-    assert second == first + first
-    assert (await graph.aget_state(config)).values["items"] == second
-
-
-async def test_state_history_reports_live_order_at_every_step(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    runs = 3
-    graph = _build_fan_out_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-    for _ in range(runs):
-        await graph.ainvoke({"items": []}, config)
-    live = FAN_OUT_NAMES * runs
-
-    seen = [
-        s.values["items"]
-        async for s in graph.aget_state_history(config)
-        if "items" in s.values
-    ]
-
-    assert max(map(len, seen)) == len(live)
-    for values in seen:
-        assert values == live[: len(values)], f"{values} is not a prefix of {live}"