fix(checkpoint-sqlite): walk delta ancestors by parent pointer

Stage 1 of the sqlite delta history filtered `checkpoint_id <= target` and
streamed `ORDER BY checkpoint_id DESC`. Both encode an extra assumption: that
every child's checkpoint id sorts above its parent's.

Ancestry is defined by `parent_checkpoint_id`, and nothing in the contract
requires ids to be monotonic. A parent whose id sorted above its child's was
dropped from the stream, so its stored value and its writes were lost with no
error raised. Removing the range filter alone would not help: in DESC order
that parent arrives before the target, so the walk passes it before it has
started.

Replace it with a recursive CTE anchored at the target that follows
`parent_checkpoint_id`. Rows now arrive in walk order, so the off-path skip
and parent tracking in `step_walk_with_row` are dead and removed, and the
query reads only true ancestors instead of every row at or below the target.

Following pointers can loop where a bounded id scan could not, and a loop is
reachable through `put` alone: it writes with `INSERT OR REPLACE`, so
re-putting an existing checkpoint id under a descendant's config repoints
that checkpoint at its own descendant. The walk therefore stops on a repeated
checkpoint id. sqlite yields recursive rows lazily, so abandoning the cursor
ends the recursion.

Postgres needs no equivalent change: it pages the whole thread without an id
bound and follows parent pointers in Python, and its upsert never rewrites
`parent_checkpoint_id`, so it cannot form this loop.

Fixes #8550

Co-authored-by: lylelllll <59271327+lylelllll@users.noreply.github.com>

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

@@ -507,13 +507,12 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
         Two-stage query:
 
-        * Stage 1 (paged): newest-first slice of `checkpoints` returning
-          `(checkpoint_id, parent_checkpoint_id, type, checkpoint)` per
-          ancestor. Sqlite has no JSONB, so we ship the full serialized
-          checkpoint blob and inspect `channel_values` in Python. Pages
-          newest-first by `checkpoint_id` with a `< cursor` predicate;
-          page size is `DELTA_PAGE_SIZE`. Stops paging when every channel
-          has found its seed or the chain is exhausted.
+        * Stage 1 (streamed): recursive CTE over `checkpoints` following
+          `parent_checkpoint_id` from the target, returning
+          `(checkpoint_id, type, checkpoint)` per ancestor. Sqlite has no
+          JSONB, so we ship the full serialized checkpoint blob and inspect
+          `channel_values` in Python. Stops reading when every channel has
+          found its seed or the chain is exhausted.
 
         * Stage 2 (per-channel UNION ALL): one branch per channel reading
           `writes` filtered to that channel's specific `chain_cids`. No
@@ -538,12 +537,14 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         seeded: set[str] = set()
 
         with self.cursor(transaction=False) as cur:
-            cur.execute(DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id))
+            cur.execute(
+                DELTA_STAGE1_SQL,
+                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
+            )
             for row in cur:
-                cid, parent_cid, type_tag, blob = row
+                cid, type_tag, blob = row
                 if step_walk_with_row(
                     cid=cid,
-                    parent_cid=parent_cid,
                     type_tag=type_tag,
                     blob=blob,
                     target_id=checkpoint_id,

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

@@ -26,16 +26,29 @@ from typing import Any
 
 from langgraph.checkpoint.base import DeltaChannelHistory, PendingWrite
 
-# Stage 1 streams ancestors of `target_cid` newest-first. The `<=`
-# predicate keeps target itself in the stream so we can read its
-# `parent_checkpoint_id` from the first row without a separate lookup;
-# the caller skips target's own writes/seed (matches the
-# `BaseCheckpointSaver` contract).
+# Stage 1 streams target, then its ancestors nearest-first, by following
+# `parent_checkpoint_id`. Ids carry no ordering guarantee, so a range scan by
+# id can miss a parent whose id sorts above its child's. Target is the anchor
+# row; its own writes/seed are skipped (matches the `BaseCheckpointSaver`
+# contract).
+#
+# `put` is `INSERT OR REPLACE`, so re-putting an existing id under a
+# descendant's config makes the chain a loop. `step_walk_with_row` stops on a
+# repeated id; sqlite yields recursive rows lazily, so abandoning the cursor
+# ends the recursion.
 DELTA_STAGE1_SQL = (
+    "WITH RECURSIVE ancestors(checkpoint_id, parent_checkpoint_id, type, "
+    "checkpoint) AS ("
     "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
     "FROM checkpoints "
-    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id <= ? "
-    "ORDER BY checkpoint_id DESC"
+    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? "
+    "UNION ALL "
+    "SELECT c.checkpoint_id, c.parent_checkpoint_id, c.type, c.checkpoint "
+    "FROM checkpoints c JOIN ancestors a "
+    "ON c.checkpoint_id = a.parent_checkpoint_id "
+    "WHERE c.thread_id = ? AND c.checkpoint_ns = ?"
+    ") "
+    "SELECT checkpoint_id, type, checkpoint FROM ancestors"
 )
 
 
@@ -68,7 +81,6 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
 def step_walk_with_row(
     *,
     cid: str,
-    parent_cid: str | None,
     type_tag: str,
     blob: bytes,
     target_id: str,
@@ -81,36 +93,32 @@ def step_walk_with_row(
 ) -> bool:
     """Process one streamed stage-1 row in the merged ancestor walk.
 
-    The cursor returns (cid, parent_cid, type, blob) rows in
-    `checkpoint_id` DESC order starting at target. The first row is
-    target itself; we read its parent_cid to seed the walk and otherwise
-    skip it (target's own writes/seed are not part of the contract).
+    The cursor returns (cid, type, blob) rows in walk order starting at
+    target. The first row is target itself and is skipped (target's own
+    writes/seed are not part of the contract).
 
-    For each subsequent row, if `cid` matches the walk's current
-    position, we deserialize the blob, append the cid to every
-    not-yet-seeded channel's chain, and check `channel_values` for
+    For each subsequent row we deserialize the blob, append the cid to
+    every not-yet-seeded channel's chain, and check `channel_values` for
     seeds. The deserialized checkpoint is dropped before advancing — no
     cross-row cache, so peak in-flight is one deserialized checkpoint.
 
-    Off-path rows (different branch on the same thread) advance the
-    cursor without doing any work.
-
-    Returns True when every requested channel is seeded — the caller
-    can stop iterating and close the cursor.
+    Returns True when the caller can stop iterating and close the cursor:
+    every requested channel is seeded, or the chain revisited a checkpoint.
     """
     if "started" not in walk_state:
         if cid == target_id:
             walk_state["started"] = True
-            walk_state["cur_cid"] = parent_cid
             walk_state["active"] = {ch for ch in channels if ch not in seeded}
+            walk_state["walked"] = {cid}
         # Not target yet (or target not present): keep streaming.
         return False
     active: set[str] = walk_state["active"]
     if not active:
         return True
-    if cid != walk_state["cur_cid"]:
-        # Off-path row from a sibling branch — skip without deserializing.
-        return False
+    walked: set[str] = walk_state["walked"]
+    if cid in walked:
+        return True
+    walked.add(cid)
     for ch in active:
         chain_by_ch[ch].append(cid)
     ckpt = serde.loads_typed((type_tag, blob))
@@ -120,7 +128,6 @@ def step_walk_with_row(
         seeded.add(ch)
         active.discard(ch)
     del ckpt, channel_values
-    walk_state["cur_cid"] = parent_cid
     return not active
 
 

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

@@ -625,8 +625,8 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         """Fast-path override of `BaseCheckpointSaver.aget_delta_channel_history`.
 
         See `SqliteSaver.get_delta_channel_history` for design notes; this
-        is the async equivalent using `aiosqlite` cursors. Stage 1 pages
-        the parent chain newest-first and Python-deserializes each
+        is the async equivalent using `aiosqlite` cursors. Stage 1 streams
+        the parent chain from the target and Python-deserializes each
         checkpoint blob to find per-channel snapshots; stage 2 fetches
         only the relevant writes via per-channel UNION ALL.
         """
@@ -650,13 +650,13 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
         async with self.lock, self.conn.cursor() as cur:
             await cur.execute(
-                DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id)
+                DELTA_STAGE1_SQL,
+                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
             )
             async for row in cur:
-                cid, parent_cid, type_tag, blob = row
+                cid, type_tag, blob = row
                 if step_walk_with_row(
                     cid=cid,
-                    parent_cid=parent_cid,
                     type_tag=type_tag,
                     blob=blob,
                     target_id=checkpoint_id,

libs/checkpoint-sqlite/tests/test_delta_parent_walk.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from langgraph.checkpoint.base import (
+    BaseCheckpointSaver,
+    Checkpoint,
+    DeltaChannelHistory,
+    empty_checkpoint,
+)
+
+from langgraph.checkpoint.sqlite import SqliteSaver
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+CHANNEL = "ch"
+CONFIG: dict[str, Any] = {"configurable": {"thread_id": "t", "checkpoint_ns": ""}}
+EXPECTED: DeltaChannelHistory = {
+    "writes": [("task", CHANNEL, "write-root")],
+    "seed": "seed",
+}
+
+
+def _checkpoint(checkpoint_id: str, values: dict[str, Any]) -> Checkpoint:
+    value = empty_checkpoint()
+    value["id"] = checkpoint_id
+    value["channel_values"] = values
+    return value
+
+
+PARENT_ID_ORDERS = [
+    pytest.param("z-older", "a-newer", id="parent_id_sorts_above_child"),
+    pytest.param("a-older", "z-newer", id="parent_id_sorts_below_child"),
+]
+
+
+@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
+def test_sync_walk_reaches_parent_whatever_the_id_order(
+    root_id: str, child_id: str
+) -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        root = saver.put(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
+        saver.put_writes(root, [(CHANNEL, "write-root")], "task")
+        child = saver.put(root, _checkpoint(child_id, {}), {}, {})
+
+        got = saver.get_delta_channel_history(config=child, channels=[CHANNEL])
+        reference = BaseCheckpointSaver.get_delta_channel_history(
+            saver, config=child, channels=[CHANNEL]
+        )
+        assert got[CHANNEL] == EXPECTED
+        assert got[CHANNEL] == reference[CHANNEL], "fast path disagrees with base"
+
+
+@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
+async def test_async_walk_reaches_parent_whatever_the_id_order(
+    root_id: str, child_id: str
+) -> None:
+    async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+        root = await saver.aput(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
+        await saver.aput_writes(root, [(CHANNEL, "write-root")], "task")
+        child = await saver.aput(root, _checkpoint(child_id, {}), {}, {})
+
+        got = await saver.aget_delta_channel_history(config=child, channels=[CHANNEL])
+        assert got[CHANNEL] == EXPECTED
+
+
+def test_walk_reaches_root_of_long_chain_with_descending_ids() -> None:
+    steps = 40
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        parent = saver.put(
+            CONFIG, _checkpoint(f"id-{steps:03d}", {CHANNEL: "seed"}), {}, {}
+        )
+        saver.put_writes(parent, [(CHANNEL, "write-root")], "task")
+        for step in range(steps - 1, 0, -1):
+            parent = saver.put(parent, _checkpoint(f"id-{step:03d}", {}), {}, {})
+
+        got = saver.get_delta_channel_history(config=parent, channels=[CHANNEL])
+        assert got[CHANNEL] == EXPECTED
+
+
+def test_walk_terminates_when_put_makes_the_parent_chain_cycle() -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        a = saver.put(CONFIG, _checkpoint("cid-a", {}), {}, {})
+        b = saver.put(a, _checkpoint("cid-b", {}), {}, {})
+        repoint_a_under_b = _checkpoint("cid-a", {})
+        saver.put(b, repoint_a_under_b, {}, {})
+
+        got = saver.get_delta_channel_history(config=b, channels=[CHANNEL])
+        assert got[CHANNEL] == {"writes": []}

libs/cli/langgraph_cli/__init__.py

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

libs/cli/langgraph_cli/deploy.py

@@ -26,7 +26,6 @@ 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,
@@ -102,16 +101,16 @@ _NATIVE_AMD64_MACHINE = "x86_64"
 _PUSH_ATTEMPTS = 3
 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
 _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
-_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."
+_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"
+
 
 _TERMINAL_STATUSES = frozenset(
     [
@@ -162,134 +161,6 @@ 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
@@ -508,16 +379,15 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
     client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
-    for resource in listed:
-        if resource.get("name") == name and resource.get("id"):
+    listed = client.list_deployments(name_contains=name)
+    resources = listed.get("resources", []) if isinstance(listed, dict) else []
+    for resource in resources:
+        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
 
 
@@ -805,46 +675,6 @@ def _find_deployment(
     selector: ByName | ByAgent,
     *,
     not_found_message: str,
-<<<<<<< HEAD
-    agent: dict[str, str] | None = None,
-) -> tuple[str | None, bool, int]:
-    """Resolve an existing deployment by ID or exact name match."""
-    needs_creation = False
-    if deployment_id:
-        _log_deploy_step(step, f"Using deployment {deployment_id}")
-        _call_host_backend_with_optional_tenant(
-            client, lambda c: c.get_deployment(deployment_id)
-        )
-        return deployment_id, needs_creation, step + 1
-
-    if agent is not None:
-        _log_deploy_step(
-            step, f"Looking up agent '{agent['agent_id']}' in {agent['environment']}"
-        )
-        existing = _call_host_backend_with_optional_tenant(
-            client,
-            lambda c: c.list_deployments(
-                agent_id=agent["agent_id"], agent_environment=agent["environment"]
-            ),
-        )
-        found_id = next(
-            (
-                dep["id"]
-                for dep in existing.get("resources", [])
-                if not dep.get("is_preview")
-            ),
-            None,
-        )
-    else:
-        _log_deploy_step(step, f"Looking up deployment '{name}'")
-        found_id = _call_host_backend_with_optional_tenant(
-            client, lambda c: find_deployment_id_by_name(c, name)
-        )
-    em = _get_emitter()
-    if found_id:
-        deployment_id = str(found_id)
-        em.info(f"Found existing deployment (ID: {deployment_id})")
-=======
 ) -> tuple[ExistingDeployment | None, int]:
     if isinstance(selector, ByAgent):
         _log_deploy_step(
@@ -853,26 +683,17 @@ 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,
-                limit=MAX_PAGE_SIZE,
+                agent_id=selector.agent_id, agent_environment=selector.environment
             ),
         )
-        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
-                if dep.get("id") and not dep.get("is_preview")
+                for dep in existing.get("resources", [])
+                if not dep.get("is_preview")
             ),
             None,
         )
->>>>>>> origin
     else:
         _log_deploy_step(step, f"Looking up deployment '{selector.name}'")
         found = _call_host_backend_with_optional_tenant(
@@ -897,22 +718,12 @@ def _create_deployment(
     step: int,
     *,
     name: str | None,
-<<<<<<< HEAD
-    deployment_type: str,
-    source: str,
-    config_rel: str | None = None,
-    secrets: list[dict[str, str]] | None = None,
-    agent: dict[str, str] | None = None,
-) -> tuple[str, int]:
-    """Create a deployment and return its ID and next step number."""
-=======
     source: str,
     source_config: dict[str, object],
     source_revision_config: dict[str, object],
     secrets: list[dict[str, str]],
     agent: dict[str, str] | None = None,
 ) -> tuple[CreatedDeployment, int]:
->>>>>>> origin
     _log_deploy_step(
         step,
         f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}"
@@ -922,15 +733,9 @@ def _create_deployment(
     try:
         created = client.create_deployment(
             name=name,
-<<<<<<< HEAD
-            deployment_type=deployment_type,
-            source=source,
-            config_path=config_rel,
-=======
             source=source,
             source_config=source_config,
             source_revision_config=source_revision_config,
->>>>>>> origin
             secrets=secrets,
             agent=agent,
         )
@@ -947,51 +752,27 @@ def _create_deployment(
             "POST /v2/deployments succeeded but response missing a valid 'id'"
         )
     if agent is not None:
-<<<<<<< HEAD
-        _get_emitter().info(f"Deployment name: {created['name']}")
-    _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
-    return created_id, step + 1
-
-
-def _smith_dashboard_base_url(host_url: str | None) -> str:
-    """Derive the LangSmith dashboard base URL from the API host URL."""
-    from urllib.parse import urlparse
-
-    if not host_url:
-        return "https://smith.langchain.com"
-    parsed = urlparse(host_url)
-    hostname = parsed.hostname or ""
-    if hostname in ("localhost", "127.0.0.1"):
-        return host_url.rstrip("/")
-
-    for api_host_suffix in ("api.host.langchain.com", "api.smith.langchain.com"):
-        if hostname == api_host_suffix:
-            return "https://smith.langchain.com"
-        if hostname.endswith(f".{api_host_suffix}"):
-            prefix = hostname[: -(len(api_host_suffix) + 1)]
-            return f"https://{prefix}.smith.langchain.com"
-
-    return "https://smith.langchain.com"
-=======
         _get_emitter().info(f"Deployment name: {created.get('name')}")
     _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
     return CreatedDeployment(created_id, created), step + 1
->>>>>>> origin
 
 
 def _get_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> 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
-    return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
+    base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
+    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
 
 
 def _emit_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> str | None:
-    url = _get_deployment_status_url(updated, deployment_id, endpoints)
+    """Emit the deployment status URL and return it."""
+    url = _get_deployment_status_url(updated, deployment_id, host_url)
     if url:
         _get_emitter().status_url(url)
     return url
@@ -1009,11 +790,14 @@ def _poll_revision_status(
 ) -> tuple[str, str | None]:
     """Poll latest revision status until terminal status or timeout."""
     em = _get_emitter()
-    revisions = client.list_revisions(deployment_id, limit=1)
-    if not revisions:
+    revisions_resp = client.list_revisions(deployment_id, limit=1)
+    resources = (
+        revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else []
+    )
+    if not resources:
         return "", None
 
-    revision_id = str(revisions[0]["id"])
+    revision_id = str(resources[0]["id"])
     last_status = ""
     deadline = time.time() + timeout_seconds
     start_time = time.monotonic()
@@ -1534,7 +1318,6 @@ def _run_remote_build(
 @dataclass(frozen=True, slots=True)
 class DeployContext:
     client: HostBackendClient
-    endpoints: ControlPlaneEndpoints
     spec: BuildSpec
     verbose: bool
     selector: DeploymentSelector
@@ -1564,66 +1347,19 @@ def _resolve_or_create(
     )
     if found is not None:
         return found.id, step
-    try:
-        created, step = _create_deployment(
-            ctx.client,
-            step,
-            name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
-            agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
-            source=source,
-            source_config={"deployment_type": ctx.deployment_type},
-            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
+    created, step = _create_deployment(
+        ctx.client,
+        step,
+        name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
+        agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
+        source=source,
+        source_config={"deployment_type": ctx.deployment_type},
+        source_revision_config={},
+        secrets=ctx.secrets,
+    )
     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(
@@ -1686,7 +1422,6 @@ class RemoteBuildSource:
 class CustomerRegistrySource:
     reference: ImageReference
     prebuilt_image: str | None
-    requested_placement: RequestedPlacement
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1708,7 +1443,6 @@ 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(
@@ -1722,25 +1456,7 @@ 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(
@@ -1751,19 +1467,13 @@ class CustomerRegistrySource:
                 if isinstance(ctx.selector, ByAgent)
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
-                source_config={
-                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
-                    **placement.source_config(),
-                },
+                source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC},
                 source_revision_config={"image_uri": image_uri},
                 secrets=ctx.secrets,
             )
         except HostBackendError as err:
-            if _needs_a_listener(err):
-                raise ListenerRequiredError(
-                    "Re-run with --listener-id and --k8s-namespace.\n"
-                    f"{err.detail or err.message}"
-                ) from None
+            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
+                raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None
             raise
         return DeployOutcome(
             created.id, _image_revision_result(created.resource, "Deployment created")
@@ -1824,31 +1534,14 @@ 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=reference,
-            prebuilt_image=image,
-            requested_placement=placement,
-        )
+        return CustomerRegistrySource(reference, prebuilt_image=image)
     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(
@@ -1954,7 +1647,9 @@ 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 = client.endpoints.dashboard_url
+                smith_base = ControlPlaneEndpoints.from_control_plane_url(
+                    client.base_url
+                ).dashboard_url
                 raise HostBackendError(
                     "LangSmith Deployment is not enabled for this organization. "
                     f"Enable it at {smith_base}/host/deployments"
@@ -1995,17 +1690,11 @@ OPT_HOST_URL = click.option(
 )
 
 OPT_AGENT_ID = click.option(
-    "--agent-id",
-    envvar="LANGSMITH_AGENT_ID",
-    show_envvar=True,
-    help="Logical agent ID (requires agent mode enabled for the tenant).",
+    "--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)."
 )
 
 OPT_AGENT_ENVIRONMENT = click.option(
-    "--agent-environment",
-    "environment",
-    envvar="LANGSMITH_AGENT_ENVIRONMENT",
-    show_envvar=True,
+    "--environment",
     type=click.Choice(["development", "staging", "production"]),
     help="Agent environment (requires agent mode enabled for the tenant).",
 )
@@ -2159,21 +1848,6 @@ 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",
@@ -2284,8 +1958,6 @@ 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,
@@ -2310,7 +1982,6 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
-<<<<<<< HEAD
         if not agent_id or not agent_id.strip() or not environment:
             raise click.UsageError(
                 "--agent-id and --environment are required together."
@@ -2318,16 +1989,6 @@ def _deploy_cmd(
         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."
-=======
-        em.note("Note: --agent-id and --agent-environment flags are in private beta")
-        if not agent_id or not agent_id.strip() or not environment:
-            raise click.UsageError(
-                "--agent-id and --agent-environment are required together."
-            )
-        if name is not None or deployment_id is not None:
-            raise click.UsageError(
-                "--agent-id and --agent-environment cannot be combined with --name or --deployment-id."
->>>>>>> origin
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2363,53 +2024,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)
-<<<<<<< HEAD
-    step = 1
-
-    deployment_id, needs_creation, step = _resolve_deployment(
-        client,
-        step,
-        deployment_id,
-        name,
-        not_found_message=(
-            "No deployment found. Will create."
-            if use_remote_build
-            else "No deployment found. Will create after build."
-        ),
-        agent=agent,
-    )
-
-    if needs_creation:
-        deployment_id, step = _create_deployment(
-            client,
-            step,
-            name=name,
-            deployment_type=deployment_type,
-            source="internal_source" if use_remote_build else "internal_docker",
-            secrets=secrets,
-            agent=agent,
-        )
-
-    if not deployment_id:
-        raise click.ClickException("Failed to determine deployment ID")
-
-    # Scan local sources for tracked packages so the new revision carries
-    # the same metadata GitHub-backed deploys produce. Failures must never
-    # block a deploy.
-=======
->>>>>>> origin
     try:
         tracked_packages = find_tracked_packages(config, config_json) or None
     except Exception as exc:
@@ -2419,7 +2042,6 @@ def _deploy_cmd(
     outcome = source.run(
         DeployContext(
             client=client,
-            endpoints=client.endpoints,
             spec=BuildSpec(
                 config=config,
                 config_json=config_json,
@@ -2431,7 +2053,9 @@ def _deploy_cmd(
                 build_command=build_command,
             ),
             verbose=verbose,
-            selector=selector,
+            selector=ByAgent(**agent)
+            if agent
+            else deployment_selector(deployment_id, name),
             deployment_type=deployment_type,
             secrets=secrets,
             tracked_packages=tracked_packages,
@@ -2440,7 +2064,7 @@ def _deploy_cmd(
     dep_status_url = _emit_deployment_status_url(
         outcome.build_result.updated,
         outcome.deployment_id,
-        client.endpoints,
+        client.base_url,
     )
 
     if no_wait:
@@ -2514,14 +2138,6 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
-<<<<<<< HEAD
-=======
-    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",
-        )
->>>>>>> origin
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}
@@ -2530,18 +2146,15 @@ def deploy_list(
     if environment is not None:
         filters["agent_environment"] = environment
     client = _create_host_backend_client(host_url, api_key)
-    deployments = _call_host_backend_with_optional_tenant(
+    response = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_deployments(name_contains=name_contains, **filters),
-<<<<<<< HEAD
     )
     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 []
-=======
->>>>>>> origin
     )
     if not deployments:
         click.echo("No deployments found.")
@@ -2582,10 +2195,16 @@ 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)
-    revisions = _call_host_backend_with_optional_tenant(
+    response = _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
@@ -2735,12 +2354,17 @@ def deploy_logs(
         dep_id = found.id
 
     if log_type == "build" and not revision_id:
-        revisions = client.list_revisions(dep_id, limit=1)
-        if not revisions:
+        revisions_resp = client.list_revisions(dep_id, limit=1)
+        resources = (
+            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(revisions[0]["id"])
+        revision_id = str(resources[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,7 +18,6 @@ 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"]
 
 
@@ -37,13 +36,6 @@ 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("/")
@@ -91,36 +83,12 @@ 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,
-        detail: str | None = None,
-    ):
+    def __init__(self, message: str, status_code: int | 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:
@@ -142,8 +110,7 @@ class HostBackendClient:
         }
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
-        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
-        self._base_url = self._endpoints.control_plane_url
+        self._base_url = base_url.rstrip("/")
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
@@ -155,10 +122,6 @@ 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
 
@@ -173,12 +136,10 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = _error_detail(err.response)
-            reason = detail or err.response.text or str(err.response.status_code)
+            detail = err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {reason}",
+                f"{method} {path} failed with status {err.response.status_code}: {detail}",
                 status_code=err.response.status_code,
-                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -217,29 +178,20 @@ 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,
-    ) -> list[dict[str, Any]]:
-        given = (
-            ("name", name),
-            ("name_contains", name_contains),
-            ("limit", limit),
-            ("agent_id", agent_id),
-            ("agent_environment", agent_environment),
-        )
-        params = {key: value for key, value in given if value is not None}
-        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})
+    ) -> dict[str, Any]:
+        params = {"name_contains": name_contains}
+        if agent_id is not None:
+            params["agent_id"] = agent_id
+        if agent_environment is not None:
+            params["agent_environment"] = agent_environment
+        return self._request(
+            "GET",
+            "/v2/deployments",
+            params=params,
         )
 
     def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -314,15 +266,10 @@ 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
-    ) -> list[dict[str, Any]]:
-        return _resources(
-            self._request(
-                "GET",
-                f"/v2/deployments/{deployment_id}/revisions",
-                params={"limit": limit},
-            )
+    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions?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,18 +382,20 @@ def test_deploy_list_command(monkeypatch) -> None:
 
         def list_deployments(self, name_contains: str = ""):
             captured["name_contains"] = name_contains
-            return [
-                {
-                    "id": "dep-123",
-                    "name": "alpha",
-                    "source_config": {"custom_url": "https://alpha.example.com"},
-                },
-                {
-                    "id": "dep-456",
-                    "name": "beta",
-                    "source_config": {"custom_url": "https://beta.example.com"},
-                },
-            ]
+            return {
+                "resources": [
+                    {
+                        "id": "dep-123",
+                        "name": "alpha",
+                        "source_config": {"custom_url": "https://alpha.example.com"},
+                    },
+                    {
+                        "id": "dep-456",
+                        "name": "beta",
+                        "source_config": {"custom_url": "https://beta.example.com"},
+                    },
+                ]
+            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -433,7 +435,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_deployments(self, name_contains: str = ""):
-            return []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -466,18 +468,20 @@ 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 [
-                {
-                    "id": "rev-123",
-                    "status": "CREATING",
-                    "created_at": "2023-11-07T05:31:56Z",
-                },
-                {
-                    "id": "rev-456",
-                    "status": "DEPLOYED",
-                    "created_at": "2023-11-08T10:00:00Z",
-                },
-            ]
+            return {
+                "resources": [
+                    {
+                        "id": "rev-123",
+                        "status": "CREATING",
+                        "created_at": "2023-11-07T05:31:56Z",
+                    },
+                    {
+                        "id": "rev-456",
+                        "status": "DEPLOYED",
+                        "created_at": "2023-11-08T10:00:00Z",
+                    },
+                ]
+            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -518,7 +522,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_revisions(self, deployment_id: str, limit: int = 1):
-            return []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -551,7 +555,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 []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 

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

@@ -1,6 +1,5 @@
 import asyncio
 import json
-import uuid
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
@@ -18,7 +17,6 @@ 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"
@@ -26,25 +24,10 @@ 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_ID = "11111111-1111-4111-8111-111111111111"
-OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222"
-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"]},
-}
+LISTENER_REQUIRED = (
+    "Source configuration error: 'source_config.listener_id' is required for "
+    "workspace with available listener IDs: ['listener-1']"
+)
 CREATED_ID = "dep-created"
 TRACKED_PACKAGES = ["langgraph:1.0.0"]
 SIGNED_UPLOAD_URL = "https://storage.example.com/signed"
@@ -55,12 +38,7 @@ 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"
 
 
@@ -80,22 +58,12 @@ 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:
@@ -103,45 +71,14 @@ class ControlPlaneDouble:
         self.timeline.append(route)
         if request.content:
             self.bodies[route] = json.loads(request.content)
-        return self._respond(request)
+        return self._respond(request.method, request.url.path)
 
-    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])
+    def _respond(self, method: str, path: str) -> httpx.Response:
         if (method, path) == ("GET", "/v2/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
-                    ]
-                },
-            )
+            return httpx.Response(200, json={"resources": self.existing_deployments})
         if (method, path) == ("POST", "/v2/deployments"):
             if self.create_error is not None:
-                return httpx.Response(400, json={"detail": self.create_error})
+                return httpx.Response(400, text=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:
@@ -262,7 +199,7 @@ class DeployProject:
     timeline: list[str]
     uploads: list[tuple[str, str, int]]
 
-    def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
+    def run(self, *args: str) -> Result:
         return CliRunner().invoke(
             cli,
             [
@@ -270,7 +207,7 @@ class DeployProject:
                 "--api-key",
                 "test-key",
                 "--host-url",
-                host_url,
+                CONTROL_PLANE_URL,
                 "--name",
                 "my-app",
                 "--no-input",
@@ -674,6 +611,18 @@ 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:
@@ -703,414 +652,3 @@ 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",
-    "--agent-environment",
+    "--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,17 +103,3 @@ 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,17 +13,10 @@ 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,
@@ -34,7 +27,6 @@ from langgraph_cli.deploy import (
     _resolve_pushed_image_digest,
     _select_source,
     _validate_prebuilt_image,
-    find_deployment_by_name,
     normalize_image_tag,
     normalize_name,
 )
@@ -288,13 +280,11 @@ class TestCallHostBackendWithOptionalTenant:
         return c
 
     def test_success_passes_through(self):
-        client = self._make_client(
-            lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]})
-        )
+        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
         result = _call_host_backend_with_optional_tenant(
             client, lambda c: c.list_deployments()
         )
-        assert result == [{"id": "dep-1"}]
+        assert result == {"ok": True}
 
     def test_403_not_enabled_gives_actionable_error(self):
         detail = (
@@ -617,8 +607,6 @@ class TestSelectSource:
         "image_name": None,
         "tag": None,
         "remote_build_flag": None,
-        "placement": RequestedPlacement(),
-        "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
 
@@ -629,9 +617,7 @@ class TestSelectSource:
                 {"push_to": REPOSITORY},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -639,9 +625,7 @@ class TestSelectSource:
                 {"push_to": f"{REPOSITORY}:v2"},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v2"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -649,9 +633,7 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "tag": "v3"},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v3"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -659,25 +641,10 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "image": "app:dev"},
                 False,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
-                    prebuilt_image="app:dev",
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
                 ),
                 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,
@@ -753,16 +720,6 @@ 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):
@@ -933,289 +890,3 @@ 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,6 +79,19 @@ 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()
@@ -408,7 +421,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url():
         transport=httpx.MockTransport(handler),
     )
 
-    assert c.list_revisions("dep-1", limit=2) == []
+    assert c.list_revisions("dep-1", limit=2) == {"ok": True}
     assert seen == {
         "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
         "api_key": "key",
@@ -533,144 +546,3 @@ 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)