chore(cli): start 0.4.32 dev cycle (#9066)

CLI version update to 0.4.32.dev0

CLAUDE.md

@@ -1,65 +0,0 @@
-# AGENTS Instructions
-
-This repository is a monorepo. Each library lives in a subdirectory under `libs/`.
-
-<corridor>
-
-## Corridor security analysis
-
-Before generating or modifying code, create a plan and use Corridor's `analyzePlan` tool to analyze it. Apply the resulting security guidance before writing code.
-
-</corridor>
-
-When you modify code in any library, run the following commands in that library's directory before creating a pull request:
-
-- `make format` – run code formatters
-- `make lint` – run the linter
-- `make test` – execute the test suite
-
-To run a particular test file or to pass additional pytest options you can specify the `TEST` variable:
-
-```
-TEST=path/to/test.py make test
-```
-
-Other pytest arguments can also be supplied inside the `TEST` variable.
-
-## Libraries
-
-The repository contains several Python and JavaScript/TypeScript libraries.
-Below is a high-level overview:
-
-- **checkpoint** – base interfaces for LangGraph checkpointers.
-- **checkpoint-postgres** – Postgres implementation of the checkpoint saver.
-- **checkpoint-sqlite** – SQLite implementation of the checkpoint saver.
-- **cli** – official command-line interface for LangGraph.
-- **langgraph** – core framework for building stateful, multi-actor agents.
-- **prebuilt** – high-level APIs for creating and running agents and tools.
-- **sdk-js** – JS/TS SDK for interacting with the LangGraph REST API.
-- **sdk-py** – Python SDK for the LangGraph Server API.
-
-### Dependency map
-
-The diagram below lists downstream libraries for each production dependency as
-declared in that library's `pyproject.toml` (or `package.json`).
-
-```text
-checkpoint
-├── checkpoint-postgres
-├── checkpoint-sqlite
-├── prebuilt
-└── langgraph
-
-prebuilt
-└── langgraph
-
-sdk-py
-├── langgraph
-└── cli
-
-sdk-js (standalone)
-```
-
-Changes to a library may impact all of its dependents shown above.
-
-- Do NOT use Sphinx-style double backtick formatting (` ``code`` `). Use single backticks (`` `code` ``) for inline code references in docstrings and comments.

libs/cli/langgraph_cli/__init__.py

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

libs/cli/langgraph_cli/exec.py

@@ -1,12 +1,18 @@
 import asyncio
 import signal
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
 from contextlib import contextmanager
-from typing import cast
+from typing import Any, Protocol, TypeVar, cast
 
 import click.exceptions
 
+_T = TypeVar("_T")
+
+
+class CommandRunner(Protocol):
+    def run(self, coro: Coroutine[Any, Any, _T]) -> _T: ...
+
 
 @contextmanager
 def Runner():

libs/cli/langgraph_cli/host_backend.py

@@ -2,18 +2,125 @@
 
 from __future__ import annotations
 
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Literal
+from urllib.parse import urlparse
 
 import click
 import httpx
 
+CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com"
+CLOUD_DASHBOARD_URL = "https://smith.langchain.com"
+CLOUD_DOMAIN = "langchain.com"
+CLOUD_API_HOST = "api.smith.langchain.com"
+CLOUD_CONTROL_PLANE_HOST = "api.host.langchain.com"
+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"]
+
+
+@dataclass(frozen=True, slots=True)
+class ControlPlaneEndpoints:
+    control_plane_url: str
+    dashboard_url: str
+
+    @classmethod
+    def resolve(
+        cls, host_url: str | None, langsmith_endpoint: str | None
+    ) -> ControlPlaneEndpoints:
+        if host_url:
+            return cls.from_control_plane_url(host_url)
+        if langsmith_endpoint:
+            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("/")
+        hostname = urlparse(control_plane_url).hostname or ""
+        if control_plane_url.endswith(CONTROL_PLANE_PATH):
+            return cls(control_plane_url, control_plane_url[: -len(CONTROL_PLANE_PATH)])
+        if hostname in LOCAL_HOSTNAMES:
+            return cls(control_plane_url, control_plane_url)
+        return cls(control_plane_url, _cloud_dashboard_for(hostname))
+
+    @classmethod
+    def from_langsmith_endpoint(cls, endpoint: str) -> ControlPlaneEndpoints:
+        parsed = urlparse(endpoint.rstrip("/"))
+        hostname = parsed.hostname or ""
+        if _is_cloud_host(hostname):
+            return cls.from_control_plane_url(
+                f"https://{_cloud_control_plane_host_for(hostname)}"
+            )
+        root = f"{parsed.scheme}://{parsed.netloc}{_without_api_path(parsed.path)}"
+        return cls(f"{root}{CONTROL_PLANE_PATH}", root)
+
+
+def _is_cloud_host(hostname: str) -> bool:
+    return hostname == CLOUD_DOMAIN or hostname.endswith(f".{CLOUD_DOMAIN}")
+
+
+def _cloud_control_plane_host_for(langsmith_api_host: str) -> str:
+    if langsmith_api_host.endswith(f".{CLOUD_API_HOST}"):
+        region = langsmith_api_host[: -len(CLOUD_API_HOST)]
+        return f"{region}{CLOUD_CONTROL_PLANE_HOST}"
+    return CLOUD_CONTROL_PLANE_HOST
+
+
+def _cloud_dashboard_for(control_plane_host: str) -> str:
+    if control_plane_host.endswith(f".{CLOUD_CONTROL_PLANE_HOST}"):
+        region = control_plane_host[: -len(CLOUD_CONTROL_PLANE_HOST) - 1]
+        return f"https://{region}.{CLOUD_DASHBOARD_HOST}"
+    return CLOUD_DASHBOARD_URL
+
+
+def _without_api_path(path: str) -> str:
+    for api_path in LANGSMITH_API_PATHS:
+        if path.endswith(api_path):
+            return path[: -len(api_path)]
+    return path
+
+
+def _resources(payload: object) -> list[dict[str, Any]]:
+    if not isinstance(payload, dict):
+        return []
+    resources = payload.get("resources")
+    if not isinstance(resources, list):
+        return []
+    return [item for item in resources if isinstance(item, dict)]
+
 
 class HostBackendError(click.ClickException):
     """Raised when the host backend returns an error response."""
 
-    def __init__(self, message: str, status_code: int | None = None):
+    def __init__(
+        self,
+        message: str,
+        status_code: int | None = None,
+        detail: str | None = None,
+    ):
         super().__init__(message)
         self.status_code = status_code
+        self.detail = detail
+
+
+def _error_detail(response: httpx.Response) -> str | None:
+    try:
+        body = response.json()
+    except ValueError:
+        return None
+    detail = body.get("detail") if isinstance(body, dict) else None
+    return detail if isinstance(detail, str) else None
 
 
 class HostBackendClient:
@@ -24,24 +131,37 @@ class HostBackendClient:
         base_url: str,
         api_key: str,
         tenant_id: str | None = None,
+        *,
+        transport: httpx.BaseTransport | None = None,
     ):
         if not base_url:
             raise click.UsageError("Host backend URL is required")
-        transport = httpx.HTTPTransport(retries=3)
         headers: dict[str, str] = {
             "X-Api-Key": api_key,
             "Accept": "application/json",
         }
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
-        self._base_url = base_url.rstrip("/")
+        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
+        self._base_url = self._endpoints.control_plane_url
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
-            transport=transport,
+            transport=transport or httpx.HTTPTransport(retries=3),
             timeout=30,
         )
 
+    @property
+    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
+
     def _request(
         self,
         method: str,
@@ -53,10 +173,12 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = err.response.text or str(err.response.status_code)
+            detail = _error_detail(err.response)
+            reason = detail or err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                f"{method} {path} failed with status {err.response.status_code}: {reason}",
                 status_code=err.response.status_code,
+                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -72,30 +194,52 @@ class HostBackendClient:
 
     def create_deployment(
         self,
-        name: str,
-        deployment_type: str,
-        source: str,
-        config_path: str | None = None,
+        *,
+        name: str | None,
+        source: SourceName,
+        source_config: dict[str, object],
+        source_revision_config: dict[str, object],
         secrets: list[dict[str, str]] | None = None,
+        agent: dict[str, str] | None = None,
     ) -> dict[str, Any]:
-        """Create a deployment."""
         payload: dict[str, Any] = {
-            "name": name,
             "source": source,
-            "source_config": {"deployment_type": deployment_type},
-            "source_revision_config": {},
+            "source_config": source_config,
+            "source_revision_config": source_revision_config,
         }
-        if source == "internal_source" and config_path:
-            payload["source_revision_config"]["langgraph_config_path"] = config_path
+        if agent is not None:
+            payload["agent"] = agent
+        else:
+            payload["name"] = name
         if secrets is not None:
             payload["secrets"] = secrets
         return self._request("POST", "/v2/deployments", payload)
 
-    def list_deployments(self, name_contains: str = "") -> dict[str, Any]:
-        return self._request(
-            "GET",
-            "/v2/deployments",
-            params={"name_contains": name_contains},
+    def list_deployments(
+        self,
+        *,
+        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})
         )
 
     def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -121,22 +265,21 @@ class HostBackendClient:
         self,
         deployment_id: str,
         image_uri: str,
+        *,
+        revision_source: SourceName | None,
         secrets: list[dict[str, str]] | None = None,
         tracked_packages: list[str] | None = None,
     ) -> dict[str, Any]:
         payload: dict[str, Any] = {
-            "revision_source": "internal_docker",
             "source_revision_config": {"image_uri": image_uri},
         }
+        if revision_source is not None:
+            payload["revision_source"] = revision_source
         if tracked_packages:
             payload["tracked_packages"] = tracked_packages
         if secrets is not None:
             payload["secrets"] = secrets
-        return self._request(
-            "PATCH",
-            f"/v2/deployments/{deployment_id}",
-            payload,
-        )
+        return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
     def update_deployment_internal_source(
         self,
@@ -171,10 +314,15 @@ class HostBackendClient:
             payload["secrets"] = secrets
         return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
-    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
-        return self._request(
-            "GET",
-            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
+    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 get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:

libs/cli/langgraph_cli/image_reference.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+
+DIGEST_SEPARATOR = "@sha256:"
+DIGEST_MARKER = "@"
+TAG_SEPARATOR = ":"
+PATH_SEPARATOR = "/"
+
+
+@dataclass(frozen=True, slots=True)
+class ImageReference:
+    repository: str
+    tag: str | None = None
+
+    @classmethod
+    def parse(cls, reference: str) -> ImageReference:
+        if DIGEST_MARKER in reference:
+            raise ValueError(f"{reference!r} carries a digest and cannot be tagged")
+        path_start = reference.rfind(PATH_SEPARATOR) + 1
+        name, separator, tag = reference[path_start:].partition(TAG_SEPARATOR)
+        if not separator:
+            return cls(reference)
+        return cls(reference[:path_start] + name, tag)
+
+    def with_tag(self, tag: str) -> ImageReference:
+        return replace(self, tag=tag)
+
+    def matches_digest(self, repo_digest: str) -> bool:
+        return repo_digest.startswith(f"{self.repository}{DIGEST_SEPARATOR}")
+
+    def __str__(self) -> str:
+        if self.tag is None:
+            return self.repository
+        return f"{self.repository}{TAG_SEPARATOR}{self.tag}"

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

@@ -382,20 +382,18 @@ def test_deploy_list_command(monkeypatch) -> None:
 
         def list_deployments(self, name_contains: str = ""):
             captured["name_contains"] = name_contains
-            return {
-                "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"},
-                    },
-                ]
-            }
+            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"},
+                },
+            ]
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -435,7 +433,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_deployments(self, name_contains: str = ""):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -468,20 +466,18 @@ def test_deploy_revisions_list_command(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return {
-                "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",
-                    },
-                ]
-            }
+            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",
+                },
+            ]
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -522,7 +518,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_revisions(self, deployment_id: str, limit: int = 1):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -555,7 +551,7 @@ def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 

libs/cli/tests/unit_tests/test_deploy_agents.py

@@ -0,0 +1,119 @@
+import json
+from unittest.mock import Mock
+
+import httpx
+import pytest
+from click.testing import CliRunner
+
+import langgraph_cli.deploy as deploy
+from langgraph_cli.cli import cli
+from langgraph_cli.host_backend import HostBackendClient
+
+
+@pytest.fixture
+def deployment_api(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("LANGSMITH_DEPLOYMENT_NAME", raising=False)
+    monkeypatch.setattr(deploy, "_emitter", None)
+    monkeypatch.setattr(deploy, "_no_input", False)
+    (tmp_path / "langgraph.json").write_text(
+        json.dumps({"dependencies": ["."], "graphs": {"agent": "./agent.py:graph"}})
+    )
+    (tmp_path / ".env").write_text("LANGSMITH_DEPLOYMENT_NAME=legacy\n")
+    requests = []
+    state = {"enabled": True, "resources": []}
+
+    def handler(request):
+        requests.append(request)
+        assert request.url.path == "/v2/deployments"
+        if request.method == "GET":
+            if not state["enabled"] and (
+                "agent_id" in request.url.params
+                or "agent_environment" in request.url.params
+            ):
+                return httpx.Response(
+                    400, text="Agent filters are not available for this tenant."
+                )
+            return httpx.Response(200, json={"resources": state["resources"]})
+        assert request.method == "POST"
+        return httpx.Response(200, json={"id": "runtime-id", "name": "server-name"})
+
+    client = HostBackendClient("https://api.example.com", "test-key")
+    client._client.close()
+    client._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key"},
+    )
+    monkeypatch.setattr(deploy, "_create_host_backend_client", lambda *a, **kw: client)
+    monkeypatch.setattr(deploy, "find_tracked_packages", lambda *a: [])
+    remote_build = Mock(return_value=deploy.BuildResult())
+    monkeypatch.setattr(deploy, "_run_remote_build", remote_build)
+    monkeypatch.setattr(deploy, "_resolve_build_mode", lambda flag, **kw: (flag, None))
+    yield state, requests, remote_build
+    client._client.close()
+
+
+AGENT_ARGS = [
+    "deploy",
+    "--agent-id",
+    "customer-support",
+    "--agent-environment",
+    "staging",
+    "--remote",
+    "--no-wait",
+    "--no-input",
+]
+
+
+def test_agent_create(deployment_api, tmp_path, monkeypatch):
+    monkeypatch.setenv("LANGSMITH_DEPLOYMENT_NAME", "legacy")
+    _, requests, build = deployment_api
+    result = CliRunner().invoke(cli, AGENT_ARGS)
+    assert result.exit_code == 0, result.output
+    assert dict(requests[0].url.params) == {
+        "agent_id": "customer-support",
+        "agent_environment": "staging",
+        "limit": "100",
+    }
+    payload = json.loads(requests[1].content)
+    assert payload["agent"] == {
+        "agent_id": "customer-support",
+        "environment": "staging",
+    }
+    assert "name" not in payload
+    assert build.call_args.kwargs["deployment_id"] == "runtime-id"
+    assert "server-name" in result.output
+    assert (tmp_path / ".env").read_text() == "LANGSMITH_DEPLOYMENT_NAME=legacy\n"
+
+
+def test_agent_update(deployment_api):
+    state, requests, build = deployment_api
+    state["resources"] = [{"id": "existing-id", "is_preview": False}]
+    result = CliRunner().invoke(cli, AGENT_ARGS)
+    assert result.exit_code == 0, result.output
+    assert len(requests) == 1
+    assert build.call_args.kwargs["deployment_id"] == "existing-id"
+
+
+def test_agent_rejects_explicit_name(deployment_api, monkeypatch):
+    monkeypatch.setenv("LANGSMITH_DEPLOYMENT_NAME", "legacy")
+    _, requests, _ = deployment_api
+    result = CliRunner().invoke(cli, [*AGENT_ARGS, "--name", "legacy"])
+    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,6 +13,17 @@ import pytest
 
 import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
+    ById,
+    ByName,
+    CustomerRegistrySource,
+    DockerBuildCommand,
+    ExistingDeployment,
+    Listener,
+    ManagedRegistrySource,
+    OnListener,
+    RemoteBuildSource,
+    RequestedPlacement,
+    Unplaced,
     _call_host_backend_with_optional_tenant,
     _create_host_backend_client,
     _docker_config_for_token,
@@ -21,12 +32,14 @@ from langgraph_cli.deploy import (
     _parse_env_from_config,
     _resolve_env_path,
     _resolve_pushed_image_digest,
-    _smith_dashboard_base_url,
+    _select_source,
     _validate_prebuilt_image,
+    find_deployment_by_name,
     normalize_image_tag,
     normalize_name,
 )
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError
+from langgraph_cli.image_reference import ImageReference
 
 
 class TestDockerConfigForToken:
@@ -259,31 +272,29 @@ class TestEnvWithoutDeploymentName:
 
 class TestCallHostBackendWithOptionalTenant:
     def _make_client(self, handler):
-        c = HostBackendClient("https://api.example.com", "test-key")
-        c._client = httpx.Client(
-            base_url="https://api.example.com",
+        c = HostBackendClient(
+            "https://api.example.com",
+            "test-key",
             transport=httpx.MockTransport(handler),
-            headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-            timeout=30,
         )
         return c
 
     def _make_eu_client(self, handler):
-        c = HostBackendClient("https://eu.api.host.langchain.com", "test-key")
-        c._client = httpx.Client(
-            base_url="https://eu.api.host.langchain.com",
+        c = HostBackendClient(
+            "https://eu.api.host.langchain.com",
+            "test-key",
             transport=httpx.MockTransport(handler),
-            headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-            timeout=30,
         )
         return c
 
     def test_success_passes_through(self):
-        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
+        client = self._make_client(
+            lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]})
+        )
         result = _call_host_backend_with_optional_tenant(
             client, lambda c: c.list_deployments()
         )
-        assert result == {"ok": True}
+        assert result == [{"id": "dep-1"}]
 
     def test_403_not_enabled_gives_actionable_error(self):
         detail = (
@@ -334,7 +345,6 @@ class TestCallHostBackendWithOptionalTenant:
         assert exc_info.value.status_code == 403
         assert "smith.langchain.com" in exc_info.value.message
         assert seen_tenant_ids == [None, "workspace-123"]
-        assert client._client.headers["X-Tenant-ID"] == "workspace-123"
 
     def test_other_403_re_raises_original(self):
         client = self._make_client(
@@ -540,60 +550,226 @@ class TestCreateHostBackendClientNoInput:
         assert client is not None
 
 
-class TestSmithDashboardBaseUrl:
-    def test_none_returns_default(self):
-        assert _smith_dashboard_base_url(None) == "https://smith.langchain.com"
+class TestCreateHostBackendClientEndpoint:
+    def test_langsmith_endpoint_from_project_env_selects_self_hosted_control_plane(
+        self, monkeypatch
+    ):
+        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
+        monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
 
-    def test_empty_returns_default(self):
-        assert _smith_dashboard_base_url("") == "https://smith.langchain.com"
-
-    def test_prod_host_url(self):
-        assert (
-            _smith_dashboard_base_url("https://api.host.langchain.com")
-            == "https://smith.langchain.com"
+        client = _create_host_backend_client(
+            host_url=None,
+            api_key=None,
+            env_vars={"LANGSMITH_ENDPOINT": "https://smith.example.com/api/v1"},
         )
 
-    def test_dev_host_url(self):
-        assert (
-            _smith_dashboard_base_url("https://dev.api.host.langchain.com")
-            == "https://dev.smith.langchain.com"
+        assert client.base_url == "https://smith.example.com/api-host"
+
+    def test_explicit_host_url_wins_over_langsmith_endpoint(self, monkeypatch):
+        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
+        monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1")
+
+        client = _create_host_backend_client(
+            host_url="https://custom.host.com", api_key=None, env_vars={}
         )
 
-    def test_eu_host_url(self):
-        assert (
-            _smith_dashboard_base_url("https://eu.api.host.langchain.com")
-            == "https://eu.smith.langchain.com"
+        assert client.base_url == "https://custom.host.com"
+
+
+class TestDockerBuildCommand:
+    @pytest.mark.parametrize(
+        ("machine", "verbose", "expected"),
+        [
+            pytest.param(
+                "x86_64",
+                False,
+                DockerBuildCommand(("docker", "build"), ()),
+                id="amd64_host_builds_natively",
+            ),
+            pytest.param(
+                "arm64",
+                False,
+                DockerBuildCommand(
+                    ("docker", "buildx", "build"),
+                    ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+                ),
+                id="other_hosts_cross_build_quietly",
+            ),
+            pytest.param(
+                "arm64",
+                True,
+                DockerBuildCommand(
+                    ("docker", "buildx", "build"),
+                    ("--platform", "linux/amd64", "--load"),
+                ),
+                id="verbose_cross_build_keeps_progress_output",
+            ),
+        ],
+    )
+    def test_for_host_targets_the_deployment_platform(self, machine, verbose, expected):
+        assert DockerBuildCommand.for_host(machine, verbose=verbose) == expected
+
+
+class TestSelectSource:
+    OPTIONS = {
+        "push_to": None,
+        "image": None,
+        "image_name": None,
+        "tag": None,
+        "remote_build_flag": None,
+        "placement": RequestedPlacement(),
+        "selector": ByName("my-app"),
+    }
+    REPOSITORY = "registry.example.com/app"
+
+    @pytest.mark.parametrize(
+        ("flags", "docker_available", "expected"),
+        [
+            pytest.param(
+                {"push_to": REPOSITORY},
+                True,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
+                ),
+                id="push_to_selects_the_external_source_with_the_default_tag",
+            ),
+            pytest.param(
+                {"push_to": f"{REPOSITORY}:v2"},
+                True,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "v2"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
+                ),
+                id="push_to_keeps_a_tag_given_in_the_reference",
+            ),
+            pytest.param(
+                {"push_to": REPOSITORY, "tag": "v3"},
+                True,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "v3"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
+                ),
+                id="tag_flag_composes_with_push_to",
+            ),
+            pytest.param(
+                {"push_to": REPOSITORY, "image": "app:dev"},
+                False,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image="app:dev",
+                    requested_placement=RequestedPlacement(),
+                ),
+                id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
+            ),
+            pytest.param(
+                {
+                    "push_to": REPOSITORY,
+                    "placement": RequestedPlacement("listener-1", "agents"),
+                },
+                True,
+                CustomerRegistrySource(
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                ),
+                id="push_to_carries_the_requested_placement",
+            ),
+            pytest.param(
+                {"remote_build_flag": True},
+                True,
+                RemoteBuildSource(),
+                id="remote_flag_selects_the_source_upload",
+            ),
+            pytest.param(
+                {},
+                False,
+                RemoteBuildSource(),
+                id="no_local_docker_falls_back_to_the_source_upload",
+            ),
+            pytest.param(
+                {},
+                True,
+                ManagedRegistrySource(
+                    prebuilt_image=None, image_name=None, tag="latest"
+                ),
+                id="local_docker_selects_the_internal_docker_source",
+            ),
+            pytest.param(
+                {"image": "app:dev", "tag": "v1"},
+                False,
+                ManagedRegistrySource(
+                    prebuilt_image="app:dev", image_name=None, tag="v1"
+                ),
+                id="prebuilt_image_forces_the_internal_docker_source",
+            ),
+        ],
+    )
+    def test_flags_select_one_source(
+        self, monkeypatch, mocker, flags, docker_available, expected
+    ):
+        mocker.patch(
+            "langgraph_cli.deploy._get_emitter", return_value=mocker.MagicMock()
+        )
+        monkeypatch.setattr(
+            deploy_mod,
+            "can_build_locally",
+            lambda: (True, None) if docker_available else (False, "Docker is required"),
         )
 
-    def test_staging_host_url(self):
-        assert (
-            _smith_dashboard_base_url("https://staging.api.host.langchain.com")
-            == "https://staging.smith.langchain.com"
+        assert _select_source(**{**self.OPTIONS, **flags}) == expected
+
+    def test_push_to_build_requires_local_docker(self, monkeypatch):
+        monkeypatch.setattr(
+            deploy_mod, "can_build_locally", lambda: (False, "Docker is required")
         )
 
-    def test_localhost(self):
-        assert (
-            _smith_dashboard_base_url("http://localhost:8080")
-            == "http://localhost:8080"
-        )
+        with pytest.raises(click.UsageError, match="Docker is required"):
+            _select_source(**{**self.OPTIONS, "push_to": self.REPOSITORY})
 
-    def test_localhost_trailing_slash(self):
-        assert (
-            _smith_dashboard_base_url("http://localhost:8080/")
-            == "http://localhost:8080"
-        )
+    @pytest.mark.parametrize(
+        ("flags", "message"),
+        [
+            pytest.param(
+                {"push_to": REPOSITORY, "remote_build_flag": True},
+                "--push-to cannot be combined with --remote.",
+                id="push_to_with_remote",
+            ),
+            pytest.param(
+                {"push_to": f"{REPOSITORY}:v1", "tag": "v2"},
+                "already includes a tag",
+                id="push_to_with_a_tag_and_the_tag_flag",
+            ),
+            pytest.param(
+                {"push_to": f"{REPOSITORY}@sha256:abc"},
+                "not a digest",
+                id="push_to_with_a_digest",
+            ),
+            pytest.param(
+                {"image": "app:dev", "remote_build_flag": True},
+                "--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):
+        monkeypatch.setattr(deploy_mod, "can_build_locally", lambda: (True, None))
 
-    def test_127_0_0_1(self):
-        assert (
-            _smith_dashboard_base_url("http://127.0.0.1:3000")
-            == "http://127.0.0.1:3000"
-        )
-
-    def test_unknown_domain_returns_default(self):
-        assert (
-            _smith_dashboard_base_url("https://custom.example.com")
-            == "https://smith.langchain.com"
-        )
+        with pytest.raises(click.UsageError, match=message):
+            _select_source(**{**self.OPTIONS, **flags})
 
 
 class TestResolvePushedImageDigest:
@@ -644,6 +820,16 @@ class TestResolvePushedImageDigest:
         )
         assert out == "us-central1-docker.pkg.dev/proj/repo@sha256:abc123"
 
+    def test_registry_port_without_tag_still_resolves_the_digest(self):
+        runner = self._runner('["localhost:5000/repo@sha256:abc123"]')
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image="localhost:5000/repo",
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == "localhost:5000/repo@sha256:abc123"
+
     def test_empty_repodigests_falls_back_with_warning(self, mocker):
         emitter = mocker.MagicMock()
         mocker.patch("langgraph_cli.deploy._get_emitter", return_value=emitter)
@@ -747,3 +933,289 @@ class TestResolvePushedImageDigest:
         frame_locals = captured["coro"].cr_frame.f_locals
         assert "--config" not in frame_locals["args"]
         captured["coro"].close()
+
+
+class TestListener:
+    @pytest.mark.parametrize(
+        ("resource", "expected"),
+        [
+            pytest.param(
+                {
+                    "id": "listener-1",
+                    "compute_id": "prod-cluster",
+                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+                },
+                Listener("listener-1", "prod-cluster", ("agents", "agents-staging")),
+                id="reads_id_cluster_and_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": {}},
+                Listener("listener-1", "c", ()),
+                id="missing_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": None},
+                Listener("listener-1", "c", ()),
+                id="null_compute_config",
+            ),
+            pytest.param(
+                {"id": "listener-1"},
+                Listener("listener-1", "", ()),
+                id="only_an_id",
+            ),
+        ],
+    )
+    def test_from_resource_reads_the_control_plane_shape(self, resource, expected):
+        assert Listener.from_resource(resource) == expected
+
+
+ONE_NAMESPACE = Listener("listener-1", "prod-cluster", ("agents",))
+TWO_NAMESPACES = Listener("listener-2", "multi-cluster", ("agents", "agents-staging"))
+NO_NAMESPACE = Listener("listener-3", "broken-cluster", ())
+
+
+class TestRequestedPlacement:
+    @pytest.mark.parametrize(
+        ("request_", "listeners", "expected"),
+        [
+            pytest.param(
+                RequestedPlacement(), (), Unplaced(), id="no_listeners_no_request"
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (ONE_NAMESPACE,),
+                OnListener("listener-1", "agents"),
+                id="uses_the_only_possible_answer",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents-staging"),
+                (TWO_NAMESPACES,),
+                OnListener("listener-2", "agents-staging"),
+                id="namespace_alone_picks_the_only_listener",
+            ),
+        ],
+    )
+    def test_resolves_to_a_placement(self, request_, listeners, expected):
+        assert request_.among(listeners) == expected
+
+    @pytest.mark.parametrize(
+        ("request_", "listeners", "message"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-1"),
+                (),
+                "no listeners",
+                id="workspace_has_no_listeners",
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (ONE_NAMESPACE, TWO_NAMESPACES),
+                "--listener-id",
+                id="several_listeners_need_a_choice",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents"),
+                (ONE_NAMESPACE, TWO_NAMESPACES),
+                "--listener-id",
+                id="namespace_alone_is_ambiguous_with_several_listeners",
+            ),
+            pytest.param(
+                RequestedPlacement(k8s_namespace="agents"),
+                (),
+                "no listeners",
+                id="namespace_without_any_listener",
+            ),
+            pytest.param(
+                RequestedPlacement(),
+                (TWO_NAMESPACES,),
+                "--k8s-namespace",
+                id="several_namespaces_need_a_choice",
+            ),
+        ],
+    )
+    def test_refuses_and_names_the_choices(self, request_, listeners, message):
+        with pytest.raises(click.UsageError, match=message):
+            request_.among(listeners)
+
+    def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self):
+        with pytest.raises(click.UsageError) as error:
+            RequestedPlacement().among((ONE_NAMESPACE, TWO_NAMESPACES))
+
+        assert "listener-1" in error.value.message
+        assert "prod-cluster" in error.value.message
+        assert "agents-staging" in error.value.message
+
+    @pytest.mark.parametrize(
+        ("placement", "expected"),
+        [
+            pytest.param(Unplaced(), {}, id="unplaced_adds_nothing"),
+            pytest.param(
+                OnListener("listener-1", "agents"),
+                {
+                    "listener_id": "listener-1",
+                    "listener_config": {"k8s_namespace": "agents"},
+                },
+                id="placed_carries_listener_and_namespace",
+            ),
+        ],
+    )
+    def test_source_config_matches_the_control_plane_shape(self, placement, expected):
+        assert placement.source_config() == expected
+
+
+def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_version():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["params"] = dict(req.url.params)
+        return httpx.Response(
+            200,
+            json={"resources": [{"id": "dep-1", "name": "agent", "source": "github"}]},
+        )
+
+    client = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    found = find_deployment_by_name(client, "agent")
+
+    assert seen["params"] == {
+        "name": "agent",
+        "name_contains": "agent",
+        "limit": "100",
+    }
+    assert found == ExistingDeployment("dep-1", "github")
+
+
+def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deployment():
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(
+                200,
+                json={
+                    "resources": [
+                        {
+                            "id": "dep-other",
+                            "name": "another-teams-agent",
+                            "source": "external_docker",
+                        }
+                    ]
+                },
+            )
+        ),
+    )
+
+    assert find_deployment_by_name(client, "brand-new-agent") is None
+
+
+def test_a_full_page_without_a_match_refuses_to_claim_the_name_is_free():
+    page = [
+        {"id": f"dep-{index}", "name": f"other-agent-{index}"} for index in range(100)
+    ]
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(200, json={"resources": page})
+        ),
+    )
+
+    with pytest.raises(click.ClickException, match="--deployment-id"):
+        find_deployment_by_name(client, "brand-new-agent")
+
+
+def test_a_partial_page_without_a_match_means_the_name_is_free():
+    client = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(
+                200, json={"resources": [{"id": "dep-1", "name": "other"}]}
+            )
+        ),
+    )
+
+    assert find_deployment_by_name(client, "brand-new-agent") is None
+
+
+@pytest.mark.parametrize(
+    "resource",
+    [
+        pytest.param({"compute_id": "c"}, id="no_id"),
+        pytest.param({"id": ""}, id="empty_id"),
+    ],
+)
+def test_a_listener_without_an_id_is_refused(resource):
+    with pytest.raises(HostBackendError, match="without an id"):
+        Listener.from_resource(resource)
+
+
+def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
+    monkeypatch,
+):
+    def explode() -> tuple[bool, str | None]:
+        raise AssertionError("docker must not be probed for an argv-only conflict")
+
+    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
+
+    with pytest.raises(click.UsageError, match="--deployment-id"):
+        _select_source(
+            push_to="registry.example.com/app",
+            image=None,
+            image_name=None,
+            tag=None,
+            remote_build_flag=None,
+            placement=RequestedPlacement(listener_id="listener-1"),
+            selector=ById("dep-1"),
+        )
+
+
+class TestPlacementOnAKnownListener:
+    @pytest.mark.parametrize(
+        ("request_", "listener", "expected"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-1"),
+                ONE_NAMESPACE,
+                OnListener("listener-1", "agents"),
+                id="the_only_namespace_is_used",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"),
+                TWO_NAMESPACES,
+                OnListener("listener-2", "agents"),
+                id="the_chosen_namespace_is_used",
+            ),
+        ],
+    )
+    def test_places_on_the_listener(self, request_, listener, expected):
+        assert request_.on(listener) == expected
+
+    @pytest.mark.parametrize(
+        ("request_", "listener", "message"),
+        [
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2"),
+                TWO_NAMESPACES,
+                "--k8s-namespace",
+                id="several_namespaces_need_a_choice",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-2", k8s_namespace="nope"),
+                TWO_NAMESPACES,
+                "does not serve namespace",
+                id="unknown_namespace",
+            ),
+            pytest.param(
+                RequestedPlacement(listener_id="listener-3"),
+                NO_NAMESPACE,
+                "serves no namespaces",
+                id="listener_without_namespaces",
+            ),
+        ],
+    )
+    def test_refuses_and_names_the_namespaces(self, request_, listener, message):
+        with pytest.raises(click.UsageError, match=message):
+            request_.on(listener)

libs/cli/tests/unit_tests/test_host_backend.py

@@ -3,29 +3,16 @@ import json
 import httpx
 import pytest
 
-from langgraph_cli.host_backend import HostBackendClient, HostBackendError
-
-
-@pytest.fixture
-def mock_transport():
-    return httpx.MockTransport(lambda req: httpx.Response(200, json={"ok": True}))
-
-
-@pytest.fixture
-def client(mock_transport):
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=mock_transport,
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
-    )
-    return c
+from langgraph_cli.host_backend import (
+    ControlPlaneEndpoints,
+    HostBackendClient,
+    HostBackendError,
+)
 
 
 def test_constructor_strips_trailing_slash():
     c = HostBackendClient("https://api.example.com/", "key")
-    assert str(c._client.base_url) == "https://api.example.com"
+    assert c.base_url == "https://api.example.com"
 
 
 def test_constructor_empty_url_raises():
@@ -39,12 +26,8 @@ def test_request_sends_headers():
         assert req.headers["accept"] == "application/json"
         return httpx.Response(200, json={"ok": True})
 
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
     )
     result = c._request("GET", "/test")
     assert result == {"ok": True}
@@ -56,12 +39,8 @@ def test_request_sends_json_payload():
         assert req.content == b'{"key":"value"}'
         return httpx.Response(200, json={"created": True})
 
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
     )
     result = c._request("POST", "/test", {"key": "value"})
     assert result == {"created": True}
@@ -69,25 +48,13 @@ def test_request_sends_json_payload():
 
 def test_request_empty_body_returns_none():
     transport = httpx.MockTransport(lambda req: httpx.Response(200, content=b""))
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=transport,
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
-    )
+    c = HostBackendClient("https://api.example.com", "test-key", transport=transport)
     assert c._request("DELETE", "/test") is None
 
 
 def test_request_http_error_raises():
     transport = httpx.MockTransport(lambda req: httpx.Response(404, text="not found"))
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=transport,
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
-    )
+    c = HostBackendClient("https://api.example.com", "test-key", transport=transport)
     with pytest.raises(HostBackendError, match="404"):
         c._request("GET", "/missing")
 
@@ -96,13 +63,7 @@ def test_request_invalid_json_raises():
     transport = httpx.MockTransport(
         lambda req: httpx.Response(200, content=b"not json")
     )
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=transport,
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
-    )
+    c = HostBackendClient("https://api.example.com", "test-key", transport=transport)
     with pytest.raises(HostBackendError, match="Failed to decode"):
         c._request("GET", "/bad-json")
 
@@ -111,84 +72,20 @@ def test_request_transport_error_raises():
     def handler(req: httpx.Request) -> httpx.Response:
         raise httpx.ConnectError("connection refused")
 
-    c = HostBackendClient("https://api.example.com", "test-key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
     )
     with pytest.raises(HostBackendError, match="connection refused"):
         c._request("GET", "/test")
 
 
-def test_create_deployment(client):
-    result = client.create_deployment(
-        name="my-deploy", deployment_type="dev", source="internal_docker"
-    )
-    assert result == {"ok": True}
-
-
-def test_get_deployment(client):
-    result = client.get_deployment("dep-123")
-    assert result == {"ok": True}
-
-
-def test_list_deployments(client):
-    result = client.list_deployments("my-app")
-    assert result == {"ok": True}
-
-
-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")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
-        timeout=30,
-    )
-    result = c.list_deployments("my app")
-    assert result == {"ok": True}
-
-
-def test_delete_deployment(client):
-    result = client.delete_deployment("dep-123")
-    assert result == {"ok": True}
-
-
-def test_request_push_token(client):
-    result = client.request_push_token("dep-123")
-    assert result == {"ok": True}
-
-
-def test_update_deployment(client):
-    result = client.update_deployment(
-        "dep-123", "image:latest", secrets=[{"name": "KEY", "value": "val"}]
-    )
-    assert result == {"ok": True}
-
-
-def test_update_deployment_no_secrets(client):
-    result = client.update_deployment("dep-123", "image:latest")
-    assert result == {"ok": True}
-
-
 def _capturing_client(captured: dict) -> HostBackendClient:
     def handler(req: httpx.Request) -> httpx.Response:
         captured["body"] = req.read()
         return httpx.Response(200, json={"ok": True})
 
-    c = HostBackendClient("https://api.example.com", "key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
     )
     return c
 
@@ -199,6 +96,7 @@ def test_update_deployment_forwards_tracked_packages():
     c.update_deployment(
         "dep-123",
         "image:latest",
+        revision_source="internal_docker",
         tracked_packages=["google-adk:1.0.0"],
     )
     body = json.loads(captured["body"])
@@ -209,7 +107,7 @@ def test_update_deployment_forwards_tracked_packages():
 def test_update_deployment_omits_tracked_packages_when_absent():
     captured: dict = {}
     c = _capturing_client(captured)
-    c.update_deployment("dep-123", "image:latest")
+    c.update_deployment("dep-123", "image:latest", revision_source="internal_docker")
     body = json.loads(captured["body"])
     assert "tracked_packages" not in body
 
@@ -241,33 +139,14 @@ def test_update_deployment_internal_source_omits_tracked_packages_when_absent():
     assert "tracked_packages" not in body
 
 
-def test_list_revisions(client):
-    result = client.list_revisions("dep-123", limit=5)
-    assert result == {"ok": True}
-
-
-def test_get_revision(client):
-    result = client.get_revision("dep-123", "rev-456")
-    assert result == {"ok": True}
-
-
-def test_get_build_logs(client):
-    result = client.get_build_logs("proj-1", "rev-1", {"limit": 10})
-    assert result == {"ok": True}
-
-
 def test_get_deploy_logs_all_revisions():
     def handler(req: httpx.Request) -> httpx.Response:
         assert "/v1/projects/proj-1/deploy_logs" in str(req.url)
         assert "/revisions/" not in str(req.url)
         return httpx.Response(200, json={"logs": [{"message": "running"}]})
 
-    c = HostBackendClient("https://api.example.com", "key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
     )
     result = c.get_deploy_logs("proj-1", {"limit": 10})
     assert result == {"logs": [{"message": "running"}]}
@@ -278,12 +157,520 @@ def test_get_deploy_logs_specific_revision():
         assert "/v1/projects/proj-1/revisions/rev-2/deploy_logs" in str(req.url)
         return httpx.Response(200, json={"logs": []})
 
-    c = HostBackendClient("https://api.example.com", "key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "key", "Accept": "application/json"},
-        timeout=30,
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
     )
     result = c.get_deploy_logs("proj-1", {"limit": 10}, revision_id="rev-2")
     assert result == {"logs": []}
+
+
+def _routing_client(seen: dict) -> HostBackendClient:
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["method"] = req.method
+        seen["url"] = str(req.url)
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient(
+        "https://api.example.com/prefix", "key", transport=httpx.MockTransport(handler)
+    )
+    return c
+
+
+@pytest.mark.parametrize(
+    ("call", "expected_body"),
+    [
+        pytest.param(
+            lambda c: c.create_deployment(
+                name="my-deploy",
+                source="internal_docker",
+                source_config={"deployment_type": "dev"},
+                source_revision_config={},
+            ),
+            {
+                "name": "my-deploy",
+                "source": "internal_docker",
+                "source_config": {"deployment_type": "dev"},
+                "source_revision_config": {},
+            },
+            id="internal_docker_create_omits_secrets_key_when_not_given",
+        ),
+        pytest.param(
+            lambda c: c.create_deployment(
+                name="my-deploy",
+                source="internal_docker",
+                source_config={"deployment_type": "prod"},
+                source_revision_config={},
+                secrets=[{"name": "KEY", "value": "val"}],
+            ),
+            {
+                "name": "my-deploy",
+                "source": "internal_docker",
+                "source_config": {"deployment_type": "prod"},
+                "source_revision_config": {},
+                "secrets": [{"name": "KEY", "value": "val"}],
+            },
+            id="internal_docker_create_forwards_secrets",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment(
+                "dep-123",
+                "registry.example.com/app@sha256:abc",
+                revision_source="internal_docker",
+                secrets=[{"name": "KEY", "value": "val"}],
+            ),
+            {
+                "revision_source": "internal_docker",
+                "source_revision_config": {
+                    "image_uri": "registry.example.com/app@sha256:abc"
+                },
+                "secrets": [{"name": "KEY", "value": "val"}],
+            },
+            id="internal_docker_revision_names_its_source",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment_internal_source(
+                "dep-123",
+                source_tarball_path="tarballs/src.tgz",
+                config_path="langgraph.json",
+                secrets=[],
+                install_command="yarn install",
+                build_command="yarn build",
+            ),
+            {
+                "revision_source": "internal_source",
+                "source_revision_config": {
+                    "source_tarball_path": "tarballs/src.tgz",
+                    "langgraph_config_path": "langgraph.json",
+                },
+                "source_config": {
+                    "install_command": "yarn install",
+                    "build_command": "yarn build",
+                },
+                "secrets": [],
+            },
+            id="internal_source_revision_sends_js_build_commands",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment_internal_source(
+                "dep-123",
+                source_tarball_path="tarballs/src.tgz",
+                config_path="langgraph.json",
+            ),
+            {
+                "revision_source": "internal_source",
+                "source_revision_config": {
+                    "source_tarball_path": "tarballs/src.tgz",
+                    "langgraph_config_path": "langgraph.json",
+                },
+            },
+            id="internal_source_revision_omits_source_config_without_commands",
+        ),
+        pytest.param(
+            lambda c: c.create_deployment(
+                name="agent",
+                source="external_docker",
+                source_config={"resource_spec": {}},
+                source_revision_config={
+                    "image_uri": "registry.example.com/agent@sha256:1"
+                },
+                secrets=[],
+            ),
+            {
+                "name": "agent",
+                "source": "external_docker",
+                "source_config": {"resource_spec": {}},
+                "source_revision_config": {
+                    "image_uri": "registry.example.com/agent@sha256:1"
+                },
+                "secrets": [],
+            },
+            id="create_sends_the_source_configs_as_given",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment(
+                "dep-1", "registry.example.com/agent@sha256:2", revision_source=None
+            ),
+            {
+                "source_revision_config": {
+                    "image_uri": "registry.example.com/agent@sha256:2"
+                }
+            },
+            id="revision_without_source_override_omits_revision_source",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment(
+                "dep-1",
+                "registry.example.com/agent@sha256:2",
+                revision_source="internal_docker",
+                tracked_packages=["langgraph:1.0.0"],
+            ),
+            {
+                "revision_source": "internal_docker",
+                "source_revision_config": {
+                    "image_uri": "registry.example.com/agent@sha256:2"
+                },
+                "tracked_packages": ["langgraph:1.0.0"],
+            },
+            id="revision_with_source_override_names_it",
+        ),
+    ],
+)
+def test_request_body_matches_control_plane_contract(call, expected_body):
+    captured: dict = {}
+    call(_capturing_client(captured))
+    assert json.loads(captured["body"]) == expected_body
+
+
+@pytest.mark.parametrize(
+    ("call", "method", "route"),
+    [
+        pytest.param(
+            lambda c: c.create_deployment(
+                name="n",
+                source="internal_docker",
+                source_config={"deployment_type": "dev"},
+                source_revision_config={},
+            ),
+            "POST",
+            "/v2/deployments",
+            id="create_deployment",
+        ),
+        pytest.param(
+            lambda c: c.get_deployment("dep-1"),
+            "GET",
+            "/v2/deployments/dep-1",
+            id="get_deployment",
+        ),
+        pytest.param(
+            lambda c: c.delete_deployment("dep-1"),
+            "DELETE",
+            "/v2/deployments/dep-1",
+            id="delete_deployment",
+        ),
+        pytest.param(
+            lambda c: c.update_deployment("dep-1", "img", revision_source=None),
+            "PATCH",
+            "/v2/deployments/dep-1",
+            id="patch_deployment",
+        ),
+        pytest.param(
+            lambda c: c.request_push_token("dep-1"),
+            "POST",
+            "/v2/deployments/dep-1/push-token",
+            id="push_token",
+        ),
+        pytest.param(
+            lambda c: c.request_upload_url("dep-1"),
+            "POST",
+            "/v2/deployments/dep-1/upload-url",
+            id="upload_url",
+        ),
+        pytest.param(
+            lambda c: c.list_revisions("dep-1", limit=5),
+            "GET",
+            "/v2/deployments/dep-1/revisions?limit=5",
+            id="list_revisions_puts_limit_in_query",
+        ),
+        pytest.param(
+            lambda c: c.get_revision("dep-1", "rev-2"),
+            "GET",
+            "/v2/deployments/dep-1/revisions/rev-2",
+            id="get_revision",
+        ),
+        pytest.param(
+            lambda c: c.get_build_logs("dep-1", "rev-2", {"limit": 10}),
+            "POST",
+            "/v1/projects/dep-1/revisions/rev-2/build_logs",
+            id="build_logs",
+        ),
+    ],
+)
+def test_request_targets_control_plane_route_under_base_url(call, method, route):
+    seen: dict = {}
+    call(_routing_client(seen))
+    assert (seen["method"], seen["url"]) == (
+        method,
+        f"https://api.example.com/prefix{route}",
+    )
+
+
+def test_injected_transport_receives_requests_under_the_prefixed_base_url():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["url"] = str(req.url)
+        seen["api_key"] = req.headers["x-api-key"]
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient(
+        "https://smith.example.com/api-host",
+        "key",
+        transport=httpx.MockTransport(handler),
+    )
+
+    assert c.list_revisions("dep-1", limit=2) == []
+    assert seen == {
+        "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
+        "api_key": "key",
+    }
+
+
+CLOUD = ("https://api.host.langchain.com", "https://smith.langchain.com")
+
+
+@pytest.mark.parametrize(
+    ("host_url", "langsmith_endpoint", "expected"),
+    [
+        pytest.param(None, None, CLOUD, id="nothing_configured_targets_cloud"),
+        pytest.param(
+            None, "https://api.smith.langchain.com", CLOUD, id="cloud_langsmith_api"
+        ),
+        pytest.param(
+            None,
+            "https://api.smith.langchain.com/api/v1",
+            CLOUD,
+            id="cloud_langsmith_api_with_versioned_path",
+        ),
+        pytest.param(
+            None, "https://api.langchain.com", CLOUD, id="cloud_langchain_api_alias"
+        ),
+        pytest.param(
+            None,
+            "https://xapi.smith.langchain.com",
+            CLOUD,
+            id="lookalike_cloud_host_is_not_rewritten_into_a_control_plane",
+        ),
+        pytest.param(
+            None,
+            "https://eu.api.smith.langchain.com",
+            ("https://eu.api.host.langchain.com", "https://eu.smith.langchain.com"),
+            id="eu_cloud_maps_to_eu_control_plane",
+        ),
+        pytest.param(
+            None,
+            "https://dev.api.smith.langchain.com",
+            ("https://dev.api.host.langchain.com", "https://dev.smith.langchain.com"),
+            id="dev_cloud_maps_to_dev_control_plane",
+        ),
+        pytest.param(
+            None,
+            "https://aks.smith.langchain.dev/api",
+            (
+                "https://aks.smith.langchain.dev/api-host",
+                "https://aks.smith.langchain.dev",
+            ),
+            id="self_hosted_api_path_becomes_api_host",
+        ),
+        pytest.param(
+            None,
+            "https://smith.example.com/api/v1",
+            ("https://smith.example.com/api-host", "https://smith.example.com"),
+            id="self_hosted_versioned_api_path_becomes_api_host",
+        ),
+        pytest.param(
+            None,
+            "https://smith.example.com",
+            ("https://smith.example.com/api-host", "https://smith.example.com"),
+            id="self_hosted_origin_gets_api_host_appended",
+        ),
+        pytest.param(
+            None,
+            "https://corp.example.com/langsmith/api/v1",
+            (
+                "https://corp.example.com/langsmith/api-host",
+                "https://corp.example.com/langsmith",
+            ),
+            id="self_hosted_path_prefix_is_kept",
+        ),
+        pytest.param(
+            "https://custom.host.example",
+            "https://aks.smith.langchain.dev/api",
+            ("https://custom.host.example", "https://smith.langchain.com"),
+            id="explicit_host_url_beats_langsmith_endpoint",
+        ),
+        pytest.param(
+            "https://api.host.langchain.com",
+            "https://aks.smith.langchain.dev/api",
+            CLOUD,
+            id="explicit_cloud_host_url_beats_self_hosted_endpoint",
+        ),
+        pytest.param(
+            "https://smith.example.com/api-host/",
+            None,
+            ("https://smith.example.com/api-host", "https://smith.example.com"),
+            id="explicit_api_host_url_derives_dashboard_root",
+        ),
+        pytest.param(
+            "https://corp.example.com/langsmith/api-host",
+            None,
+            (
+                "https://corp.example.com/langsmith/api-host",
+                "https://corp.example.com/langsmith",
+            ),
+            id="explicit_api_host_url_keeps_path_prefix_in_dashboard",
+        ),
+        pytest.param(
+            "http://localhost:8080",
+            None,
+            ("http://localhost:8080", "http://localhost:8080"),
+            id="localhost_dashboard_is_the_same_origin",
+        ),
+        pytest.param(
+            "http://localhost:8080/api-host",
+            None,
+            ("http://localhost:8080/api-host", "http://localhost:8080"),
+            id="localhost_api_host_dashboard_is_the_origin",
+        ),
+        pytest.param(
+            "https://eu.api.host.langchain.com",
+            None,
+            ("https://eu.api.host.langchain.com", "https://eu.smith.langchain.com"),
+            id="regional_control_plane_maps_to_regional_dashboard",
+        ),
+    ],
+)
+def test_control_plane_endpoints_resolve(host_url, langsmith_endpoint, expected):
+    endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
+
+    assert (endpoints.control_plane_url, endpoints.dashboard_url) == expected
+
+
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        pytest.param(
+            {"resources": [{"id": "a"}, {"id": "b"}]},
+            [{"id": "a"}, {"id": "b"}],
+            id="list_returns_the_resources",
+        ),
+        pytest.param({"resources": []}, [], id="empty_list"),
+        pytest.param({}, [], id="missing_key"),
+        pytest.param({"resources": None}, [], id="null_resources"),
+        pytest.param(
+            {"resources": ["nope", {"id": "a"}]}, [{"id": "a"}], id="skips_non_objects"
+        ),
+        pytest.param([], [], id="unexpected_envelope"),
+    ],
+)
+def test_list_endpoints_return_resource_objects(payload, expected):
+    def handler(req: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=payload)
+
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    assert c.list_deployments() == expected
+
+
+def test_list_listeners_asks_for_a_full_page():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["url"] = str(req.url)
+        return httpx.Response(200, json={"resources": [{"id": "listener-1"}]})
+
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    assert c.list_listeners() == [{"id": "listener-1"}]
+    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"
+
+
+@pytest.mark.parametrize(
+    ("control_plane_url", "expected"),
+    [
+        pytest.param("https://api.host.langchain.com", True, id="cloud"),
+        pytest.param("https://eu.api.host.langchain.com", True, id="cloud_region"),
+        pytest.param("https://dev.api.host.langchain.com", True, id="cloud_dev"),
+        pytest.param("https://smith.example.com/api-host", False, id="self_hosted"),
+        pytest.param(
+            "https://corp.example.com/langsmith/api-host",
+            False,
+            id="self_hosted_prefix",
+        ),
+        pytest.param("http://localhost:8080/api-host", False, id="local"),
+        pytest.param(
+            "https://evil-api.host.langchain.com", False, id="lookalike_needs_a_dot"
+        ),
+    ],
+)
+def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expected):
+    endpoints = ControlPlaneEndpoints.from_control_plane_url(control_plane_url)
+
+    assert endpoints.is_cloud is expected
+
+
+@pytest.mark.parametrize(
+    ("call", "expected_params"),
+    [
+        pytest.param(
+            lambda c: c.list_deployments(name="agent"),
+            {"name": "agent"},
+            id="exact_name_filters_server_side",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(name_contains="age"),
+            {"name_contains": "age"},
+            id="substring_search_keeps_its_own_parameter",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(),
+            {},
+            id="no_filter_sends_no_parameters",
+        ),
+        pytest.param(
+            lambda c: c.list_deployments(
+                name="agent", name_contains="agent", limit=100
+            ),
+            {"name": "agent", "name_contains": "agent", "limit": "100"},
+            id="both_filters_travel_together_for_older_servers",
+        ),
+    ],
+)
+def test_list_deployments_sends_one_name_filter(call, expected_params):
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen.update(dict(req.url.params))
+        return httpx.Response(200, json={"resources": []})
+
+    call(
+        HostBackendClient(
+            "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+        )
+    )
+
+    assert seen == expected_params
+
+
+@pytest.mark.parametrize(
+    ("body", "expected"),
+    [
+        pytest.param(
+            {"detail": "Source configuration error: bad listener"},
+            "Source configuration error: bad listener",
+            id="fastapi_detail_is_unwrapped",
+        ),
+        pytest.param(
+            {"detail": {"loc": ["body"], "msg": "nope"}},
+            None,
+            id="a_structured_detail_is_left_alone",
+        ),
+        pytest.param({"other": "shape"}, None, id="an_unknown_shape_is_left_alone"),
+    ],
+)
+def test_error_detail_is_readable(body, expected):
+    c = HostBackendClient(
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(lambda req: httpx.Response(400, json=body)),
+    )
+
+    with pytest.raises(HostBackendError) as error:
+        c.get_deployment("dep-1")
+
+    assert error.value.detail == expected
+    if expected is not None:
+        assert error.value.message.endswith(expected)

libs/cli/tests/unit_tests/test_image_reference.py

@@ -0,0 +1,71 @@
+import pytest
+
+from langgraph_cli.image_reference import ImageReference
+
+
+@pytest.mark.parametrize(
+    ("reference", "repository", "tag"),
+    [
+        pytest.param(
+            "registry.example.com/team/app:v1",
+            "registry.example.com/team/app",
+            "v1",
+            id="tag_after_last_slash",
+        ),
+        pytest.param(
+            "registry.example.com/team/app",
+            "registry.example.com/team/app",
+            None,
+            id="no_tag",
+        ),
+        pytest.param(
+            "localhost:5000/app",
+            "localhost:5000/app",
+            None,
+            id="registry_port_is_not_a_tag",
+        ),
+        pytest.param(
+            "localhost:5000/app:latest",
+            "localhost:5000/app",
+            "latest",
+            id="registry_port_with_tag",
+        ),
+        pytest.param("app:dev", "app", "dev", id="bare_name_with_tag"),
+    ],
+)
+def test_parse_splits_repository_and_tag(reference, repository, tag):
+    assert ImageReference.parse(reference) == ImageReference(repository, tag)
+
+
+def test_with_tag_replaces_the_tag():
+    assert ImageReference("r/app", "v1").with_tag("v2") == ImageReference("r/app", "v2")
+
+
+@pytest.mark.parametrize(
+    ("reference", "expected"),
+    [
+        pytest.param(ImageReference("r/app", "v1"), "r/app:v1", id="tagged"),
+        pytest.param(ImageReference("r/app"), "r/app", id="untagged"),
+    ],
+)
+def test_str_renders_the_docker_reference(reference, expected):
+    assert str(reference) == expected
+
+
+@pytest.mark.parametrize(
+    ("repo_digest", "expected"),
+    [
+        pytest.param("localhost:5000/app@sha256:abc", True, id="same_repository"),
+        pytest.param("localhost:5000/app-2@sha256:abc", False, id="other_repository"),
+        pytest.param("mirror.example.com/app@sha256:abc", False, id="other_registry"),
+    ],
+)
+def test_matches_digest_only_for_the_same_repository(repo_digest, expected):
+    assert ImageReference("localhost:5000/app", "v1").matches_digest(repo_digest) is (
+        expected
+    )
+
+
+def test_parse_rejects_a_digest_reference():
+    with pytest.raises(ValueError, match="digest"):
+        ImageReference.parse("registry.example.com/app@sha256:abc")