build: centralize pinned base image arguments

tests/unit/test_builtin_agent_images.py

@@ -2,30 +2,43 @@
 
 from __future__ import annotations
 
+import json
 import re
 import unittest
 from pathlib import Path
 
 
-_CONTRIB_DIR = Path(__file__).resolve().parents[2] / "bot_bottle/contrib"
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_CONTRIB_DIR = _REPO_ROOT / "bot_bottle/contrib"
 _AGENT_DOCKERFILES = tuple(sorted(_CONTRIB_DIR.glob("*/Dockerfile")))
 
 
 class TestBuiltinAgentImages(unittest.TestCase):
     def test_all_share_one_digest_pinned_node_trixie_base(self):
         self.assertTrue(_AGENT_DOCKERFILES)
-        bases = []
+        inputs = json.loads((_REPO_ROOT / "image-build-args.json").read_text())
+        self.assertRegex(
+            inputs["NODE_BASE_IMAGE"],
+            r"^node:22\.\d+\.\d+-trixie-slim@sha256:[0-9a-f]{64}$",
+        )
         for dockerfile in _AGENT_DOCKERFILES:
             with self.subTest(provider=dockerfile.parent.name):
-                match = re.search(
-                    r"(?m)^FROM "
-                    r"(node:22\.\d+\.\d+-trixie-slim@sha256:[0-9a-f]{64})\s*$",
-                    dockerfile.read_text(),
-                )
-                if match is None:
-                    self.fail(f"{dockerfile} does not use a digest-pinned Node base")
-                bases.append(match.group(1))
-        self.assertEqual(1, len(set(bases)))
+                text = dockerfile.read_text()
+                self.assertRegex(text, r"(?m)^ARG NODE_BASE_IMAGE$")
+                self.assertIn("FROM ${NODE_BASE_IMAGE}", text)
+
+    def test_orchestrator_and_gateway_share_configurable_python_base(self):
+        inputs = json.loads((_REPO_ROOT / "image-build-args.json").read_text())
+        self.assertRegex(
+            inputs["PYTHON_BASE_IMAGE"],
+            r"^python:3\.12\.\d+-slim-trixie@sha256:[0-9a-f]{64}$",
+        )
+        for name in ("Dockerfile.orchestrator", "Dockerfile.gateway"):
+            dockerfile = _REPO_ROOT / name
+            text = dockerfile.read_text()
+            with self.subTest(dockerfile=name):
+                self.assertRegex(text, r"(?m)^ARG PYTHON_BASE_IMAGE$")
+                self.assertIn("FROM ${PYTHON_BASE_IMAGE}", text)
 
     def test_none_install_podman(self):
         # podman lives in the nested-containers derived layer (nested_containers.py),

tests/unit/test_docker_orchestrator.py

@@ -221,6 +221,10 @@ class TestDockerOrchestrator(unittest.TestCase):
         builds = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "build"]]
         self.assertEqual(1, len(builds))
         self.assertTrue(any(a.endswith("Dockerfile.orchestrator") for a in builds[0]))
+        arg_index = builds[0].index("--build-arg")
+        self.assertTrue(
+            builds[0][arg_index + 1].startswith("PYTHON_BASE_IMAGE=python:"),
+        )
 
     def test_ensure_built_raises_on_build_failure(self) -> None:
         with patch(_RUN, return_value=_proc(returncode=1, stderr="no space left")):

tests/unit/test_docker_util_image.py

@@ -122,6 +122,27 @@ class TestCommitContainer(unittest.TestCase):
 
 
 class TestBuildImage(unittest.TestCase):
+    def test_passes_centralized_image_build_args(self):
+        with patch.object(
+            docker_mod.resources,
+            "image_build_args",
+            return_value={"NODE_BASE_IMAGE": "node:pinned"},
+        ), patch.object(
+            docker_mod.subprocess, "run", return_value=_ok(),
+        ) as run, patch.object(docker_mod, "info"):
+            docker_mod.build_image(
+                "agent:test",
+                "/context",
+                dockerfile="Dockerfile.agent",
+            )
+        self.assertIn(
+            ["--build-arg", "NODE_BASE_IMAGE=node:pinned"],
+            [
+                run.call_args.args[0][index:index + 2]
+                for index in range(len(run.call_args.args[0]) - 1)
+            ],
+        )
+
     def test_passes_build_args_without_mutating_the_value(self):
         with patch.object(
             docker_mod.subprocess, "run", return_value=_ok(),

tests/unit/test_firecracker_image_builder.py

@@ -75,8 +75,39 @@ class TestBuildAgentRootfsDir(unittest.TestCase):
         other.write_text("FROM python:3.12-slim\n")
         self.assertNotEqual(base, image_builder._rootfs_digest(other))
 
+    def test_rootfs_digest_tracks_centralized_build_args(self):
+        with patch.object(
+            image_builder.resources,
+            "image_build_args",
+            return_value={"NODE_BASE_IMAGE": "node:first"},
+        ):
+            first = image_builder._rootfs_digest(self.dockerfile)
+        with patch.object(
+            image_builder.resources,
+            "image_build_args",
+            return_value={"NODE_BASE_IMAGE": "node:second"},
+        ):
+            second = image_builder._rootfs_digest(self.dockerfile)
+        self.assertNotEqual(first, second)
+
 
 class TestSmokeTest(unittest.TestCase):
+    def test_buildah_receives_centralized_image_build_args(self):
+        with patch.object(
+            image_builder,
+            "_ssh_streamed",
+            return_value=0,
+        ) as ssh, patch.object(image_builder, "info"):
+            image_builder._buildah_build(
+                Path("/k"),
+                "10.0.0.1",
+                "/tmp/context",
+                "agent:test",
+                {"NODE_BASE_IMAGE": "node:pinned"},
+            )
+        script = ssh.call_args.args[2]
+        self.assertIn("--build-arg NODE_BASE_IMAGE=node:pinned", script)
+
     def test_empty_argv_is_noop(self):
         with patch.object(image_builder, "_ssh") as ssh:
             image_builder._smoke_test(Path("/k"), "10.0.0.1", "tag", "ctr", ())

tests/unit/test_image_input_policy.py

@@ -14,10 +14,26 @@ from scripts import check_image_inputs as policy
 
 
 class TestDockerfilePolicy(unittest.TestCase):
+    _NODE_BASE = {
+        "NODE_BASE_IMAGE":
+            "node:22.23.1-trixie-slim@sha256:" + "a" * 64,
+    }
+
     def test_accepts_versioned_digest_base(self):
         text = "FROM node:22.23.1-trixie-slim@sha256:" + "a" * 64 + "\n"
         self.assertEqual([], policy.check_dockerfile(Path("Dockerfile"), text))
 
+    def test_accepts_defaultless_base_argument_with_centralized_value(self):
+        text = "ARG NODE_BASE_IMAGE\nFROM ${NODE_BASE_IMAGE}\n"
+        self.assertEqual(
+            [],
+            policy.check_dockerfile(
+                Path("Dockerfile"),
+                text,
+                self._NODE_BASE,
+            ),
+        )
+
     def test_rejects_mutable_and_latest_bases(self):
         for value in (
             "node:22-trixie-slim",
@@ -61,7 +77,15 @@ class TestDockerfilePolicy(unittest.TestCase):
     def test_rejects_missing_dynamic_and_malformed_bases(self):
         cases = (
             ("RUN true\n", "missing FROM"),
-            ("FROM ${BASE_IMAGE}\n", "dynamic base"),
+            ("FROM ${BASE_IMAGE}\n", "not declared before FROM"),
+            (
+                "ARG BASE_IMAGE=node:latest\nFROM ${BASE_IMAGE}\n",
+                "must not define a default",
+            ),
+            (
+                "ARG BASE_IMAGE\nFROM ${BASE_IMAGE}\n",
+                "lacks a centralized value",
+            ),
             (
                 "ARG ORCHESTRATOR_BASE_IMAGE=base:latest\n"
                 "FROM ${ORCHESTRATOR_BASE_IMAGE}\n",
@@ -74,6 +98,15 @@ class TestDockerfilePolicy(unittest.TestCase):
                 problems = policy.check_dockerfile(Path("Dockerfile"), text)
                 self.assertTrue(any(expected in item for item in problems))
 
+    def test_rejects_mutable_centralized_base_value(self):
+        text = "ARG NODE_BASE_IMAGE\nFROM ${NODE_BASE_IMAGE}\n"
+        problems = policy.check_dockerfile(
+            Path("Dockerfile"),
+            text,
+            {"NODE_BASE_IMAGE": "node:latest"},
+        )
+        self.assertTrue(any("lacks a sha256 digest" in item for item in problems))
+
     def test_requires_gateway_lock_and_verified_codex_markers(self):
         base = "FROM node:22.23.1@sha256:" + "a" * 64 + "\n"
         gateway = policy.check_dockerfile(Path("Dockerfile.gateway"), base)

tests/unit/test_infra_artifact.py

@@ -110,6 +110,9 @@ class TestVersionInputs(unittest.TestCase):
         for name in ("Dockerfile.orchestrator", "Dockerfile.orchestrator.fc",
                      "Dockerfile.gateway"):
             (root / name).write_text(f"FROM scratch  # {name}\n")
+        (root / "image-build-args.json").write_text(
+            '{"PYTHON_BASE_IMAGE": "python:pinned"}\n',
+        )
         (root / "requirements.gateway.lock").write_text("mitmproxy==11.1.3\n")
         (root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
 
@@ -162,6 +165,23 @@ class TestVersionInputs(unittest.TestCase):
             )
         self.assertNotEqual(before, after)
 
+    def test_base_image_argument_change_bumps_both_role_versions(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            self._fake_repo(root)
+            before = {
+                role: ia.infra_artifact_version("init", role, repo_root=root)
+                for role in ia.ROLES
+            }
+            (root / "image-build-args.json").write_text(
+                '{"PYTHON_BASE_IMAGE": "python:different"}\n',
+            )
+            after = {
+                role: ia.infra_artifact_version("init", role, repo_root=root)
+                for role in ia.ROLES
+            }
+        self.assertTrue(all(before[role] != after[role] for role in ia.ROLES))
+
     def test_pyc_and_pycache_ignored(self) -> None:
         with tempfile.TemporaryDirectory() as d:
             root = Path(d)

tests/unit/test_macos_container_util.py

@@ -115,6 +115,37 @@ resolver #2
             run.call_args_list[-1].args[0],
         )
 
+    def test_build_image_passes_centralized_image_build_args(self):
+        status = util.subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout=(
+                '[{"status":{"state":"running"},'
+                '"configuration":{"dns":{"nameservers":["9.9.9.9"]}}}]'
+            ),
+            stderr="",
+        )
+        with patch.object(util.subprocess, "run", return_value=status) as run, \
+             patch.object(
+                 util.resources,
+                 "image_build_args",
+                 return_value={"NODE_BASE_IMAGE": "node:pinned"},
+             ), patch.object(util.os, "environ", {
+                 "BOT_BOTTLE_MACOS_CONTAINER_DNS": "9.9.9.9",
+             }):
+            util.build_image(
+                "bot-bottle-agent:latest",
+                "/repo",
+                dockerfile="Dockerfile.agent",
+            )
+        self.assertIn(
+            ["--build-arg", "NODE_BASE_IMAGE=node:pinned"],
+            [
+                run.call_args_list[-1].args[0][index:index + 2]
+                for index in range(len(run.call_args_list[-1].args[0]) - 1)
+            ],
+        )
+
     def test_commit_container_execs_tar_and_builds_image(self):
         # stderr is bytes because subprocess.run uses stderr=PIPE without text=True
         completed = util.subprocess.CompletedProcess(

tests/unit/test_orchestrator_gateway.py

@@ -323,6 +323,10 @@ class TestDockerGatewayBuild(unittest.TestCase):
         self.assertEqual(1, len(builds))
         self.assertIn(self.sc.image_ref, builds[0])
         self.assertTrue(any(a.endswith("Dockerfile.gateway") for a in builds[0]))
+        arg_index = builds[0].index("--build-arg")
+        self.assertTrue(
+            builds[0][arg_index + 1].startswith("PYTHON_BASE_IMAGE=python:"),
+        )
         self.assertNotIn("--no-cache", builds[0])
 
     def test_ensure_built_no_cache_env_forces_full_rebuild(self) -> None:

tests/unit/test_resources.py

@@ -38,6 +38,18 @@ class TestCheckoutMode(unittest.TestCase):
         self.assertTrue(resources.nix_netpool_module().is_file())
         self.assertTrue(resources.netpool_script().is_file())
 
+    def test_image_build_args_come_from_the_central_input_file(self):
+        root = resources.build_root()
+        args = resources.image_build_args(
+            "Dockerfile.gateway",
+            context=root,
+        )
+        self.assertEqual({"PYTHON_BASE_IMAGE"}, set(args))
+        self.assertRegex(
+            args["PYTHON_BASE_IMAGE"],
+            r"^python:\d+\.\d+\.\d+-.+@sha256:[0-9a-f]{64}$",
+        )
+
     def test_bundled_resources_all_exist_at_root(self):
         # Drift guard: every path setup.py bundles must exist in the checkout.
         root = resources.build_root()