test(cli): drop client tests covered by the route and body tables

Reverts 9779ea5d3 at the requester’s direction.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

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,11 +2,86 @@
 
 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")
+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)
+
+    @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
+
 
 class HostBackendError(click.ClickException):
     """Raised when the host backend returns an error response."""
@@ -24,10 +99,11 @@ 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",
@@ -38,10 +114,17 @@ class HostBackendClient:
         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
+
+    def set_tenant(self, tenant_id: str) -> None:
+        self._client.headers["X-Tenant-ID"] = tenant_id
+
     def _request(
         self,
         method: str,
@@ -72,21 +155,19 @@ class HostBackendClient:
 
     def create_deployment(
         self,
+        *,
         name: str,
-        deployment_type: str,
-        source: str,
-        config_path: str | None = None,
+        source: SourceName,
+        source_config: dict[str, object],
+        source_revision_config: dict[str, object],
         secrets: list[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 secrets is not None:
             payload["secrets"] = secrets
         return self._request("POST", "/v2/deployments", payload)
@@ -121,22 +202,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,

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_deploy_command.py

@@ -0,0 +1,654 @@
+import asyncio
+import json
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import click.exceptions
+import httpx
+import pytest
+from click.testing import CliRunner, Result
+
+import langgraph_cli.archive as archive_module
+import langgraph_cli.deploy as deploy_module
+from langgraph_cli.cli import cli
+from langgraph_cli.host_backend import HostBackendClient
+from langgraph_cli.image_reference import ImageReference
+
+CONTROL_PLANE_URL = "https://control-plane.example.com"
+REGISTRY_URL = "https://registry.example.com/team"
+PUSH_TOKEN = "push-token"
+PUSHED_IMAGE = "registry.example.com/team/my-app:latest"
+PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123"
+PUSH_REPOSITORY = "registry.example.com/team/agent"
+EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest"
+EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123"
+LISTENER_REQUIRED = (
+    "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"
+ARCHIVE = ("/tmp/src.tgz", 2048, "langgraph.json")
+OBJECT_PATH = "tarballs/src.tgz"
+PLATFORM_FORMAT = "{{.Os}}/{{.Architecture}}"
+DIGESTS_FORMAT = "{{json .RepoDigests}}"
+NOT_A_CLI_DEPLOYMENT = (
+    "push token is only available for 'internal_docker' source deployments"
+)
+LIST_DEPLOYMENTS = "GET /v2/deployments"
+CREATE_DEPLOYMENT = "POST /v2/deployments"
+
+
+def _push_token(deployment_id: str) -> str:
+    return f"POST /v2/deployments/{deployment_id}/push-token"
+
+
+def _upload_url(deployment_id: str) -> str:
+    return f"POST /v2/deployments/{deployment_id}/upload-url"
+
+
+def _patch(deployment_id: str) -> str:
+    return f"PATCH /v2/deployments/{deployment_id}"
+
+
+def _get(deployment_id: str) -> str:
+    return f"GET /v2/deployments/{deployment_id}"
+
+
+@dataclass
+class ControlPlaneDouble:
+    timeline: list[str]
+    existing_deployments: list[dict] = field(default_factory=list)
+    push_token_status: int = 200
+    create_error: str | None = None
+    bodies: dict[str, dict] = field(default_factory=dict)
+
+    def handle(self, request: httpx.Request) -> httpx.Response:
+        route = f"{request.method} {request.url.path}"
+        self.timeline.append(route)
+        if request.content:
+            self.bodies[route] = json.loads(request.content)
+        return self._respond(request.method, request.url.path)
+
+    def _respond(self, method: str, path: str) -> httpx.Response:
+        if (method, path) == ("GET", "/v2/deployments"):
+            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, 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:
+                return httpx.Response(self.push_token_status, text=NOT_A_CLI_DEPLOYMENT)
+            return httpx.Response(
+                200, json={"token": PUSH_TOKEN, "registry_url": REGISTRY_URL}
+            )
+        if path.endswith("/upload-url"):
+            return httpx.Response(
+                200, json={"upload_url": SIGNED_UPLOAD_URL, "object_path": OBJECT_PATH}
+            )
+        if method == "PATCH":
+            return httpx.Response(200, json={"tenant_id": "tenant-1"})
+        if method == "GET":
+            deployment_id = path.rsplit("/", 1)[-1]
+            return httpx.Response(
+                200,
+                json=next(
+                    d for d in self.existing_deployments if d["id"] == deployment_id
+                ),
+            )
+        raise AssertionError(f"unexpected control plane call: {method} {path}")
+
+    def client_factory(self) -> Callable[..., HostBackendClient]:
+        transport = httpx.MockTransport(self.handle)
+
+        def make(
+            host_url: str, api_key: str, tenant_id: str | None = None
+        ) -> HostBackendClient:
+            return HostBackendClient(host_url, api_key, tenant_id, transport=transport)
+
+        return make
+
+
+@dataclass
+class DockerCommand:
+    args: tuple[str, ...]
+    kwargs: dict
+
+
+@dataclass
+class DockerDouble:
+    timeline: list[str]
+    failing_pushes: int = 0
+    builds: list[dict] = field(default_factory=list)
+    commands: list[DockerCommand] = field(default_factory=list)
+
+    def verbs(self) -> list[str]:
+        return [event for event in self.timeline if event.startswith("docker ")]
+
+    def command(self, verb: str) -> DockerCommand:
+        return next(c for c in self.commands if verb in c.args)
+
+    def build_docker_image(
+        self,
+        runner: object,
+        set_message: Callable[[str], None],
+        config: Path,
+        config_json: dict,
+        base_image: str | None,
+        api_version: str | None,
+        pull: bool,
+        tag: str,
+        passthrough: tuple[str, ...] = (),
+        install_command: str | None = None,
+        build_command: str | None = None,
+        docker_command: tuple[str, ...] | None = None,
+        extra_flags: tuple[str, ...] = (),
+        verbose: bool = True,
+    ) -> None:
+        self.timeline.append("docker build")
+        self.builds.append(
+            {
+                "tag": tag,
+                "docker_command": tuple(docker_command or ("docker", "build")),
+                "extra_flags": tuple(extra_flags),
+            }
+        )
+
+    async def subp_exec(
+        self, *args: str, **kwargs: object
+    ) -> tuple[str | None, str | None]:
+        self.commands.append(DockerCommand(args=args, kwargs=kwargs))
+        self.timeline.append(f"docker {self._verb(args)}")
+        if "push" in args and self.failing_pushes > 0:
+            self.failing_pushes -= 1
+            raise click.exceptions.Exit(1)
+        if PLATFORM_FORMAT in args:
+            return "linux/amd64\n", None
+        if DIGESTS_FORMAT in args:
+            repository = ImageReference.parse(args[-1]).repository
+            return json.dumps([f"{repository}@sha256:abc123"]), None
+        return None, None
+
+    @staticmethod
+    def _verb(args: tuple[str, ...]) -> str:
+        if PLATFORM_FORMAT in args:
+            return "inspect-platform"
+        if DIGESTS_FORMAT in args:
+            return "inspect-digest"
+        return next(verb for verb in ("login", "tag", "push", "pull") if verb in args)
+
+
+class _AsyncioRunner:
+    def run(self, coro):
+        return asyncio.run(coro)
+
+
+@contextmanager
+def _fake_runner() -> Iterator[_AsyncioRunner]:
+    yield _AsyncioRunner()
+
+
+@dataclass
+class DeployProject:
+    control_plane: ControlPlaneDouble
+    docker: DockerDouble
+    timeline: list[str]
+    uploads: list[tuple[str, str, int]]
+
+    def run(self, *args: str) -> Result:
+        return CliRunner().invoke(
+            cli,
+            [
+                "deploy",
+                "--api-key",
+                "test-key",
+                "--host-url",
+                CONTROL_PLANE_URL,
+                "--name",
+                "my-app",
+                "--no-input",
+                "--no-wait",
+                *args,
+            ],
+        )
+
+
+@pytest.fixture
+def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployProject:
+    (tmp_path / "langgraph.json").write_text(
+        json.dumps({"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]})
+    )
+    monkeypatch.chdir(tmp_path)
+    for name in ("LANGSMITH_TENANT_ID", "LANGSMITH_ENDPOINT", "LANGGRAPH_HOST_URL"):
+        monkeypatch.delenv(name, raising=False)
+
+    timeline: list[str] = []
+    control_plane = ControlPlaneDouble(timeline)
+    docker = DockerDouble(timeline)
+    uploads: list[tuple[str, str, int]] = []
+
+    @contextmanager
+    def fake_create_archive(config_path: Path, config: dict) -> Iterator[tuple]:
+        timeline.append("create_archive")
+        yield ARCHIVE
+
+    def fake_upload(signed_url: str, file_path: str, file_size: int) -> None:
+        timeline.append("upload_archive")
+        uploads.append((signed_url, file_path, file_size))
+
+    monkeypatch.setattr(deploy_module, "_no_input", False)
+    monkeypatch.setattr(deploy_module, "_emitter", None)
+    monkeypatch.setattr(
+        deploy_module, "HostBackendClient", control_plane.client_factory()
+    )
+    monkeypatch.setattr(deploy_module, "build_docker_image", docker.build_docker_image)
+    monkeypatch.setattr(deploy_module, "subp_exec", docker.subp_exec)
+    monkeypatch.setattr(deploy_module, "Runner", _fake_runner)
+    monkeypatch.setattr(deploy_module, "can_build_locally", lambda: (True, None))
+    monkeypatch.setattr(
+        deploy_module,
+        "find_tracked_packages",
+        lambda config, config_json: TRACKED_PACKAGES,
+    )
+    monkeypatch.setattr(deploy_module.platform, "machine", lambda: "x86_64")
+    monkeypatch.setattr(archive_module, "create_archive", fake_create_archive)
+    monkeypatch.setattr(deploy_module, "_upload_to_gcs", fake_upload)
+    return DeployProject(control_plane, docker, timeline, uploads)
+
+
+def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        LIST_DEPLOYMENTS,
+        CREATE_DEPLOYMENT,
+        "docker build",
+        _push_token(CREATED_ID),
+        "docker login",
+        "docker tag",
+        "docker push",
+        "docker inspect-digest",
+        _patch(CREATED_ID),
+    ]
+    assert "Deployment updated" in result.output
+
+
+def test_first_local_deploy_creates_an_internal_docker_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT] == {
+        "name": "my-app",
+        "source": "internal_docker",
+        "source_config": {"deployment_type": "dev"},
+        "source_revision_config": {},
+        "secrets": [],
+    }
+
+
+@pytest.mark.parametrize(
+    ("machine", "expected_command", "expected_flags"),
+    [
+        pytest.param(
+            "arm64",
+            ("docker", "buildx", "build"),
+            ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+            id="apple_silicon_cross_builds_for_linux_amd64",
+        ),
+        pytest.param(
+            "x86_64",
+            ("docker", "build"),
+            (),
+            id="amd64_host_uses_plain_docker_build",
+        ),
+    ],
+)
+def test_local_build_targets_linux_amd64(
+    deploy_project: DeployProject,
+    monkeypatch: pytest.MonkeyPatch,
+    machine: str,
+    expected_command: tuple[str, ...],
+    expected_flags: tuple[str, ...],
+) -> None:
+    monkeypatch.setattr(deploy_module.platform, "machine", lambda: machine)
+
+    deploy_project.run("--no-remote")
+
+    build = deploy_project.docker.builds[0]
+    assert build["tag"].startswith("langgraph-deploy-tmp:")
+    assert (build["docker_command"], build["extra_flags"]) == (
+        expected_command,
+        expected_flags,
+    )
+
+
+def test_local_deploy_logs_in_with_the_control_plane_push_token(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    login = deploy_project.docker.command("login")
+    assert login.args[:2] == ("docker", "--config")
+    assert login.args[3:] == (
+        "login",
+        "-u",
+        "oauth2accesstoken",
+        "--password-stdin",
+        "registry.example.com",
+    )
+    assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"
+
+
+def test_local_deploy_tags_the_build_into_the_token_registry(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    built_tag = deploy_project.docker.builds[0]["tag"]
+    assert deploy_project.docker.command("tag").args == (
+        "docker",
+        "tag",
+        built_tag,
+        PUSHED_IMAGE,
+    )
+    assert deploy_project.docker.command("push").args[-1] == PUSHED_IMAGE
+
+
+def test_local_deploy_records_the_pushed_digest_and_tracked_packages(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    assert deploy_project.control_plane.bodies[_patch(CREATED_ID)] == {
+        "revision_source": "internal_docker",
+        "source_revision_config": {"image_uri": PUSHED_DIGEST},
+        "secrets": [],
+        "tracked_packages": TRACKED_PACKAGES,
+    }
+
+
+def test_status_link_points_at_the_langsmith_dashboard(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--no-remote")
+
+    assert (
+        "View status: https://smith.langchain.com/o/tenant-1/host/deployments/dep-created"
+        in result.output
+    )
+
+
+def test_prebuilt_image_is_validated_and_pushed_without_a_build(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--image", "local/app:dev")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.builds == []
+    assert deploy_project.docker.verbs() == [
+        "docker inspect-platform",
+        "docker login",
+        "docker tag",
+        "docker push",
+        "docker inspect-digest",
+    ]
+    assert deploy_project.docker.command("tag").args[2:] == (
+        "local/app:dev",
+        PUSHED_IMAGE,
+    )
+
+
+def test_push_is_retried_until_the_third_attempt(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.failing_pushes = 2
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.verbs().count("docker push") == 3
+
+
+def test_three_failed_pushes_abort_before_the_deployment_is_updated(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.failing_pushes = 3
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code != 0
+    assert _patch(CREATED_ID) not in deploy_project.timeline
+
+
+def test_existing_deployment_matched_by_exact_name_is_updated_not_created(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-other", "name": "my-app-2"},
+        {"id": "dep-existing", "name": "my-app"},
+    ]
+
+    deploy_project.run("--no-remote")
+
+    assert CREATE_DEPLOYMENT not in deploy_project.timeline
+    assert _patch("dep-existing") in deploy_project.timeline
+
+
+def test_deployment_not_created_by_the_cli_gets_an_actionable_error(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ui", "name": "my-app"}
+    ]
+    deploy_project.control_plane.push_token_status = 400
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code != 0
+    assert "was not created by 'langgraph deploy'" in result.output
+    assert "docker login" not in deploy_project.timeline
+
+
+def test_remote_build_creates_an_internal_source_deployment_and_uploads_the_archive(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--remote", "--install-command", "yarn install")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        LIST_DEPLOYMENTS,
+        CREATE_DEPLOYMENT,
+        "create_archive",
+        _upload_url(CREATED_ID),
+        "upload_archive",
+        _patch(CREATED_ID),
+    ]
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source"] == (
+        "internal_source"
+    )
+    assert deploy_project.uploads == [(SIGNED_UPLOAD_URL, ARCHIVE[0], ARCHIVE[1])]
+    assert deploy_project.control_plane.bodies[_patch(CREATED_ID)] == {
+        "revision_source": "internal_source",
+        "source_revision_config": {
+            "source_tarball_path": OBJECT_PATH,
+            "langgraph_config_path": ARCHIVE[2],
+        },
+        "source_config": {"install_command": "yarn install"},
+        "secrets": [],
+        "tracked_packages": TRACKED_PACKAGES,
+    }
+    assert "Build triggered" in result.output
+
+
+def test_push_to_builds_pushes_then_creates_an_external_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        LIST_DEPLOYMENTS,
+        "docker build",
+        "docker push",
+        "docker inspect-digest",
+        CREATE_DEPLOYMENT,
+    ]
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT] == {
+        "name": "my-app",
+        "source": "external_docker",
+        "source_config": {"resource_spec": {}},
+        "source_revision_config": {"image_uri": EXTERNAL_DIGEST},
+        "secrets": [],
+    }
+    assert "Deployment created" in result.output
+
+
+def test_push_to_builds_directly_with_the_push_reference(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.builds[0]["tag"] == EXTERNAL_IMAGE
+    assert deploy_project.docker.command("push").args == (
+        "docker",
+        "push",
+        EXTERNAL_IMAGE,
+    )
+
+
+def test_push_to_composes_with_the_tag_flag(deploy_project: DeployProject) -> None:
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY, "--tag", "v1")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.command("push").args[-1] == f"{PUSH_REPOSITORY}:v1"
+
+
+def test_push_to_with_a_failing_push_creates_no_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.failing_pushes = 3
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code != 0
+    assert CREATE_DEPLOYMENT not in deploy_project.timeline
+
+
+def test_verbose_never_echoes_the_push_token(deploy_project: DeployProject) -> None:
+    result = deploy_project.run("--no-remote", "--verbose")
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.command("login").kwargs["verbose"] is False
+
+
+def test_push_to_retags_a_prebuilt_image_instead_of_building(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--image", "local/app:dev", "--push-to", PUSH_REPOSITORY
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.docker.builds == []
+    assert deploy_project.docker.verbs() == [
+        "docker inspect-platform",
+        "docker tag",
+        "docker push",
+        "docker inspect-digest",
+    ]
+    assert deploy_project.docker.command("tag").args == (
+        "docker",
+        "tag",
+        "local/app:dev",
+        EXTERNAL_IMAGE,
+    )
+
+
+def test_push_to_updates_an_existing_external_deployment_with_the_new_image(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        LIST_DEPLOYMENTS,
+        "docker build",
+        "docker push",
+        "docker inspect-digest",
+        _patch("dep-ext"),
+    ]
+    assert deploy_project.control_plane.bodies[_patch("dep-ext")] == {
+        "source_revision_config": {"image_uri": EXTERNAL_DIGEST},
+        "secrets": [],
+        "tracked_packages": TRACKED_PACKAGES,
+    }
+
+
+def test_push_to_rejects_a_non_external_deployment_before_any_docker_work(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-cli", "name": "my-app", "source": "internal_docker"}
+    ]
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code != 0
+    assert "cannot be updated with --push-to" in result.output
+    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:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "another-name", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run(
+        "--deployment-id", "dep-ext", "--push-to", PUSH_REPOSITORY
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.timeline == [
+        _get("dep-ext"),
+        "docker build",
+        "docker push",
+        "docker inspect-digest",
+        _patch("dep-ext"),
+    ]
+
+
+def test_invalid_tag_fails_before_any_control_plane_call(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--no-remote", "--tag", "not a tag")
+
+    assert result.exit_code != 0
+    assert "Image tag may only contain" in result.output
+    assert deploy_project.timeline == []

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -13,6 +13,10 @@ import pytest
 
 import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
+    CustomerRegistrySource,
+    DockerBuildCommand,
+    ManagedRegistrySource,
+    RemoteBuildSource,
     _call_host_backend_with_optional_tenant,
     _create_host_backend_client,
     _docker_config_for_token,
@@ -21,12 +25,13 @@ 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,
     normalize_image_tag,
     normalize_name,
 )
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError
+from langgraph_cli.image_reference import ImageReference
 
 
 class TestDockerConfigForToken:
@@ -259,22 +264,18 @@ 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
 
@@ -334,7 +335,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(
@@ -543,60 +543,193 @@ 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,
+    }
+    REPOSITORY = "registry.example.com/app"
+
+    @pytest.mark.parametrize(
+        ("flags", "docker_available", "expected"),
+        [
+            pytest.param(
+                {"push_to": REPOSITORY},
+                True,
+                CustomerRegistrySource(
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
+                ),
+                id="push_to_selects_the_external_source_with_the_default_tag",
+            ),
+            pytest.param(
+                {"push_to": f"{REPOSITORY}:v2"},
+                True,
+                CustomerRegistrySource(
+                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
+                ),
+                id="push_to_keeps_a_tag_given_in_the_reference",
+            ),
+            pytest.param(
+                {"push_to": REPOSITORY, "tag": "v3"},
+                True,
+                CustomerRegistrySource(
+                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
+                ),
+                id="tag_flag_composes_with_push_to",
+            ),
+            pytest.param(
+                {"push_to": REPOSITORY, "image": "app:dev"},
+                False,
+                CustomerRegistrySource(
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
+                ),
+                id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
+            ),
+            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",
+            ),
+        ],
+    )
+    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:
@@ -647,6 +780,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)

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,33 @@ 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,
+    c = HostBackendClient(
+        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
     )
     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 +109,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 +120,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 +152,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 +170,379 @@ 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) == {"ok": True}
+    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

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")