feat(smolmachines): build agent image from repo Dockerfile (PRD 0023 chunk 4c)

Replaces the alpine:latest placeholder with a real claude-bottle
agent image, converted into a .smolmachine artifact via an
ephemeral local OCI registry.

Why the registry hop: smolvm pack create only accepts OCI registry
refs. Empirically it rejects docker-daemon://, oci-layout://,
docker-archive: tarballs, and every other transport tested — the
crane backend treats anything with a scheme prefix as a registry
hostname. To convert a locally-built docker image into a
.smolmachine we have to push it somewhere smolvm can pull from.
Smallest path: bring up registry:2.8.3 bound to 127.0.0.1:<random>,
docker tag + docker push into it, smolvm pack create --image
localhost:<port>/claude-bottle:<id>, tear down the registry.

The .smolmachine is cached under
~/.cache/claude-bottle/smolmachines/ keyed by the docker image ID
(first 16 hex chars of the sha256), so a Dockerfile change picks
up a new image ID and invalidates the cache. Unchanged rebuilds
skip the whole build → registry → pack pipeline.

This puts `docker build` in smolmachines prepare (the docker
backend defers it to launch). Necessary because pack_create needs
the image ID to derive the cache key, and prepare is the only
hook ahead of launch that runs once per slug.

Adds:
- claude_bottle/backend/docker/util.py: image_id / tag / push
  helpers (thin docker CLI wrappers).
- claude_bottle/backend/smolmachines/local_registry.py:
  ephemeral_registry() context manager; pins registry:2.8.3 by
  digest, binds 127.0.0.1::5000 (loopback-only), force-removes on
  exit.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

tests/unit/test_docker_util_image.py

@@ -0,0 +1,79 @@
+"""Unit: image_id / tag / push helpers in
+claude_bottle.backend.docker.util (PRD 0023 chunk 4c additions).
+
+Tests mock `subprocess.run` and assert on argv shape + parsing.
+The actual docker round-trip is covered by the chunk 4c
+integration smoke."""
+
+from __future__ import annotations
+
+import subprocess
+import unittest
+from unittest.mock import patch
+
+from claude_bottle.backend.docker import util as docker_mod
+
+
+def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=0, stdout=stdout, stderr=stderr,
+    )
+
+
+def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=1, stdout="", stderr=stderr,
+    )
+
+
+class TestImageId(unittest.TestCase):
+    def test_strips_trailing_newline(self):
+        # docker image inspect --format ... emits a trailing newline.
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout="sha256:abcdef\n"),
+        ) as run:
+            self.assertEqual(
+                "sha256:abcdef", docker_mod.image_id("claude-bottle:latest")
+            )
+        argv = run.call_args.args[0]
+        self.assertEqual(
+            ["docker", "image", "inspect", "--format", "{{.Id}}", "claude-bottle:latest"],
+            argv,
+        )
+
+    def test_dies_on_inspect_failure(self):
+        with patch.object(
+            docker_mod.subprocess, "run", return_value=_fail("No such image"),
+        ), patch.object(
+            docker_mod, "die", side_effect=SystemExit("die"),
+        ) as die:
+            with self.assertRaises(SystemExit):
+                docker_mod.image_id("missing:tag")
+            die.assert_called_once()
+            self.assertIn("missing:tag", die.call_args.args[0])
+
+
+class TestTagPush(unittest.TestCase):
+    def test_tag_runs_docker_tag(self):
+        with patch.object(
+            docker_mod.subprocess, "run", return_value=_ok(),
+        ) as run:
+            docker_mod.tag("claude-bottle:latest", "localhost:5000/cb:abc")
+        argv = run.call_args.args[0]
+        self.assertEqual(
+            ["docker", "tag", "claude-bottle:latest", "localhost:5000/cb:abc"],
+            argv,
+        )
+
+    def test_push_runs_docker_push(self):
+        with patch.object(
+            docker_mod.subprocess, "run", return_value=_ok(),
+        ) as run:
+            docker_mod.push("localhost:5000/cb:abc")
+        argv = run.call_args.args[0]
+        self.assertEqual(["docker", "push", "localhost:5000/cb:abc"], argv)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_local_registry.py

@@ -0,0 +1,141 @@
+"""Unit: ephemeral local-registry helper (PRD 0023 chunk 4c).
+
+The helper brings up a `registry:2.8.3` container on a random
+loopback port, yields the port, and tears the container down on
+exit. Tests mock `subprocess.run` + `socket.create_connection` so
+they run without docker."""
+
+from __future__ import annotations
+
+import subprocess
+import unittest
+from unittest.mock import call, patch
+
+from claude_bottle.backend.smolmachines import local_registry
+
+
+def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=0, stdout=stdout, stderr=stderr,
+    )
+
+
+class TestEphemeralRegistry(unittest.TestCase):
+    def test_yields_host_port_parsed_from_docker_port(self):
+        # docker run + docker port + docker rm in that order; the
+        # port command returns `127.0.0.1:54321` for the loopback
+        # binding.
+        with patch.object(
+            local_registry.subprocess, "run",
+            side_effect=[
+                _ok(stdout="<container-id>\n"),
+                _ok(stdout="127.0.0.1:54321\n"),
+                _ok(),
+            ],
+        ) as run, patch.object(
+            local_registry.socket, "create_connection",
+            return_value=_FakeSocket(),
+        ):
+            with local_registry.ephemeral_registry() as port:
+                self.assertEqual(54321, port)
+
+        # docker run, docker port, docker rm -f
+        self.assertEqual(3, run.call_count)
+        run_argv = run.call_args_list[0].args[0]
+        self.assertEqual(["docker", "run"], run_argv[:2])
+        self.assertIn("--rm", run_argv)
+        # Loopback-only port binding so the registry isn't exposed
+        # on the LAN even briefly.
+        self.assertIn("127.0.0.1::5000", run_argv)
+
+    def test_force_removes_container_on_clean_exit(self):
+        with patch.object(
+            local_registry.subprocess, "run",
+            side_effect=[_ok(stdout="cid\n"), _ok(stdout="127.0.0.1:1234\n"), _ok()],
+        ) as run, patch.object(
+            local_registry.socket, "create_connection",
+            return_value=_FakeSocket(),
+        ):
+            with local_registry.ephemeral_registry():
+                pass
+
+        # Last call is `docker rm -f <name>`.
+        last_argv = run.call_args_list[-1].args[0]
+        self.assertEqual(["docker", "rm", "-f"], last_argv[:3])
+
+    def test_force_removes_container_on_exception_inside_with(self):
+        with patch.object(
+            local_registry.subprocess, "run",
+            side_effect=[_ok(stdout="cid\n"), _ok(stdout="127.0.0.1:1234\n"), _ok()],
+        ) as run, patch.object(
+            local_registry.socket, "create_connection",
+            return_value=_FakeSocket(),
+        ):
+            with self.assertRaises(RuntimeError):
+                with local_registry.ephemeral_registry():
+                    raise RuntimeError("inside with")
+
+        # rm -f still ran on exception.
+        last_argv = run.call_args_list[-1].args[0]
+        self.assertEqual(["docker", "rm", "-f"], last_argv[:3])
+
+    def test_wait_ready_times_out_when_socket_never_connects(self):
+        # Drop the timeout to a value that fits the test budget.
+        with patch.object(local_registry, "_READY_TIMEOUT_S", 0.1), patch.object(
+            local_registry.subprocess, "run",
+            side_effect=[_ok(stdout="cid\n"), _ok(stdout="127.0.0.1:1234\n"), _ok()],
+        ) as run, patch.object(
+            local_registry.socket, "create_connection",
+            side_effect=OSError("conn refused"),
+        ), patch.object(
+            local_registry, "die",
+            side_effect=SystemExit("die called"),
+        ) as die:
+            with self.assertRaises(SystemExit):
+                with local_registry.ephemeral_registry():
+                    self.fail("yield reached despite unreachable registry")
+            die.assert_called_once()
+        # rm -f still ran (cleanup goes through the finally block).
+        last_argv = run.call_args_list[-1].args[0]
+        self.assertEqual(["docker", "rm", "-f"], last_argv[:3])
+
+    def test_unique_container_name_per_call(self):
+        names: list[str] = []
+
+        def capture(argv, *a, **kw):
+            if argv[:2] == ["docker", "run"]:
+                names.append(argv[argv.index("--name") + 1])
+            return _ok(stdout="cid\n" if argv[:2] == ["docker", "run"]
+                       else "127.0.0.1:1\n")
+
+        with patch.object(
+            local_registry.subprocess, "run", side_effect=capture,
+        ), patch.object(
+            local_registry.socket, "create_connection",
+            return_value=_FakeSocket(),
+        ):
+            with local_registry.ephemeral_registry():
+                pass
+            with local_registry.ephemeral_registry():
+                pass
+
+        self.assertEqual(2, len(names))
+        self.assertNotEqual(names[0], names[1])
+        for n in names:
+            self.assertTrue(n.startswith("claude-bottle-registry-"))
+
+
+class _FakeSocket:
+    """Minimal context-manager stand-in for the socket
+    `create_connection` returns. The helper only uses `with` on it
+    and discards the value, so we don't need any real network."""
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_prepare_image.py

@@ -0,0 +1,121 @@
+"""Unit: smolmachines `_ensure_smolmachine` agent-image pipeline
+(PRD 0023 chunk 4c).
+
+Asserts that the cache-hit path returns without touching the
+registry / pack pipeline, and that the cache-miss path runs
+build → tag → push → pack in order against a registry port the
+helper yields."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from claude_bottle.backend.smolmachines import prepare as _prepare
+
+
+class TestEnsureSmolmachine(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="cb-cache.")
+        self._cache_patch = patch.object(
+            _prepare, "_SMOLMACHINE_CACHE_DIR", Path(self._tmp.name),
+        )
+        self._cache_patch.start()
+
+    def tearDown(self):
+        self._cache_patch.stop()
+        self._tmp.cleanup()
+
+    def test_cache_hit_skips_registry_and_pack(self):
+        # Pre-populate the cache for image id `sha256:abcdef0123456789...`.
+        digest = "abcdef0123456789"
+        sidecar = Path(self._tmp.name) / f"{digest}.smolmachine.smolmachine"
+        sidecar.write_text("")
+
+        with patch.object(
+            _prepare.docker_mod, "build_image",
+        ) as build, patch.object(
+            _prepare.docker_mod, "image_id",
+            return_value=f"sha256:{digest}fffffffffffffffff",
+        ), patch.object(
+            _prepare, "ephemeral_registry",
+        ) as registry, patch.object(
+            _prepare.docker_mod, "tag",
+        ) as tag, patch.object(
+            _prepare.docker_mod, "push",
+        ) as push, patch.object(
+            _prepare._smolvm, "pack_create",
+        ) as pack:
+            result = _prepare._ensure_smolmachine("claude-bottle:latest")
+
+        self.assertEqual(sidecar, result)
+        # build still runs (Dockerfile edits land without manual rmi)
+        build.assert_called_once()
+        # No registry, no tag, no push, no pack on cache hit.
+        registry.assert_not_called()
+        tag.assert_not_called()
+        push.assert_not_called()
+        pack.assert_not_called()
+
+    def test_cache_miss_runs_build_tag_push_pack_in_order(self):
+        digest = "0123456789abcdef"
+
+        # ephemeral_registry is a context manager yielding the port.
+        class _Reg:
+            def __enter__(self_inner):
+                return 54321
+            def __exit__(self_inner, *exc):
+                return False
+
+        calls: list[str] = []
+
+        def record(name):
+            def _f(*a, **kw):
+                calls.append(name)
+            return _f
+
+        with patch.object(
+            _prepare.docker_mod, "build_image",
+            side_effect=record("build"),
+        ), patch.object(
+            _prepare.docker_mod, "image_id",
+            return_value=f"sha256:{digest}fffffffffffffffff",
+        ), patch.object(
+            _prepare, "ephemeral_registry",
+            return_value=_Reg(),
+        ), patch.object(
+            _prepare.docker_mod, "tag",
+            side_effect=record("tag"),
+        ) as tag, patch.object(
+            _prepare.docker_mod, "push",
+            side_effect=record("push"),
+        ) as push, patch.object(
+            _prepare._smolvm, "pack_create",
+            side_effect=record("pack"),
+        ) as pack:
+            _prepare._ensure_smolmachine("claude-bottle:latest")
+
+        # build first (no point pushing if the build fails), then
+        # tag → push → pack against the registry port.
+        self.assertEqual(["build", "tag", "push", "pack"], calls)
+
+        # tag goes from the source ref to a localhost:<port> ref
+        # with the digest as the tag suffix (so different builds
+        # land on different tags in the registry).
+        tag_args = tag.call_args.args
+        self.assertEqual("claude-bottle:latest", tag_args[0])
+        self.assertEqual(f"localhost:54321/claude-bottle:{digest}", tag_args[1])
+        # push targets the same localhost ref tag picks.
+        push_args = push.call_args.args
+        self.assertEqual(f"localhost:54321/claude-bottle:{digest}", push_args[0])
+        # pack_create reads from the registry ref, writes the
+        # binary alongside the cached sidecar.
+        pack_args = pack.call_args.args
+        self.assertEqual(f"localhost:54321/claude-bottle:{digest}", pack_args[0])
+        self.assertTrue(str(pack_args[1]).endswith(f"{digest}.smolmachine"))
+
+
+if __name__ == "__main__":
+    unittest.main()