feat(cli): log in with the registry username the control plane returns

langgraph deploy now tells the control plane it supports registry usernames
when it asks for a push token, logs in with the returned username (default
oauth2accesstoken), and pushes to repository_url exactly when the response
includes it. LangSmith AWS needs all three, since ECR rejects Google's
username and its repositories are single image names.

libs/cli/langgraph_cli/deploy.py

@@ -1055,15 +1055,28 @@ def _print_deployment_result(
 # ---------------------------------------------------------------------------
 
 
+def _strip_url_scheme(url: str) -> str:
+    """Return a registry reference without a trailing slash or URL scheme."""
+    reference = url.rstrip("/")
+    if "://" in reference:
+        reference = reference.split("//", 1)[1]
+    return reference
+
+
+# Artifact Registry's username for access tokens. Control planes whose registry
+# needs another one return it as `username`.
+_DEFAULT_REGISTRY_USERNAME = "oauth2accesstoken"
+
+
 @contextmanager
-def _docker_config_for_token(registry_host: str, token: str):
+def _docker_config_for_token(registry_host: str, token: str, username: str):
     """Create a temporary Docker config with only the push token.
 
     Yields the path to a temporary config directory that can be passed
     to `docker --config <path>` so that system credential helpers
     (e.g. gcloud) don't interfere with the push token.
     """
-    auth_b64 = base64.b64encode(f"oauth2accesstoken:{token}".encode()).decode()
+    auth_b64 = base64.b64encode(f"{username}:{token}".encode()).decode()
     config_data = {"auths": {registry_host: {"auth": auth_b64}}}
     with tempfile.TemporaryDirectory() as tmpdir:
         with open(os.path.join(tmpdir, "config.json"), "w") as f:
@@ -1317,21 +1330,22 @@ def _run_local_build(
             raise click.ClickException(
                 "Push token response missing token or registry_url"
             )
+        username = push_data.get("username") or _DEFAULT_REGISTRY_USERNAME
         step += 1
 
-        normalized_registry = registry_url.rstrip("/")
-        if "://" in normalized_registry:
-            normalized_registry = normalized_registry.split("//", 1)[1]
-        repo_seed = image_name or name or spec.config.parent.name
-        remote_image = str(
-            ImageReference(
-                f"{normalized_registry}/{normalize_name(repo_seed)}",
-                tag,
+        if push_data.get("repository_url"):
+            # The control plane chose the exact repository (an ECR repository
+            # is a single image name, so nothing may be appended to it).
+            repository = _strip_url_scheme(push_data["repository_url"])
+        else:
+            repo_seed = image_name or name or spec.config.parent.name
+            repository = (
+                f"{_strip_url_scheme(registry_url)}/{normalize_name(repo_seed)}"
             )
-        )
-        registry_host = normalized_registry.split("/")[0]
+        remote_image = str(ImageReference(repository, tag))
+        registry_host = repository.split("/")[0]
 
-        with _docker_config_for_token(registry_host, deployment_token) as cfg:
+        with _docker_config_for_token(registry_host, deployment_token, username) as cfg:
             _log_deploy_step(step, f"Logging into {registry_host}")
             token_input = (
                 deployment_token
@@ -1343,7 +1357,7 @@ def _run_local_build(
                     *_docker_argv(cfg),
                     "login",
                     "-u",
-                    "oauth2accesstoken",
+                    username,
                     "--password-stdin",
                     registry_host,
                     input=token_input,

libs/cli/langgraph_cli/host_backend.py

@@ -249,9 +249,12 @@ class HostBackendClient:
         return self._request("DELETE", f"/v2/deployments/{deployment_id}")
 
     def request_push_token(self, deployment_id: str) -> dict[str, Any]:
+        # Tells the control plane this client logs in with the returned
+        # `username` and pushes to `repository_url`. LangSmith AWS requires it.
         return self._request(
             "POST",
             f"/v2/deployments/{deployment_id}/push-token",
+            {"supports_registry_username": True},
         )
 
     def request_upload_url(self, deployment_id: str) -> dict[str, Any]:

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

@@ -93,6 +93,7 @@ class ControlPlaneDouble:
     timeline: list[str]
     existing_deployments: list[dict] = field(default_factory=list)
     push_token_status: int = 200
+    push_token_extra: dict = field(default_factory=dict)
     create_error: str | None = None
     listeners: list[dict] = field(default_factory=list)
     listeners_by_id: dict[str, dict] = field(default_factory=dict)
@@ -147,7 +148,12 @@ class ControlPlaneDouble:
             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}
+                200,
+                json={
+                    "token": PUSH_TOKEN,
+                    "registry_url": REGISTRY_URL,
+                    **self.push_token_extra,
+                },
             )
         if path.endswith("/upload-url"):
             return httpx.Response(
@@ -410,6 +416,48 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
     assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"
 
 
+def test_local_deploy_tells_the_control_plane_it_honors_registry_usernames(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.run("--no-remote")
+
+    assert deploy_project.control_plane.bodies[_push_token(CREATED_ID)] == {
+        "supports_registry_username": True
+    }
+
+
+ECR_REPOSITORY = "123456789012.dkr.ecr.us-east-2.amazonaws.com/lsd/agent-1"
+ECR_PUSH_TOKEN_FIELDS = {"username": "AWS", "repository_url": ECR_REPOSITORY}
+
+
+def test_local_deploy_logs_in_with_the_username_the_control_plane_returns(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
+
+    deploy_project.run("--no-remote")
+
+    login = deploy_project.docker.command("login")
+    assert login.args[3:] == (
+        "login",
+        "-u",
+        "AWS",
+        "--password-stdin",
+        "123456789012.dkr.ecr.us-east-2.amazonaws.com",
+    )
+
+
+def test_local_deploy_pushes_to_exactly_the_returned_repository(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
+
+    deploy_project.run("--no-remote")
+
+    tag = deploy_project.docker.command("tag")
+    assert tag.args[-1] == f"{ECR_REPOSITORY}:latest"
+
+
 def test_local_deploy_tags_the_build_into_the_token_registry(
     deploy_project: DeployProject,
 ) -> None:

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -46,7 +46,9 @@ from langgraph_cli.image_reference import ImageReference
 
 class TestDockerConfigForToken:
     def test_creates_config_json(self):
-        with _docker_config_for_token("us-docker.pkg.dev", "my-token") as cfg:
+        with _docker_config_for_token(
+            "us-docker.pkg.dev", "my-token", "oauth2accesstoken"
+        ) as cfg:
             config_path = os.path.join(cfg, "config.json")
             assert os.path.isfile(config_path)
             with open(config_path) as f:
@@ -54,13 +56,26 @@ class TestDockerConfigForToken:
             expected_auth = base64.b64encode(b"oauth2accesstoken:my-token").decode()
             assert data == {"auths": {"us-docker.pkg.dev": {"auth": expected_auth}}}
 
+    def test_uses_the_given_username(self):
+        with _docker_config_for_token(
+            "123456789012.dkr.ecr.us-east-2.amazonaws.com", "ecr-password", "AWS"
+        ) as cfg:
+            with open(os.path.join(cfg, "config.json")) as f:
+                data = json.load(f)
+        expected_auth = base64.b64encode(b"AWS:ecr-password").decode()
+        assert data["auths"]["123456789012.dkr.ecr.us-east-2.amazonaws.com"] == {
+            "auth": expected_auth
+        }
+
     def test_tempdir_cleaned_up(self):
-        with _docker_config_for_token("registry.example.com", "tok") as cfg:
+        with _docker_config_for_token(
+            "registry.example.com", "tok", "oauth2accesstoken"
+        ) as cfg:
             assert os.path.isdir(cfg)
         assert not os.path.exists(cfg)
 
     def test_different_registries(self):
-        with _docker_config_for_token("gcr.io", "token123") as cfg:
+        with _docker_config_for_token("gcr.io", "token123", "oauth2accesstoken") as cfg:
             with open(os.path.join(cfg, "config.json")) as f:
                 data = json.load(f)
             assert "gcr.io" in data["auths"]