release(cli): 0.4.35 (#9310)

Bumping the CLI to 0.4.35 to cut a release including #9305
(--image-pull-secret).

libs/cli/js-examples/yarn.lock

@@ -2829,11 +2829,11 @@ graceful-fs@^4.2.11:
   integrity sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==
 
 handlebars@^4.7.9:
-  version "4.7.9"
-  resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.9.tgz#6f139082ab58dc4e5a0e51efe7db5ae890d56a0f"
-  integrity sha512-4E71E0rpOaQuJR2A3xDZ+GM1HyWYv1clR58tC8emQNeQe3RH7MAzSbat+V0wG78LQBo6m6bzSG/L4pBuCsgnUQ==
+  version "4.7.10"
+  resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.10.tgz#12419ea1b41391b3ffcf003f2164dcbce090c417"
+  integrity sha512-P5VJMVM7qgBn6vjXMw8WG9uVI+ncf2pi72j4de4yz5ZULLj2RGqLYaKOYGsgyrViQ0tePOVlN1tDCCXXtFqXKg==
   dependencies:
-    minimist "^1.2.5"
+    minimist "^1.2.8"
     neo-async "^2.6.2"
     source-map "^0.6.1"
     wordwrap "^1.0.0"
@@ -3920,7 +3920,7 @@ minimatch@^9.0.4:
   dependencies:
     brace-expansion "^2.0.2"
 
-minimist@^1.2.0, minimist@^1.2.5, minimist@^1.2.6:
+minimist@^1.2.0, minimist@^1.2.6, minimist@^1.2.8:
   version "1.2.8"
   resolved "https://registry.yarnpkg.com/minimist/-/minimist-1.2.8.tgz#c1a464e7693302e082a075cee0c057741ac4772c"
   integrity sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.34"
+__version__ = "0.4.35"

libs/cli/langgraph_cli/deploy.py

@@ -231,15 +231,6 @@ class RequestedPlacement:
     def requested(self) -> bool:
         return self.listener_id is not None or self.k8s_namespace is not None
 
-    def ensure_not_requested(self, deployment_id: str) -> None:
-        if self.requested:
-            raise click.UsageError(
-                "Listener and namespace are fixed when a deployment is created. "
-                f"Deployment {deployment_id} already exists, so drop --listener-id "
-                "and --k8s-namespace, or create a new deployment with a different "
-                "--name."
-            )
-
     def on(self, listener: Listener) -> Placement:
         return OnListener(listener.id, self._namespace(listener))
 
@@ -1055,15 +1046,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:
@@ -1201,7 +1205,7 @@ class DockerBuildCommand:
             return cls(("docker", "build"), ())
         flags: tuple[str, ...] = ("--platform", _DEPLOYMENT_PLATFORM, "--load")
         if not verbose:
-            flags += ("--progress=quiet",)
+            flags += ("--progress=plain",)
         return cls(("docker", "buildx", "build"), flags)
 
 
@@ -1317,21 +1321,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 +1348,7 @@ def _run_local_build(
                     *_docker_argv(cfg),
                     "login",
                     "-u",
-                    "oauth2accesstoken",
+                    username,
                     "--password-stdin",
                     registry_host,
                     input=token_input,
@@ -1647,10 +1652,72 @@ class PublishedImage:
 ImageSource = BuildAndPush | PublishedImage
 
 
+def _resource_spec(image_pull_secrets: tuple[str, ...]) -> Mapping[str, object]:
+    if not image_pull_secrets:
+        return _OPERATOR_DEFAULT_RESOURCE_SPEC
+    return {
+        "image_pull_secrets": [{"name": name} for name in image_pull_secrets],
+    }
+
+
+# Mirrors the control plane's own LocalObjectReference.name validation (a
+# Kubernetes DNS subdomain), so a bad secret name is rejected here instead of
+# after a build and push.
+_K8S_SECRET_NAME_RE = re.compile(
+    r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+)
+
+
+@dataclass(frozen=True, slots=True)
+class CreateOnlySettings:
+    """Options for an external-image deployment that only apply when
+    creating one, never when updating an existing deployment."""
+
+    placement: RequestedPlacement
+    image_pull_secrets: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        for name in self.image_pull_secrets:
+            if len(name) > 253 or not _K8S_SECRET_NAME_RE.match(name):
+                raise click.UsageError(
+                    f"--image-pull-secret {name!r} is not a valid Kubernetes "
+                    "secret name (lowercase alphanumeric characters, '-', or "
+                    "'.' only, max 253 characters)."
+                )
+
+    @property
+    def requested(self) -> bool:
+        return self.placement.requested or bool(self.image_pull_secrets)
+
+    def ensure_creatable(self, push_to: str | None, image_uri: str | None) -> None:
+        if push_to is None and image_uri is None and self.requested:
+            raise click.UsageError(
+                "--listener-id, --k8s-namespace, and --image-pull-secret only "
+                "apply when creating a deployment with --push-to or --image-uri."
+            )
+
+    def ensure_not_existing(self, deployment_id: str) -> None:
+        if self.placement.requested:
+            raise click.UsageError(
+                "Listener and namespace are fixed when a deployment is "
+                f"created. Deployment {deployment_id} already exists, so "
+                "drop --listener-id and --k8s-namespace, or create a new "
+                "deployment with a different --name."
+            )
+        if self.image_pull_secrets:
+            raise click.UsageError(
+                "langgraph deploy does not update image pull secrets on an "
+                f"existing deployment. Deployment {deployment_id} already "
+                "exists, so drop --image-pull-secret and update resource_spec "
+                "through the API or the LangSmith UI instead, or create a "
+                "new deployment with a different --name."
+            )
+
+
 @dataclass(frozen=True, slots=True)
 class CustomerRegistrySource:
     image: ImageSource
-    requested_placement: RequestedPlacement
+    create_only: CreateOnlySettings
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1672,7 +1739,7 @@ class CustomerRegistrySource:
         self, ctx: DeployContext, existing: ExistingDeployment, step: int
     ) -> DeployOutcome:
         _ensure_customer_registry_source(existing)
-        self.requested_placement.ensure_not_requested(existing.id)
+        self.create_only.ensure_not_existing(existing.id)
         image_uri, step = self._publish(ctx, step)
         _log_deploy_step(step, f"Updating deployment {existing.id}")
         updated = ctx.client.update_deployment(
@@ -1687,7 +1754,7 @@ class CustomerRegistrySource:
         )
 
     def _resolve_placement(self, ctx: DeployContext) -> Placement:
-        requested = self.requested_placement
+        requested = self.create_only.placement
         if requested.listener_id is not None:
             return requested.on(_requested_listener(ctx.client, requested.listener_id))
         if not (ctx.endpoints.is_cloud or requested.requested):
@@ -1716,7 +1783,9 @@ class CustomerRegistrySource:
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
                 source_config={
-                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
+                    "resource_spec": _resource_spec(
+                        self.create_only.image_pull_secrets
+                    ),
                     **placement.source_config(),
                 },
                 source_revision_config={"image_uri": image_uri},
@@ -1810,19 +1879,13 @@ def _select_source(
     tag: str | None,
     remote_build_flag: bool | None,
     placement: RequestedPlacement,
+    image_pull_secrets: tuple[str, ...],
     selector: DeploymentSelector,
 ) -> DeploymentSource:
-    if push_to is None and image_uri is None and placement.requested:
-        raise click.UsageError(
-            "--listener-id and --k8s-namespace only apply when creating a "
-            "deployment with --push-to or --image-uri."
-        )
-    if placement.requested and isinstance(selector, ById):
-        raise click.UsageError(
-            "Listener and namespace are fixed when a deployment is created, so "
-            "they cannot be set for an existing --deployment-id. Drop them, or "
-            "create a new deployment with --name."
-        )
+    create_only = CreateOnlySettings(placement, image_pull_secrets)
+    create_only.ensure_creatable(push_to, image_uri)
+    if create_only.requested and isinstance(selector, ById):
+        create_only.ensure_not_existing(selector.deployment_id)
     if image_uri is not None:
         if push_to is not None:
             raise click.UsageError("--image-uri cannot be combined with --push-to.")
@@ -1834,7 +1897,7 @@ def _select_source(
             raise click.UsageError("--image-uri cannot be combined with --remote.")
         return CustomerRegistrySource(
             image=PublishedImage(_validate_image_uri(image_uri)),
-            requested_placement=placement,
+            create_only=create_only,
         )
     if push_to is not None:
         if remote_build_flag is True:
@@ -1844,7 +1907,7 @@ def _select_source(
             _require_local_docker()
         return CustomerRegistrySource(
             image=BuildAndPush(reference=reference, prebuilt_image=image),
-            requested_placement=placement,
+            create_only=create_only,
         )
     if image and remote_build_flag is True:
         raise click.UsageError("--image cannot be combined with --remote builds.")
@@ -2188,6 +2251,18 @@ def _deploy_base_options(
                     "creating a deployment with --push-to or --image-uri."
                 ),
             ),
+            click.option(
+                "--image-pull-secret",
+                "image_pull_secrets",
+                multiple=True,
+                help=(
+                    "Name of a Kubernetes secret (type kubernetes.io/"
+                    "dockerconfigjson) in the deployment's namespace to use for "
+                    "pulling the image from a private registry. Repeat to give "
+                    "more than one. Only used when creating a deployment with "
+                    "--push-to or --image-uri."
+                ),
+            ),
             click.option(
                 "--config",
                 "-c",
@@ -2301,6 +2376,7 @@ def _deploy_cmd(
     image_uri: str | None,
     listener_id: str | None,
     k8s_namespace: str | None,
+    image_pull_secrets: tuple[str, ...],
     tag: str | None,
     base_image: str | None,
     install_command: str | None,
@@ -2377,6 +2453,7 @@ def _deploy_cmd(
         tag=tag,
         remote_build_flag=remote_build_flag,
         placement=RequestedPlacement(listener_id, k8s_namespace),
+        image_pull_secrets=image_pull_secrets,
         selector=selector,
     )
 

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(
@@ -363,7 +369,7 @@ def test_first_local_deploy_creates_an_internal_docker_deployment(
         pytest.param(
             "arm64",
             ("docker", "buildx", "build"),
-            ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+            ("--platform", "linux/amd64", "--load", "--progress=plain"),
             id="apple_silicon_cross_builds_for_linux_amd64",
         ),
         pytest.param(
@@ -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:
@@ -918,6 +966,100 @@ def test_self_hosted_control_plane_places_when_asked(
     }
 
 
+def test_image_pull_secret_is_included_in_resource_spec_on_create(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--image-pull-secret",
+        "regcred",
+        "--image-pull-secret",
+        "other-secret",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {
+            "image_pull_secrets": [
+                {"name": "regcred"},
+                {"name": "other-secret"},
+            ],
+        },
+    }
+
+
+def test_image_pull_secret_works_with_image_uri_too(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
+    )
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "resource_spec": {"image_pull_secrets": [{"name": "regcred"}]},
+    }
+
+
+def test_image_pull_secret_without_push_to_or_image_uri_is_refused(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run("--image-pull-secret", "regcred")
+
+    assert result.exit_code != 0
+    assert "only apply when creating a deployment with --push-to" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_an_invalid_image_pull_secret_name_is_rejected_before_any_docker_work(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--push-to", PUSH_REPOSITORY, "--image-pull-secret", "Not_Valid!"
+    )
+
+    assert result.exit_code != 0
+    assert "not a valid Kubernetes" in result.output
+    assert deploy_project.docker.verbs() == []
+    assert deploy_project.timeline == []
+
+
+def test_image_pull_secret_cannot_update_an_existing_deployment(
+    deploy_project: DeployProject,
+) -> None:
+    result = deploy_project.run(
+        "--deployment-id",
+        "dep-1",
+        "--push-to",
+        PUSH_REPOSITORY,
+        "--image-pull-secret",
+        "regcred",
+    )
+
+    assert result.exit_code != 0
+    assert "--image-pull-secret" in result.output
+    assert "dep-1" in result.output
+    assert deploy_project.timeline == []
+
+
+def test_image_pull_secret_cannot_update_a_deployment_found_by_name(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.existing_deployments = [
+        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
+    ]
+
+    result = deploy_project.run(
+        "--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
+    )
+
+    assert result.exit_code != 0
+    assert "--image-pull-secret" in result.output
+    assert deploy_project.docker.verbs() == []
+    assert _patch("dep-ext") not in deploy_project.timeline
+
+
 def test_updating_a_deployment_never_looks_up_listeners(
     deploy_project: DeployProject,
 ) -> None:

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -16,6 +16,7 @@ from langgraph_cli.deploy import (
     BuildAndPush,
     ById,
     ByName,
+    CreateOnlySettings,
     CustomerRegistrySource,
     DockerBuildCommand,
     ExistingDeployment,
@@ -34,6 +35,7 @@ from langgraph_cli.deploy import (
     _parse_env_from_config,
     _resolve_env_path,
     _resolve_pushed_image_digest,
+    _resource_spec,
     _select_source,
     _validate_prebuilt_image,
     find_deployment_by_name,
@@ -46,7 +48,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 +58,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"]
@@ -593,9 +610,9 @@ class TestDockerBuildCommand:
                 False,
                 DockerBuildCommand(
                     ("docker", "buildx", "build"),
-                    ("--platform", "linux/amd64", "--load", "--progress=quiet"),
+                    ("--platform", "linux/amd64", "--load", "--progress=plain"),
                 ),
-                id="other_hosts_cross_build_quietly",
+                id="other_hosts_capture_plain_build_logs",
             ),
             pytest.param(
                 "arm64",
@@ -621,6 +638,7 @@ class TestSelectSource:
         "tag": None,
         "remote_build_flag": None,
         "placement": RequestedPlacement(),
+        "image_pull_secrets": (),
         "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
@@ -636,7 +654,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -648,7 +666,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "v2"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -660,7 +678,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "v3"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -672,7 +690,7 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image="app:dev",
                     ),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
             ),
@@ -687,16 +705,35 @@ class TestSelectSource:
                         reference=ImageReference(REPOSITORY, "latest"),
                         prebuilt_image=None,
                     ),
-                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement("listener-1", "agents")
+                    ),
                 ),
                 id="push_to_carries_the_requested_placement",
             ),
+            pytest.param(
+                {
+                    "push_to": REPOSITORY,
+                    "image_pull_secrets": ("regcred", "other-secret"),
+                },
+                True,
+                CustomerRegistrySource(
+                    image=BuildAndPush(
+                        reference=ImageReference(REPOSITORY, "latest"),
+                        prebuilt_image=None,
+                    ),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement(), ("regcred", "other-secret")
+                    ),
+                ),
+                id="push_to_carries_the_image_pull_secrets",
+            ),
             pytest.param(
                 {"image_uri": f"{REPOSITORY}@sha256:abc123"},
                 True,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="image_uri_selects_the_published_image_source_by_digest",
             ),
@@ -705,7 +742,7 @@ class TestSelectSource:
                 False,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement(),
+                    create_only=CreateOnlySettings(RequestedPlacement()),
                 ),
                 id="image_uri_needs_no_local_docker_and_is_trimmed",
             ),
@@ -717,7 +754,9 @@ class TestSelectSource:
                 False,
                 CustomerRegistrySource(
                     image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
-                    requested_placement=RequestedPlacement("listener-1", "agents"),
+                    create_only=CreateOnlySettings(
+                        RequestedPlacement("listener-1", "agents")
+                    ),
                 ),
                 id="image_uri_carries_the_requested_placement",
             ),
@@ -806,6 +845,11 @@ class TestSelectSource:
                 "only apply when creating a deployment with --push-to",
                 id="namespace_without_push_to",
             ),
+            pytest.param(
+                {"image_pull_secrets": ("regcred",)},
+                "only apply when creating a deployment with --push-to",
+                id="image_pull_secret_without_push_to",
+            ),
             pytest.param(
                 {"image_uri": REPOSITORY, "push_to": REPOSITORY},
                 "--image-uri cannot be combined with --push-to.",
@@ -1243,7 +1287,7 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
 
     monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
 
-    with pytest.raises(click.UsageError, match="--deployment-id"):
+    with pytest.raises(click.UsageError, match="dep-1"):
         _select_source(
             push_to="registry.example.com/app",
             image=None,
@@ -1252,10 +1296,77 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
             tag=None,
             remote_build_flag=None,
             placement=RequestedPlacement(listener_id="listener-1"),
+            image_pull_secrets=(),
             selector=ById("dep-1"),
         )
 
 
+def test_a_deployment_id_with_image_pull_secrets_is_refused_without_probing_docker(
+    monkeypatch,
+):
+    def explode() -> tuple[bool, str | None]:
+        raise AssertionError("docker must not be probed for an argv-only conflict")
+
+    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
+
+    with pytest.raises(click.UsageError, match="dep-1"):
+        _select_source(
+            push_to="registry.example.com/app",
+            image=None,
+            image_uri=None,
+            image_name=None,
+            tag=None,
+            remote_build_flag=None,
+            placement=RequestedPlacement(),
+            image_pull_secrets=("regcred",),
+            selector=ById("dep-1"),
+        )
+
+
+class TestResourceSpec:
+    def test_no_secrets_keeps_the_operator_default(self):
+        assert _resource_spec(()) == {}
+
+    def test_secrets_become_local_object_references_in_order(self):
+        assert _resource_spec(("regcred", "other-secret")) == {
+            "image_pull_secrets": [
+                {"name": "regcred"},
+                {"name": "other-secret"},
+            ],
+        }
+
+
+class TestCreateOnlySettings:
+    @pytest.mark.parametrize(
+        "name",
+        [
+            pytest.param("Bad_Name", id="uppercase_and_underscore"),
+            pytest.param("", id="empty"),
+            pytest.param("-leading-hyphen", id="leading_hyphen"),
+            pytest.param("a" * 254, id="too_long"),
+        ],
+    )
+    def test_an_invalid_secret_name_is_rejected_up_front(self, name):
+        with pytest.raises(click.UsageError, match="not a valid Kubernetes"):
+            CreateOnlySettings(RequestedPlacement(), (name,))
+
+    def test_a_valid_secret_name_is_accepted(self):
+        CreateOnlySettings(RequestedPlacement(), ("regcred-1.example",))
+
+    def test_ensure_not_existing_mentions_the_api_and_ui_for_pull_secrets(self):
+        settings = CreateOnlySettings(RequestedPlacement(), ("regcred",))
+        with pytest.raises(click.UsageError, match="API or the LangSmith UI"):
+            settings.ensure_not_existing("dep-1")
+
+    def test_ensure_not_existing_keeps_the_placement_message_unchanged(self):
+        settings = CreateOnlySettings(RequestedPlacement(listener_id="listener-1"))
+        with pytest.raises(
+            click.UsageError,
+            match="Listener and namespace are fixed when a deployment is created",
+        ):
+            settings.ensure_not_existing("dep-1")
+
+
 class TestPlacementOnAKnownListener:
     @pytest.mark.parametrize(
         ("request_", "listener", "expected"),

libs/langgraph/langgraph/channels/binop.py

@@ -2,6 +2,7 @@ import collections.abc
 from collections.abc import Callable, Sequence
 from typing import Any, Generic
 
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import NotRequired, Required, Self
 
 from langgraph._internal._constants import OVERWRITE
@@ -116,6 +117,9 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
     def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ, self.operator)
         empty.key = self.key
+        # A thread written while this field was a DeltaChannel stores snapshots.
+        if isinstance(checkpoint, _DeltaSnapshot):
+            checkpoint = checkpoint.value
         if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -4,7 +4,7 @@ import uuid
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from datetime import datetime, timezone
 from inspect import signature
-from typing import Any, Literal, cast
+from typing import Any, Literal
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -12,6 +12,7 @@ from langgraph.checkpoint.base import (
     ChannelVersions,
     Checkpoint,
     CheckpointTuple,
+    DeltaChannelHistory,
     PendingWrite,
 )
 from langgraph.checkpoint.base.id import uuid6
@@ -32,7 +33,7 @@ from langgraph._internal._constants import (
 )
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.binop import _get_overwrite
+from langgraph.channels.binop import BinaryOperatorAggregate, _get_overwrite
 from langgraph.channels.delta import DeltaChannel
 from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
 
@@ -409,23 +410,39 @@ def versions_seen_without_bumps(
 def _delta_channels_to_replay(
     specs: Mapping[str, BaseChannel], checkpoint: Checkpoint
 ) -> list[str]:
-    """DeltaChannels whose value at `checkpoint` the ancestor walk rebuilds.
+    """Channels whose value at `checkpoint` the ancestor walk rebuilds.
 
     A `_DeltaSnapshot` blob or a plain value (migration) resolves directly via
     `from_checkpoint`, so only a channel with nothing stored here needs the
     walk. A channel with no version was never written, so it is empty without
     one; a walk for it would find no snapshot to stop at and read every
     ancestor, every time the thread is loaded.
+
+    A `BinaryOperatorAggregate` stores its value at every version, so one with
+    nothing stored was written as a `DeltaChannel` before its annotation
+    changed, and is rebuilt the same way.
     """
     return [
         k
         for k, spec in specs.items()
-        if isinstance(spec, DeltaChannel)
+        if isinstance(spec, (DeltaChannel, BinaryOperatorAggregate))
         and k in checkpoint["channel_versions"]
         and checkpoint["channel_values"].get(k, MISSING) is MISSING
     ]
 
 
+def _replay(spec: BaseChannel, history: DeltaChannelHistory) -> BaseChannel:
+    ch = spec.from_checkpoint(history.get("seed", MISSING))
+    if isinstance(ch, DeltaChannel):
+        ch.replay_writes(history["writes"])
+    else:
+        # One update per write: the writes span many supersteps, and a single
+        # call would reject a second Overwrite and drop the writes after one.
+        for _, _, value in history["writes"]:
+            ch.update([value])
+    return ch
+
+
 def _require_saver_for_history(
     delta_channels: list[str],
     saver: BaseCheckpointSaver | None,
@@ -465,7 +482,7 @@ def channels_from_checkpoint(
 
     delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
     _require_saver_for_history(delta_channels, saver, config)
-    histories: Mapping[str, Any] = {}
+    histories: Mapping[str, DeltaChannelHistory] = {}
     if delta_channels and saver is not None and config is not None:
         histories = saver.get_delta_channel_history(
             config=config, channels=delta_channels
@@ -475,11 +492,7 @@ def channels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
-            delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
-            replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
-            replay_ch.replay_writes(history["writes"])
-            ch = replay_ch
+            ch = _replay(spec, histories[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch
@@ -504,7 +517,7 @@ async def achannels_from_checkpoint(
 
     delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
     _require_saver_for_history(delta_channels, saver, config)
-    histories: Mapping[str, Any] = {}
+    histories: Mapping[str, DeltaChannelHistory] = {}
     if delta_channels and saver is not None and config is not None:
         histories = await saver.aget_delta_channel_history(
             config=config, channels=delta_channels
@@ -514,11 +527,7 @@ async def achannels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         if k in histories:
-            delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
-            replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
-            replay_ch.replay_writes(history["writes"])
-            ch = replay_ch
+            ch = _replay(spec, histories[k])
         else:
             ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch

libs/langgraph/tests/test_delta_channel_to_plain.py

@@ -0,0 +1,148 @@
+import operator
+from typing import Annotated, Any
+from unittest.mock import patch
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from typing_extensions import TypedDict
+
+from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer, add_messages
+from langgraph.pregel._checkpoint import channels_from_checkpoint, empty_checkpoint
+from langgraph.types import Overwrite
+
+pytestmark = pytest.mark.anyio
+
+
+class DeltaState(TypedDict):
+    messages: Annotated[
+        list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3)
+    ]
+
+
+class PlainState(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+def _app(schema: type, checkpointer: BaseCheckpointSaver) -> Any:
+    builder = StateGraph(schema)
+    builder.add_node(
+        "chat",
+        lambda state: {"messages": [AIMessage(f"reply {len(state['messages'])}")]},
+    )
+    builder.add_edge(START, "chat")
+    builder.add_edge("chat", END)
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _delta_thread(checkpointer: BaseCheckpointSaver) -> RunnableConfig:
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    delta = _app(DeltaState, checkpointer)
+    for i in range(4):
+        delta.invoke({"messages": [HumanMessage(f"q{i}")]}, config)
+    assert any(
+        isinstance(t.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
+        for t in checkpointer.list(config)
+    ), "the thread should have a snapshot to read through"
+    return config
+
+
+def test_plain_reducer_reads_every_delta_written_checkpoint(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _delta_thread(sync_checkpointer)
+    delta = _app(DeltaState, sync_checkpointer)
+    plain = _app(PlainState, sync_checkpointer)
+
+    assert [s.values for s in plain.get_state_history(config)] == [
+        s.values for s in delta.get_state_history(config)
+    ]
+
+
+def test_plain_reducer_continues_delta_written_thread(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _delta_thread(sync_checkpointer)
+    delta = _app(DeltaState, sync_checkpointer)
+    before = delta.get_state(config).values["messages"]
+
+    after = _app(PlainState, sync_checkpointer).invoke(
+        {"messages": [HumanMessage("q4")]}, config
+    )["messages"]
+
+    assert after[:-2] == before
+    assert [m.content for m in after[-2:]] == ["q4", f"reply {len(before) + 1}"]
+    head = sync_checkpointer.get_tuple(config)
+    assert head is not None
+    assert head.checkpoint["channel_values"]["messages"] == after
+    assert delta.get_state(config).values["messages"] == after
+
+
+async def test_plain_reducer_reads_and_continues_delta_written_thread_async(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    delta = _app(DeltaState, async_checkpointer)
+    for i in range(4):
+        await delta.ainvoke({"messages": [HumanMessage(f"q{i}")]}, config)
+    plain = _app(PlainState, async_checkpointer)
+
+    expected = [s.values async for s in delta.aget_state_history(config)]
+    assert [s.values async for s in plain.aget_state_history(config)] == expected
+
+    after = (await plain.ainvoke({"messages": [HumanMessage("q4")]}, config))[
+        "messages"
+    ]
+    assert after[:-2] == expected[0]["messages"]
+
+
+@pytest.mark.parametrize(
+    ("seed", "writes", "expected"),
+    [
+        (_DeltaSnapshot(["a"]), [["b"], ["c"]], ["a", "b", "c"]),
+        (["a"], [["b"], Overwrite(["c"]), ["d"]], ["c", "d"]),
+        (None, [["b"], Overwrite(["c"]), Overwrite(["d"]), ["e"]], ["d", "e"]),
+    ],
+)
+def test_plain_reducer_replays_delta_history_one_write_at_a_time(
+    seed: Any, writes: list[Any], expected: list[str]
+) -> None:
+    history: dict[str, Any] = {"writes": [("t", "items", w) for w in writes]}
+    if seed is not None:
+        history["seed"] = seed
+    checkpoint = empty_checkpoint()
+    checkpoint["channel_versions"]["items"] = 1
+    saver = InMemorySaver()
+
+    with patch.object(
+        saver, "get_delta_channel_history", return_value={"items": history}
+    ):
+        channels, _ = channels_from_checkpoint(
+            {"items": BinaryOperatorAggregate(list, operator.add)},
+            checkpoint,
+            saver=saver,
+            config={"configurable": {"thread_id": "1"}},
+        )
+
+    assert channels["items"].get() == expected
+
+
+def test_plain_thread_reads_never_walk_history() -> None:
+    saver = InMemorySaver()
+    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
+    plain = _app(PlainState, saver)
+
+    with patch.object(saver, "get_delta_channel_history") as walk:
+        plain.invoke({"messages": [HumanMessage("q0")]}, config)
+        plain.update_state(config, {"messages": [HumanMessage("note")]})
+        plain.invoke({"messages": [HumanMessage("q1")]}, config)
+        plain.get_state(config)
+        list(plain.get_state_history(config))
+
+    walk.assert_not_called()