fix(cli): pin internal_docker deploy images by digest (#7924)

## Summary

`langgraph deploy` now pins images by digest when handing the URI to the
LangGraph host backend. After `docker push`, the CLI reads the manifest
digest from the local Docker daemon's `RepoDigests` and sends
`registry/repo@sha256:<hex>` to the host backend instead of the
tag-based reference. Mutable tags cause downstream inconsistency — the
same revision can refer to different images over time.

The image is still pushed under the user-supplied `--tag` (default
`:latest`) so it stays discoverable by tag in the registry — only the
URI persisted with the revision changes.

## Behavior on failure

If the digest can't be resolved (empty `RepoDigests`, or no entry
matching the just-pushed repo), the CLI warns and falls back to the
tag-based reference. Deploys never fail on a digest-resolution issue.

## Test plan

- [x] `make test` passes (new `TestResolvePushedImageDigest` cases
included)
- [x] `--verbose` deploy shows the `docker image inspect` call resolving
  the digest
- [x] Deploy with a clean Docker daemon (no matching `RepoDigests`)
  emits the fallback warning and still completes successfully

---------

Co-authored-by: Josh Rogers <josh@langchain.dev>

libs/cli/langgraph_cli/deploy.py

@@ -849,6 +849,41 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _resolve_pushed_image_digest(
+    runner,
+    *,
+    remote_image: str,
+    docker_config_dir: str | None,
+    verbose: bool,
+) -> str:
+    """Return ``{registry}/{repo}@sha256:<hex>`` for a freshly-pushed image.
+
+    Reads ``RepoDigests`` via ``docker image inspect`` — the local daemon
+    records the registry's manifest digest there after a successful push.
+    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))
+    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:"):
+            return d
+    _get_emitter().warn(
+        f"Could not resolve image digest for {remote_image}; "
+        "falling back to the tag-based reference. Re-run with --verbose for details."
+    )
+    return remote_image
+
+
 def _run_local_build(
     *,
     client: HostBackendClient,
@@ -1015,9 +1050,18 @@ def _run_local_build(
                         raise
         step += 1
 
+        resolved_image = _resolve_pushed_image_digest(
+            runner,
+            remote_image=remote_image,
+            docker_config_dir=None,
+            verbose=verbose,
+        )
+
         # -- Step: Update deployment --
         _log_deploy_step(step, f"Updating deployment {deployment_id}")
-        updated = client.update_deployment(deployment_id, remote_image, secrets=secrets)
+        updated = client.update_deployment(
+            deployment_id, resolved_image, secrets=secrets
+        )
 
     return BuildResult(
         updated=updated if isinstance(updated, dict) else {},

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -3,6 +3,7 @@ import io
 import json
 import os
 import sys
+from unittest.mock import MagicMock
 
 import click
 import httpx
@@ -16,6 +17,7 @@ from langgraph_cli.deploy import (
     _env_without_deployment_name,
     _parse_env_from_config,
     _resolve_env_path,
+    _resolve_pushed_image_digest,
     _smith_dashboard_base_url,
     normalize_image_tag,
     normalize_name,
@@ -532,3 +534,156 @@ class TestSmithDashboardBaseUrl:
             _smith_dashboard_base_url("https://custom.example.com")
             == "https://smith.langchain.com"
         )
+
+
+class TestResolvePushedImageDigest:
+    """Tests for ``_resolve_pushed_image_digest`` — runner is mocked to
+    return the ``(stdout, stderr)`` tuple that ``subp_exec(collect=True)``
+    would produce.
+    """
+
+    @staticmethod
+    def _runner(stdout: str | None) -> MagicMock:
+        # Close the unawaited subp_exec coroutine to silence gc warnings.
+        runner = MagicMock()
+
+        def _run(coro, *args, **kwargs):
+            if hasattr(coro, "close"):
+                coro.close()
+            return (stdout, "")
+
+        runner.run.side_effect = _run
+        return runner
+
+    def test_happy_path_returns_digest(self):
+        runner = self._runner('["us-central1-docker.pkg.dev/proj/repo@sha256:abc123"]')
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image="us-central1-docker.pkg.dev/proj/repo:latest",
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == "us-central1-docker.pkg.dev/proj/repo@sha256:abc123"
+
+    def test_filters_to_matching_repo(self):
+        # Same image ID can hold digests for multiple repos — pick the one
+        # matching the just-pushed repo.
+        runner = self._runner(
+            json.dumps(
+                [
+                    "other-registry.example.com/some/repo@sha256:000000",
+                    "us-central1-docker.pkg.dev/proj/repo@sha256:abc123",
+                ]
+            )
+        )
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image="us-central1-docker.pkg.dev/proj/repo:v1.2.3",
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == "us-central1-docker.pkg.dev/proj/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)
+        runner = self._runner("[]")
+        remote = "us-central1-docker.pkg.dev/proj/repo:latest"
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image=remote,
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == remote
+        assert emitter.warn.called
+        assert remote in emitter.warn.call_args.args[0]
+
+    def test_null_repodigests_falls_back_with_warning(self, mocker):
+        # ``docker inspect --format '{{json .RepoDigests}}'`` emits ``null``
+        # when the field is absent.
+        emitter = mocker.MagicMock()
+        mocker.patch("langgraph_cli.deploy._get_emitter", return_value=emitter)
+        runner = self._runner("null")
+        remote = "us-central1-docker.pkg.dev/proj/repo:latest"
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image=remote,
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == remote
+        assert emitter.warn.called
+
+    def test_no_matching_repo_falls_back_with_warning(self, mocker):
+        # No matching digest for the pushed repo — warn and fall back to the
+        # tag-based ref rather than failing the deploy.
+        emitter = mocker.MagicMock()
+        mocker.patch("langgraph_cli.deploy._get_emitter", return_value=emitter)
+        runner = self._runner('["other-registry.example.com/some/repo@sha256:000000"]')
+        remote = "us-central1-docker.pkg.dev/proj/repo:latest"
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image=remote,
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == remote
+        assert emitter.warn.called
+
+    def test_registry_with_port_in_host(self):
+        # Only the rightmost ``:`` (the ``:latest`` tag) should be stripped.
+        runner = self._runner('["localhost:5000/repo@sha256:deadbeef"]')
+        out = _resolve_pushed_image_digest(
+            runner,
+            remote_image="localhost:5000/repo:latest",
+            docker_config_dir=None,
+            verbose=False,
+        )
+        assert out == "localhost:5000/repo@sha256:deadbeef"
+
+    @staticmethod
+    def _capturing_runner(stdout: str) -> tuple[MagicMock, dict]:
+        """Like ``_runner`` but exposes the coroutine for arg introspection.
+        Caller must close ``captured["coro"]``.
+        """
+        runner = MagicMock()
+        captured: dict = {}
+
+        def _run(coro, *args, **kwargs):
+            captured["coro"] = coro
+            return (stdout, "")
+
+        runner.run.side_effect = _run
+        return runner, captured
+
+    def test_passes_docker_config_dir(self):
+        runner, captured = self._capturing_runner(
+            '["us-central1-docker.pkg.dev/proj/repo@sha256:abc"]'
+        )
+        _resolve_pushed_image_digest(
+            runner,
+            remote_image="us-central1-docker.pkg.dev/proj/repo:latest",
+            docker_config_dir="/tmp/some-cfg",
+            verbose=False,
+        )
+        frame_locals = captured["coro"].cr_frame.f_locals
+        assert frame_locals["cmd"] == "docker"
+        assert "--config" in frame_locals["args"]
+        cfg_idx = frame_locals["args"].index("--config")
+        assert frame_locals["args"][cfg_idx + 1] == "/tmp/some-cfg"
+        captured["coro"].close()
+
+    def test_omits_docker_config_dir_when_none(self):
+        runner, captured = self._capturing_runner(
+            '["us-central1-docker.pkg.dev/proj/repo@sha256:abc"]'
+        )
+        _resolve_pushed_image_digest(
+            runner,
+            remote_image="us-central1-docker.pkg.dev/proj/repo:latest",
+            docker_config_dir=None,
+            verbose=False,
+        )
+        frame_locals = captured["coro"].cr_frame.f_locals
+        assert "--config" not in frame_locals["args"]
+        captured["coro"].close()