diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index a6e8403e6..16439d549 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -8,10 +8,11 @@ import platform import re import tempfile import time -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime, timezone +from typing import Protocol, TypeVar import click import click.exceptions @@ -19,11 +20,18 @@ from dotenv import dotenv_values, set_key import langgraph_cli.config from langgraph_cli.analytics import log_command +from langgraph_cli.config import Config from langgraph_cli.constants import DEFAULT_CONFIG from langgraph_cli.dependency_tracking import find_tracked_packages from langgraph_cli.docker import build_docker_image, can_build_locally -from langgraph_cli.exec import Runner, subp_exec -from langgraph_cli.host_backend import HostBackendClient, HostBackendError +from langgraph_cli.exec import CommandRunner, Runner, subp_exec +from langgraph_cli.host_backend import ( + ControlPlaneEndpoints, + HostBackendClient, + HostBackendError, + SourceName, +) +from langgraph_cli.image_reference import ImageReference from langgraph_cli.progress import Progress from langgraph_cli.util import warn_non_wolfi_distro @@ -84,7 +92,25 @@ _API_KEY_ENV_NAMES = ( "LANGCHAIN_API_KEY", ) +_T = TypeVar("_T") + _DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME" +_DEFAULT_IMAGE_TAG = "latest" +_DEPLOYMENT_PLATFORM = "linux/amd64" +_NATIVE_AMD64_MACHINE = "x86_64" +_PUSH_ATTEMPTS = 3 +_LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp" +_OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {} +_LISTENER_REQUIRED_MARKER = "listener_id' is required" +_HYBRID_LISTENER_GUIDANCE = ( + "This workspace deploys through a listener in your own cluster, and the " + "control plane needs a listener ID to create a deployment. Create the " + "deployment once in the LangSmith UI, choosing the listener and namespace, " + "then re-run with --deployment-id ." +) + +_CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker" + _TERMINAL_STATUSES = frozenset( [ @@ -116,6 +142,25 @@ class BuildResult: show_build_logs_on_failure: bool = False +@dataclass(frozen=True, slots=True) +class ById: + deployment_id: str + + +@dataclass(frozen=True, slots=True) +class ByName: + name: str + + +DeploymentSelector = ById | ByName + + +@dataclass(frozen=True, slots=True) +class ExistingDeployment: + id: str + source: str | None + + # --------------------------------------------------------------------------- # Structured output emitter # --------------------------------------------------------------------------- @@ -287,12 +332,14 @@ def _get_emitter() -> _Emitter: # --------------------------------------------------------------------------- -def validate_deployment_selector(deployment_id: str | None, name: str | None) -> None: - """Ensure either deployment_id or name is provided.""" +def deployment_selector( + deployment_id: str | None, name: str | None +) -> DeploymentSelector: if deployment_id: - return - if not name: - raise click.UsageError("Either --deployment-id or --name is required.") + return ById(deployment_id) + if name: + return ByName(name) + raise click.UsageError("Either --deployment-id or --name is required.") def validate_deploy_commands( @@ -318,19 +365,25 @@ def validate_deploy_commands( # --------------------------------------------------------------------------- -def find_deployment_id_by_name( - client: HostBackendClient, name: str | None -) -> str | None: - """Return deployment ID for an exact name match, or None if not found.""" - if not name: +def _source_of(resource: object) -> str | None: + if not isinstance(resource, dict): return None - existing = client.list_deployments(name_contains=name) - if isinstance(existing, dict): - for dep in existing.get("resources", []): - if isinstance(dep, dict) and dep.get("name") == name: - found_id = dep.get("id") - if found_id: - return str(found_id) + source = resource.get("source") + return source if isinstance(source, str) else None + + +def find_deployment_by_name( + client: HostBackendClient, name: str +) -> ExistingDeployment | None: + listed = client.list_deployments(name_contains=name) + resources = listed.get("resources", []) if isinstance(listed, dict) else [] + for resource in resources: + if ( + isinstance(resource, dict) + and resource.get("name") == name + and resource.get("id") + ): + return ExistingDeployment(str(resource["id"]), _source_of(resource)) return None @@ -358,7 +411,7 @@ def normalize_image_tag(value: str) -> str: Tags may only contain [A-Za-z0-9_.-]. Defaults to "latest" when empty. """ if not value: - value = "latest" + value = _DEFAULT_IMAGE_TAG if not re.fullmatch(r"[A-Za-z0-9_.-]+", value): raise click.UsageError( "Image tag may only contain characters A-Z, a-z, 0-9, '_', '-', '.'" @@ -366,7 +419,9 @@ def normalize_image_tag(value: str) -> str: return value -def _validate_prebuilt_image(runner, image: str, *, verbose: bool) -> None: +def _validate_prebuilt_image( + runner: CommandRunner, image: str, *, verbose: bool +) -> None: """Ensure a prebuilt image exists locally for linux/amd64.""" try: stdout, _ = runner.run( @@ -393,13 +448,14 @@ def _validate_prebuilt_image(runner, image: str, *, verbose: bool) -> None: ) from None image_platform = (stdout or "").strip() - if image_platform != "linux/amd64": + if image_platform != _DEPLOYMENT_PLATFORM: detected = image_platform or "unknown" raise click.ClickException( f"Docker image '{image}' targets {detected}, but LangSmith Deployment " - "requires linux/amd64. Rebuild or pull the image for linux/amd64 before " - "deploying with --image." + f"requires {_DEPLOYMENT_PLATFORM}. Rebuild or pull the image for " + f"{_DEPLOYMENT_PLATFORM} before deploying with --image." ) + _get_emitter().info(f"Image is available for {_DEPLOYMENT_PLATFORM}") def _extract_deployment_url(deployment: dict[str, object]) -> str: @@ -599,35 +655,39 @@ def _log_deploy_step(step: int, message: str, **extra: object) -> None: _get_emitter().step(step, message, **extra) -def _resolve_deployment( +def _fetch_deployment( + client: HostBackendClient, step: int, selector: ById +) -> tuple[ExistingDeployment, int]: + _log_deploy_step(step, f"Using deployment {selector.deployment_id}") + resource = _call_host_backend_with_optional_tenant( + client, lambda c: c.get_deployment(selector.deployment_id) + ) + return ExistingDeployment(selector.deployment_id, _source_of(resource)), step + 1 + + +def _find_deployment( client: HostBackendClient, step: int, - deployment_id: str | None, - name: str | None, + selector: ByName, *, not_found_message: str, -) -> tuple[str | None, bool, int]: - """Resolve an existing deployment by ID or exact name match.""" - needs_creation = False - if deployment_id: - _log_deploy_step(step, f"Using deployment {deployment_id}") - _call_host_backend_with_optional_tenant( - client, lambda c: c.get_deployment(deployment_id) - ) - return deployment_id, needs_creation, step + 1 - - _log_deploy_step(step, f"Looking up deployment '{name}'") - found_id = _call_host_backend_with_optional_tenant( - client, lambda c: find_deployment_id_by_name(c, name) +) -> tuple[ExistingDeployment | None, int]: + _log_deploy_step(step, f"Looking up deployment '{selector.name}'") + found = _call_host_backend_with_optional_tenant( + client, lambda c: find_deployment_by_name(c, selector.name) ) em = _get_emitter() - if found_id: - deployment_id = str(found_id) - em.info(f"Found existing deployment (ID: {deployment_id})") - else: - needs_creation = True + if found is None: em.warn(not_found_message) - return deployment_id, needs_creation, step + 1 + else: + em.info(f"Found existing deployment (ID: {found.id})") + return found, step + 1 + + +@dataclass(frozen=True, slots=True) +class CreatedDeployment: + id: str + resource: dict[str, object] def _create_deployment( @@ -635,18 +695,17 @@ def _create_deployment( step: int, *, name: str, - deployment_type: str, source: str, - config_rel: str | None = None, - secrets: list[dict[str, str]] | None = None, -) -> tuple[str, int]: - """Create a deployment and return its ID and next step number.""" + source_config: dict[str, object], + source_revision_config: dict[str, object], + secrets: list[dict[str, str]], +) -> tuple[CreatedDeployment, int]: _log_deploy_step(step, f"Creating deployment '{name}'") created = client.create_deployment( name=name, - deployment_type=deployment_type, source=source, - config_path=config_rel, + source_config=source_config, + source_revision_config=source_revision_config, secrets=secrets, ) created_id = created.get("id") if isinstance(created, dict) else None @@ -655,43 +714,22 @@ def _create_deployment( "POST /v2/deployments succeeded but response missing a valid 'id'" ) _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id) - return created_id, step + 1 - - -def _smith_dashboard_base_url(host_url: str | None) -> str: - """Derive the LangSmith dashboard base URL from the API host URL.""" - from urllib.parse import urlparse - - if not host_url: - return "https://smith.langchain.com" - parsed = urlparse(host_url) - hostname = parsed.hostname or "" - if hostname in ("localhost", "127.0.0.1"): - return host_url.rstrip("/") - - api_host_suffix = "api.host.langchain.com" - if hostname == api_host_suffix: - return "https://smith.langchain.com" - if hostname.endswith(f".{api_host_suffix}"): - prefix = hostname[: -(len(api_host_suffix) + 1)] - return f"https://{prefix}.smith.langchain.com" - - return "https://smith.langchain.com" + return CreatedDeployment(created_id, created), step + 1 def _get_deployment_status_url( - updated: object, deployment_id: str, host_url: str | None = None + updated: object, deployment_id: str, host_url: str ) -> str | None: """Compute the LangSmith dashboard URL for a deployment, if possible.""" tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None if not tenant_id: return None - base = _smith_dashboard_base_url(host_url) + base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}" def _emit_deployment_status_url( - updated: object, deployment_id: str, host_url: str | None = None + updated: object, deployment_id: str, host_url: str ) -> str | None: """Emit the deployment status URL and return it.""" url = _get_deployment_status_url(updated, deployment_id, host_url) @@ -893,7 +931,7 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None: def _resolve_pushed_image_digest( - runner, + runner: CommandRunner, *, remote_image: str, docker_config_dir: str | None, @@ -906,19 +944,25 @@ def _resolve_pushed_image_digest( Falls back to ``remote_image`` with a warning if no matching digest is found, rather than failing the deploy. """ - # rsplit preserves ``:port`` in the registry host. - repo_no_tag = remote_image.rsplit(":", 1)[0] - args: list[str] = ["docker"] - if docker_config_dir: - args += ["--config", docker_config_dir] - args += ["image", "inspect", "--format", "{{json .RepoDigests}}", remote_image] - stdout, _ = runner.run(subp_exec(*args, collect=True, verbose=verbose)) + reference = ImageReference.parse(remote_image) + stdout, _ = runner.run( + subp_exec( + *_docker_argv(docker_config_dir), + "image", + "inspect", + "--format", + "{{json .RepoDigests}}", + remote_image, + collect=True, + verbose=verbose, + ) + ) try: digests = json_mod.loads(stdout or "[]") or [] except json_mod.JSONDecodeError: digests = [] for d in digests: - if isinstance(d, str) and d.startswith(f"{repo_no_tag}@sha256:"): + if isinstance(d, str) and reference.matches_digest(d): return d _get_emitter().warn( f"Could not resolve image digest for {remote_image}; " @@ -927,86 +971,123 @@ def _resolve_pushed_image_digest( return remote_image +@dataclass(frozen=True, slots=True) +class BuildSpec: + config: pathlib.Path + config_json: Config + base_image: str | None + api_version: str | None + pull: bool + docker_build_args: Sequence[str] + install_command: str | None + build_command: str | None + + +@dataclass(frozen=True, slots=True) +class DockerBuildCommand: + command: tuple[str, ...] + flags: tuple[str, ...] + + @classmethod + def for_host(cls, machine: str, *, verbose: bool) -> "DockerBuildCommand": + if machine == _NATIVE_AMD64_MACHINE: + return cls(("docker", "build"), ()) + flags: tuple[str, ...] = ("--platform", _DEPLOYMENT_PLATFORM, "--load") + if not verbose: + flags += ("--progress=quiet",) + return cls(("docker", "buildx", "build"), flags) + + +def _docker_argv(docker_config_dir: str | None) -> tuple[str, ...]: + if docker_config_dir is None: + return ("docker",) + return ("docker", "--config", docker_config_dir) + + +def _build_image( + runner: CommandRunner, spec: BuildSpec, tag: str, *, verbose: bool +) -> None: + build = DockerBuildCommand.for_host(platform.machine(), verbose=verbose) + with Progress(message="Building...", elapsed=not verbose): + build_docker_image( + runner, + lambda _msg: None, + spec.config, + spec.config_json, + spec.base_image, + spec.api_version, + spec.pull, + tag, + spec.docker_build_args, + spec.install_command, + spec.build_command, + docker_command=build.command, + extra_flags=build.flags, + verbose=verbose, + ) + + +def _push_image( + runner: CommandRunner, + image: str, + *, + docker_config_dir: str | None, + verbose: bool, +) -> None: + for attempt in range(1, _PUSH_ATTEMPTS + 1): + try: + with Progress(message="Pushing...", elapsed=not verbose): + runner.run( + subp_exec( + *_docker_argv(docker_config_dir), "push", image, verbose=verbose + ) + ) + return + except click.exceptions.Exit: + if attempt == _PUSH_ATTEMPTS: + raise + _get_emitter().warn( + f" Push failed, retrying (attempt {attempt + 1} of {_PUSH_ATTEMPTS})..." + ) + + +def _image_revision_result(resource: object, no_result_message: str) -> BuildResult: + return BuildResult( + updated=resource if isinstance(resource, dict) else {}, + progress_message="Deploying...", + timeout_seconds=300, + poll_interval_seconds=1, + no_result_message=no_result_message, + ) + + def _run_local_build( *, client: HostBackendClient, deployment_id: str, step: int, - config: pathlib.Path, - config_json: dict, + spec: BuildSpec, verbose: bool, - pull: bool, - api_version: str | None, - base_image: str | None, image_name: str | None, prebuilt_image: str | None, name: str | None, tag: str, - install_command: str | None, - build_command: str | None, - docker_build_args: Sequence[str], secrets: list[dict[str, str]], tracked_packages: list[str] | None, ) -> BuildResult: """Build locally with Docker, push to registry, update deployment.""" - # Use buildx to cross-compile for amd64 when running on a non-x86_64 host - # (e.g. Apple Silicon). On amd64 hosts, plain docker build is sufficient. - needs_buildx = platform.machine() != "x86_64" - local_tag = f"langgraph-deploy-tmp:{int(time.time())}" + local_tag = f"{_LOCAL_BUILD_TAG_PREFIX}:{int(time.time())}" image_to_push = prebuilt_image or local_tag with Runner() as runner: if prebuilt_image: _log_deploy_step(step, f"Validating image {prebuilt_image}") _validate_prebuilt_image(runner, prebuilt_image, verbose=verbose) - click.secho(" Image is available for linux/amd64", fg="green") else: - # -- Step: Build image -- _log_deploy_step(step, "Building image") - if needs_buildx: - build_flags: list[str] = [ - "--platform", - "linux/amd64", - "--load", - ] - if not verbose: - build_flags.append("--progress=quiet") - with Progress(message="Building...", elapsed=not verbose): - build_docker_image( - runner, - lambda _msg: None, - config, - config_json, - base_image, - api_version, - pull, - local_tag, - docker_build_args, - install_command, - build_command, - docker_command=("docker", "buildx", "build"), - extra_flags=build_flags, - verbose=verbose, - ) - else: - with Progress(message="Building...", elapsed=not verbose): - build_docker_image( - runner, - lambda _msg: None, - config, - config_json, - base_image, - api_version, - pull, - local_tag, - docker_build_args, - install_command, - build_command, - verbose=verbose, - ) + _build_image(runner, spec, local_tag, verbose=verbose) step += 1 - # -- Step: Get push token and authenticate -- _log_deploy_step(step, "Requesting push token") try: push_data = client.request_push_token(deployment_id) @@ -1034,15 +1115,15 @@ def _run_local_build( normalized_registry = registry_url.rstrip("/") if "://" in normalized_registry: normalized_registry = normalized_registry.split("//", 1)[1] - repo_seed = image_name or name or config.parent.name - repo_name = normalize_name(repo_seed) - tag_value = normalize_image_tag(tag) - remote_image = f"{normalized_registry}/{repo_name}:{tag_value}" - + repo_seed = image_name or name or spec.config.parent.name + remote_image = str( + ImageReference( + f"{normalized_registry}/{normalize_name(repo_seed)}", + tag, + ) + ) registry_host = normalized_registry.split("/")[0] - # Use a clean Docker config with only the push token so that - # system credential helpers (e.g. gcloud) don't interfere. with _docker_config_for_token(registry_host, deployment_token) as cfg: _log_deploy_step(step, f"Logging into {registry_host}") token_input = ( @@ -1052,53 +1133,23 @@ def _run_local_build( ) runner.run( subp_exec( - "docker", - "--config", - cfg, + *_docker_argv(cfg), "login", "-u", "oauth2accesstoken", "--password-stdin", registry_host, input=token_input, - verbose=verbose, + verbose=False, ) ) step += 1 - # -- Step: Tag and push -- _log_deploy_step(step, f"Pushing image {remote_image}") runner.run( - subp_exec( - "docker", - "tag", - image_to_push, - remote_image, - verbose=verbose, - ) + subp_exec("docker", "tag", image_to_push, remote_image, verbose=verbose) ) - max_push_retries = 3 - for attempt in range(max_push_retries): - try: - with Progress(message="Pushing...", elapsed=not verbose): - runner.run( - subp_exec( - "docker", - "--config", - cfg, - "push", - remote_image, - verbose=verbose, - ) - ) - break - except click.exceptions.Exit: - if attempt < max_push_retries - 1: - _get_emitter().warn( - f" Push failed, retrying (attempt {attempt + 2} of {max_push_retries})..." - ) - else: - raise + _push_image(runner, remote_image, docker_config_dir=cfg, verbose=verbose) step += 1 resolved_image = _resolve_pushed_image_digest( @@ -1108,22 +1159,16 @@ def _run_local_build( verbose=verbose, ) - # -- Step: Update deployment -- _log_deploy_step(step, f"Updating deployment {deployment_id}") updated = client.update_deployment( deployment_id, resolved_image, + revision_source="internal_docker", secrets=secrets, tracked_packages=tracked_packages, ) - return BuildResult( - updated=updated if isinstance(updated, dict) else {}, - progress_message="Deploying...", - timeout_seconds=300, - poll_interval_seconds=1, - no_result_message="Deployment updated", - ) + return _image_revision_result(updated, "Deployment updated") def _run_remote_build( @@ -1131,11 +1176,8 @@ def _run_remote_build( client: HostBackendClient, deployment_id: str, step: int, - config: pathlib.Path, - config_json: dict, + spec: BuildSpec, verbose: bool, - install_command: str | None, - build_command: str | None, secrets: list[dict[str, str]], tracked_packages: list[str] | None, ) -> BuildResult: @@ -1144,7 +1186,11 @@ def _run_remote_build( em = _get_emitter() _log_deploy_step(step, "Creating source archive") - with create_archive(config, config_json) as (archive_path, file_size, config_rel): + with create_archive(spec.config, spec.config_json) as ( + archive_path, + file_size, + config_rel, + ): em.info(f"Archive created ({file_size / _BYTES_PER_MIB:.1f} MB)") step += 1 @@ -1166,8 +1212,8 @@ def _run_remote_build( source_tarball_path=object_path, config_path=config_rel, secrets=secrets, - install_command=install_command, - build_command=build_command, + install_command=spec.install_command, + build_command=spec.build_command, tracked_packages=tracked_packages, ) @@ -1224,6 +1270,251 @@ def _run_remote_build( ) +# --------------------------------------------------------------------------- +# Deployment sources +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class DeployContext: + client: HostBackendClient + spec: BuildSpec + verbose: bool + selector: DeploymentSelector + deployment_type: str + secrets: list[dict[str, str]] + tracked_packages: list[str] | None + + +@dataclass(frozen=True, slots=True) +class DeployOutcome: + deployment_id: str + build_result: BuildResult + + +class DeploymentSource(Protocol): + def run(self, ctx: DeployContext) -> DeployOutcome: ... + + +def _resolve_or_create( + ctx: DeployContext, *, source: SourceName, not_found_message: str +) -> tuple[str, int]: + if isinstance(ctx.selector, ById): + existing, step = _fetch_deployment(ctx.client, 1, ctx.selector) + return existing.id, step + found, step = _find_deployment( + ctx.client, 1, ctx.selector, not_found_message=not_found_message + ) + if found is not None: + return found.id, step + created, step = _create_deployment( + ctx.client, + step, + name=ctx.selector.name, + source=source, + source_config={"deployment_type": ctx.deployment_type}, + source_revision_config={}, + secrets=ctx.secrets, + ) + return created.id, step + + +def _ensure_customer_registry_source(existing: ExistingDeployment) -> None: + if existing.source != _CUSTOMER_REGISTRY_SOURCE: + raise click.UsageError( + f"Deployment {existing.id} was not created from an external image " + "and cannot be updated with --push-to. Run without --push-to to keep " + "its current build mode, or use a different --name to create a new " + "deployment." + ) + + +@dataclass(frozen=True, slots=True) +class ManagedRegistrySource: + prebuilt_image: str | None + image_name: str | None + tag: str + + def run(self, ctx: DeployContext) -> DeployOutcome: + deployment_id, step = _resolve_or_create( + ctx, + source="internal_docker", + not_found_message="No deployment found. Will create after build.", + ) + build_result = _run_local_build( + client=ctx.client, + deployment_id=deployment_id, + step=step, + spec=ctx.spec, + verbose=ctx.verbose, + image_name=self.image_name, + prebuilt_image=self.prebuilt_image, + name=ctx.selector.name if isinstance(ctx.selector, ByName) else None, + tag=self.tag, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome(deployment_id, build_result) + + +@dataclass(frozen=True, slots=True) +class RemoteBuildSource: + def run(self, ctx: DeployContext) -> DeployOutcome: + deployment_id, step = _resolve_or_create( + ctx, + source="internal_source", + not_found_message="No deployment found. Will create.", + ) + build_result = _run_remote_build( + client=ctx.client, + deployment_id=deployment_id, + step=step, + spec=ctx.spec, + verbose=ctx.verbose, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome(deployment_id, build_result) + + +@dataclass(frozen=True, slots=True) +class CustomerRegistrySource: + reference: ImageReference + prebuilt_image: str | None + + def run(self, ctx: DeployContext) -> DeployOutcome: + if isinstance(ctx.selector, ById): + existing, step = _fetch_deployment(ctx.client, 1, ctx.selector) + return self._update(ctx, existing, step) + found, step = _find_deployment( + ctx.client, + 1, + ctx.selector, + not_found_message="No deployment found. Will create after push.", + ) + if found is not None: + return self._update(ctx, found, step) + return self._create(ctx, ctx.selector.name, step) + + def _update( + self, ctx: DeployContext, existing: ExistingDeployment, step: int + ) -> DeployOutcome: + _ensure_customer_registry_source(existing) + image_uri, step = self._publish(ctx, step) + _log_deploy_step(step, f"Updating deployment {existing.id}") + updated = ctx.client.update_deployment( + existing.id, + image_uri, + revision_source=None, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome( + existing.id, _image_revision_result(updated, "Deployment updated") + ) + + def _create(self, ctx: DeployContext, name: str, step: int) -> DeployOutcome: + image_uri, step = self._publish(ctx, step) + try: + created, _ = _create_deployment( + ctx.client, + step, + name=name, + source=_CUSTOMER_REGISTRY_SOURCE, + source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC}, + source_revision_config={"image_uri": image_uri}, + secrets=ctx.secrets, + ) + except HostBackendError as err: + if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message: + raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None + raise + return DeployOutcome( + created.id, _image_revision_result(created.resource, "Deployment created") + ) + + def _publish(self, ctx: DeployContext, step: int) -> tuple[str, int]: + image = str(self.reference) + with Runner() as runner: + if self.prebuilt_image: + _log_deploy_step(step, f"Validating image {self.prebuilt_image}") + _validate_prebuilt_image( + runner, self.prebuilt_image, verbose=ctx.verbose + ) + runner.run( + subp_exec( + "docker", "tag", self.prebuilt_image, image, verbose=ctx.verbose + ) + ) + else: + _log_deploy_step(step, f"Building image {image}") + _build_image(runner, ctx.spec, image, verbose=ctx.verbose) + step += 1 + _log_deploy_step(step, f"Pushing image {image}") + _push_image(runner, image, docker_config_dir=None, verbose=ctx.verbose) + step += 1 + digest = _resolve_pushed_image_digest( + runner, remote_image=image, docker_config_dir=None, verbose=ctx.verbose + ) + return digest, step + + +def _require_local_docker() -> None: + supported, error = can_build_locally() + if not supported: + raise click.UsageError(error or "Unable to build locally.") + + +def _push_reference(push_to: str, tag: str | None) -> ImageReference: + try: + reference = ImageReference.parse(push_to) + except ValueError: + raise click.UsageError( + "--push-to takes a repository with an optional tag, not a digest." + ) from None + if reference.tag is not None and tag is not None: + raise click.UsageError( + "--push-to already includes a tag; do not combine it with --tag." + ) + if reference.tag is not None: + return reference + return reference.with_tag(normalize_image_tag(tag or _DEFAULT_IMAGE_TAG)) + + +def _select_source( + *, + push_to: str | None, + image: str | None, + image_name: str | None, + tag: str | None, + remote_build_flag: bool | None, +) -> DeploymentSource: + if push_to is not None: + if remote_build_flag is True: + raise click.UsageError("--push-to cannot be combined with --remote.") + reference = _push_reference(push_to, tag) + if image is None: + _require_local_docker() + return CustomerRegistrySource(reference, prebuilt_image=image) + if image and remote_build_flag is True: + raise click.UsageError("--image cannot be combined with --remote builds.") + use_remote_build, local_build_error = _resolve_build_mode( + remote_build_flag, force_local=image is not None + ) + if not use_remote_build: + return ManagedRegistrySource( + prebuilt_image=image, + image_name=image_name, + tag=normalize_image_tag(tag or _DEFAULT_IMAGE_TAG), + ) + if remote_build_flag is None and local_build_error: + em = _get_emitter() + em.note(f"{local_build_error}\nUsing remote build instead.") + if not em.json_mode: + click.echo() + return RemoteBuildSource() + + # --------------------------------------------------------------------------- # Host backend client factory # --------------------------------------------------------------------------- @@ -1261,13 +1552,19 @@ def _create_host_backend_client( tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get( "LANGSMITH_TENANT_ID" ) - return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id) + langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get( + "LANGSMITH_ENDPOINT" + ) + endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint) + return HostBackendClient( + endpoints.control_plane_url, resolved_api_key, tenant_id=tenant_id + ) def _call_host_backend_with_optional_tenant( client: HostBackendClient, - operation: Callable[[HostBackendClient], object], -) -> object: + operation: Callable[[HostBackendClient], _T], +) -> _T: """Run *operation*, prompting for a workspace ID on org-scoped 403s. On success the original *client* is returned as-is. If the user is @@ -1300,11 +1597,13 @@ def _call_host_backend_with_optional_tenant( "Find your workspace ID in LangSmith under Settings > Workspaces.", fg="yellow", ) - client._client.headers["X-Tenant-ID"] = click.prompt("Workspace ID") + client.set_tenant(click.prompt("Workspace ID")) prompted_for_tenant = True continue if err.status_code == 403 and "not enabled" in err.message.lower(): - smith_base = _smith_dashboard_base_url(client._base_url) + smith_base = ControlPlaneEndpoints.from_control_plane_url( + client.base_url + ).dashboard_url raise HostBackendError( "LangSmith Deployment is not enabled for this organization. " f"Enable it at {smith_base}/host/deployments" @@ -1340,7 +1639,7 @@ OPT_HOST_DEPLOYMENT_NAME = click.option( OPT_HOST_URL = click.option( "--host-url", envvar="LANGGRAPH_HOST_URL", - default="https://api.host.langchain.com", + default=None, hidden=True, ) @@ -1454,7 +1753,10 @@ def _deploy_base_options( type=click.Choice(["dev", "prod"]), default="dev", show_default=True, - help="Deployment type (used when creating a new deployment).", + help=( + "Deployment type (used when creating a new deployment). " + "Ignored with --push-to." + ), ), click.option( "--no-wait", @@ -1468,9 +1770,8 @@ def _deploy_base_options( click.option( "--tag", "-t", - default="latest", - show_default=True, - help="Tag to use for the pushed deployment image.", + default=None, + help="Tag to use for the pushed deployment image. [default: latest]", ), click.option( "--image", @@ -1479,6 +1780,16 @@ def _deploy_base_options( "skip building. The image must target linux/amd64." ), ), + click.option( + "--push-to", + help=( + "Push the image to this repository in a registry you manage, " + "then deploy it from there. For self-hosted and hybrid " + "LangSmith. Uses your existing Docker credentials. Builds the " + "project, or retags the local image given with --image. " + "Give the tag here or with --tag (default: latest)." + ), + ), click.option( "--config", "-c", @@ -1580,7 +1891,8 @@ def _deploy_cmd( name: str | None, image_name: str | None, image: str | None, - tag: str, + push_to: str | None, + tag: str | None, base_image: str | None, install_command: str | None, build_command: str | None, @@ -1601,7 +1913,6 @@ def _deploy_cmd( if not json_output: click.echo() - # -- 1. Preflight -- validate_deploy_commands(install_command, build_command) if not config.exists(): message = ( @@ -1636,117 +1947,66 @@ def _deploy_cmd( secrets = _secrets_from_env(_env_without_deployment_name(env_vars)) - if image and remote_build_flag is True: - raise click.UsageError("--image cannot be combined with --remote builds.") - - use_remote_build, local_build_error = _resolve_build_mode( - remote_build_flag, force_local=image is not None + source = _select_source( + push_to=push_to, + image=image, + image_name=image_name, + tag=tag, + remote_build_flag=remote_build_flag, ) - if use_remote_build and remote_build_flag is None and local_build_error: - em.note(f"{local_build_error}\nUsing remote build instead.") - if not json_output: - click.echo() - # -- 2. Resolve / create deployment -- client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) - step = 1 - - deployment_id, needs_creation, step = _resolve_deployment( - client, - step, - deployment_id, - name, - not_found_message=( - "No deployment found. Will create." - if use_remote_build - else "No deployment found. Will create after build." - ), - ) - - if needs_creation: - deployment_id, step = _create_deployment( - client, - step, - name=name, - deployment_type=deployment_type, - source="internal_source" if use_remote_build else "internal_docker", - secrets=secrets, - ) - - if not deployment_id: - raise click.ClickException("Failed to determine deployment ID") - - # Scan local sources for tracked packages so the new revision carries - # the same metadata GitHub-backed deploys produce. Failures must never - # block a deploy. try: tracked_packages = find_tracked_packages(config, config_json) or None except Exception as exc: em.warn(f"Skipped tracked-package scan: {exc}") tracked_packages = None - # -- 3. Build (divergent path) -- - if use_remote_build: - build_result = _run_remote_build( + outcome = source.run( + DeployContext( client=client, - deployment_id=deployment_id, - step=step, - config=config, - config_json=config_json, + spec=BuildSpec( + config=config, + config_json=config_json, + base_image=base_image, + api_version=api_version, + pull=pull, + docker_build_args=docker_build_args, + install_command=install_command, + build_command=build_command, + ), verbose=verbose, - install_command=install_command, - build_command=build_command, + selector=deployment_selector(deployment_id, name), + deployment_type=deployment_type, secrets=secrets, tracked_packages=tracked_packages, ) - else: - build_result = _run_local_build( - client=client, - deployment_id=deployment_id, - step=step, - config=config, - config_json=config_json, - verbose=verbose, - pull=pull, - api_version=api_version, - base_image=base_image, - image_name=image_name, - prebuilt_image=image, - name=name, - tag=tag, - install_command=install_command, - build_command=build_command, - docker_build_args=docker_build_args, - secrets=secrets, - tracked_packages=tracked_packages, - ) - - # -- 4. Shared wait + result -- + ) dep_status_url = _emit_deployment_status_url( - build_result.updated, - deployment_id, - host_url, + outcome.build_result.updated, + outcome.deployment_id, + client.base_url, ) if no_wait: - em.info(build_result.no_result_message) + em.info(outcome.build_result.no_result_message) return last_status, revision_id = _poll_revision_status( client, - deployment_id, - progress_message=build_result.progress_message, - timeout_seconds=build_result.timeout_seconds, - poll_interval_seconds=build_result.poll_interval_seconds, - on_poll=build_result.on_poll, - on_interrupt=build_result.on_interrupt, + outcome.deployment_id, + progress_message=outcome.build_result.progress_message, + timeout_seconds=outcome.build_result.timeout_seconds, + poll_interval_seconds=outcome.build_result.poll_interval_seconds, + on_poll=outcome.build_result.on_poll, + on_interrupt=outcome.build_result.on_interrupt, ) if not last_status: - em.info(build_result.no_result_message) + em.info(outcome.build_result.no_result_message) return if ( - build_result.show_build_logs_on_failure + outcome.build_result.show_build_logs_on_failure and last_status == "BUILD_FAILED" and not verbose and revision_id is not None @@ -1754,7 +2014,7 @@ def _deploy_cmd( em.error("Last build log lines:") try: logs_resp = client.get_build_logs( - deployment_id, + outcome.deployment_id, revision_id, {"order": "desc", "limit": 30}, ) @@ -1770,7 +2030,7 @@ def _deploy_cmd( _print_deployment_result( client, - deployment_id, + outcome.deployment_id, last_status, dashboard_label="Deployment dashboard", status_url=dep_status_url, @@ -1981,22 +2241,23 @@ def deploy_logs( start_time: str | None, end_time: str | None, follow: bool, - host_url: str, + host_url: str | None, ): env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG) client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) if not deployment_id and not name: name = env_vars.get(_DEPLOYMENT_NAME_ENV) - validate_deployment_selector(deployment_id, name) - if deployment_id: - dep_id = deployment_id + selector = deployment_selector(deployment_id, name) + if isinstance(selector, ById): + dep_id = selector.deployment_id else: + name_to_find = selector.name found = _call_host_backend_with_optional_tenant( - client, lambda c: find_deployment_id_by_name(c, name) + client, lambda c: find_deployment_by_name(c, name_to_find) ) - if not found: - raise click.ClickException(f"Deployment '{name}' not found.") - dep_id = str(found) + if found is None: + raise click.ClickException(f"Deployment '{name_to_find}' not found.") + dep_id = found.id if log_type == "build" and not revision_id: revisions_resp = client.list_revisions(dep_id, limit=1) diff --git a/libs/cli/langgraph_cli/exec.py b/libs/cli/langgraph_cli/exec.py index 974fc75f3..32c3d3470 100644 --- a/libs/cli/langgraph_cli/exec.py +++ b/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(): diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index 4ab608675..ebfd3f9d0 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/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, diff --git a/libs/cli/langgraph_cli/image_reference.py b/libs/cli/langgraph_cli/image_reference.py new file mode 100644 index 000000000..803ea3d4c --- /dev/null +++ b/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}" diff --git a/libs/cli/tests/unit_tests/cli/test_deploy_command.py b/libs/cli/tests/unit_tests/cli/test_deploy_command.py new file mode 100644 index 000000000..fd9817633 --- /dev/null +++ b/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 == [] diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index 5112450f8..62c34625b 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/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( @@ -540,60 +540,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: @@ -644,6 +777,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) diff --git a/libs/cli/tests/unit_tests/test_host_backend.py b/libs/cli/tests/unit_tests/test_host_backend.py index 85c2a93d5..686b516c3 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/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 diff --git a/libs/cli/tests/unit_tests/test_image_reference.py b/libs/cli/tests/unit_tests/test_image_reference.py new file mode 100644 index 000000000..901555940 --- /dev/null +++ b/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")