bot_bottle/backend/__init__.py

@@ -276,6 +276,18 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
 CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan)
 
 
+@dataclass(frozen=True)
+class BottleImages:
+    """Resolved image references (or artifact paths) for a bottle launch.
+
+    For Docker/macOS-container backends, `agent` and `sidecar` are string
+    image refs. For the smolmachines backend they are Path objects pointing
+    to pre-built `.smolmachine` artifacts."""
+
+    agent: "str | Path"
+    sidecar: "str | Path"
+
+
 class BottleBackend(ABC, Generic[PlanT, CleanupT]):
     """Abstract base for selectable bottle backends. Concrete subclasses
     (e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -435,26 +447,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
+    def prelaunch_checks(self, plan: PlanT) -> None:
+        """Raise StaleImageError if any cached image used by this plan is stale.
+        No-op default; backends override to call the shared check_stale*
+        helpers on their image/artifact timestamps. Called by the CLI before
+        launch so the operator can be prompted outside the launch context."""
+
     @contextmanager
-    def launch(
-        self, plan: PlanT, *, skip_stale: bool = False
-    ) -> Generator[Bottle, None, None]:
-        """Template: optionally check for stale cached images, then delegate
-        to `_launch_impl`. Pass `skip_stale=True` to bypass the stale check
-        (used by the interactive CLI after the operator confirms)."""
-        if not skip_stale:
-            self._image_stale_checks(plan)
-        with self._launch_impl(plan) as bottle:
+    def launch(self, plan: PlanT) -> Generator[Bottle, None, None]:
+        """Template: build or load images, then delegate to _launch_impl."""
+        images = self._build_or_load_images(plan)
+        with self._launch_impl(plan, images) as bottle:
             yield bottle
 
-    def _image_stale_checks(self, plan: PlanT) -> None:
-        """Raise StaleImageError if any cached image used by this plan is stale.
-        No-op default; backends override to call the shared `check_stale*`
-        helpers on their image/artifact timestamps."""
+    @abstractmethod
+    def _build_or_load_images(self, plan: PlanT) -> "BottleImages":
+        """Return the agent and sidecar image references (or artifact paths)
+        for this plan, building fresh images when the policy requires it."""
 
     @abstractmethod
-    def _launch_impl(self, plan: PlanT) -> AbstractContextManager[Bottle]:
-        """Build/run the bottle and yield a handle; tear down on exit."""
+    def _launch_impl(self, plan: PlanT, images: "BottleImages") -> AbstractContextManager[Bottle]:
+        """Bring up the bottle using pre-resolved images; yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy host-side files (CA cert, prompt, skills, .git) into

bot_bottle/backend/docker/backend.py

@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -85,12 +85,15 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
-    def _image_stale_checks(self, plan: DockerBottlePlan) -> None:
+    def prelaunch_checks(self, plan: DockerBottlePlan) -> None:
         _launch.stale_checks(plan)
 
+    def _build_or_load_images(self, plan: DockerBottlePlan) -> BottleImages:
+        return _launch.build_or_load_images(plan)
+
     @contextmanager
-    def _launch_impl(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+    def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:

bot_bottle/backend/docker/launch.py

@@ -43,6 +43,7 @@ from ...git_gate import (
 )
 from ...image_cache import check_stale
 from ...log import die, info, warn
+from .. import BottleImages
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -71,16 +72,50 @@ from .sidecar_bundle import SIDECAR_BUNDLE_IMAGE
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 
 
+def build_or_load_images(plan: DockerBottlePlan) -> BottleImages:
+    """Resolve the agent and sidecar image refs for this plan.
+
+    Returns the committed snapshot if one exists, the local cached images
+    when the policy is 'cached', or builds fresh images and returns those."""
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return BottleImages(agent=committed, sidecar=SIDECAR_BUNDLE_IMAGE)
+    if plan.spec.image_policy == "cached":
+        if not docker_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        if not docker_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
+            die(
+                f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
+        return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+    docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+
+
 @contextmanager
 def launch(
     plan: DockerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
 ) -> Generator[DockerBottle, None, None]:
-    """Build, launch, and provision a Docker bottle via compose.
-    Teardown on exit."""
+    """Launch and provision a Docker bottle via compose. Teardown on exit."""
     stack = ExitStack()
 
+    # Stamp the resolved agent image ref into the plan so compose rendering
+    # picks up the right image (may be a committed snapshot or cached ref).
+    plan = dataclasses.replace(
+        plan,
+        agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
+    )
+
     _bottle_for_revoke = plan.manifest.bottle
     _git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
 
@@ -97,36 +132,6 @@ def launch(
         )
 
     try:
-        # Step 1: agent image. Use a committed snapshot when one exists
-        # and is present in the local daemon; otherwise build from the
-        # Dockerfile. Sidecar images get built lazily by `docker compose
-        # up` via the renderer's `build:` directives.
-        committed = read_committed_image(plan.slug)
-        cached_policy = plan.spec.image_policy == "cached"
-        if committed and docker_mod.image_exists(committed):
-            info(f"using committed image {committed!r}")
-            plan = dataclasses.replace(
-                plan,
-                agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
-            )
-        elif cached_policy:
-            if not docker_mod.image_exists(plan.image):
-                die(
-                    f"cached agent image {plan.image!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            if not docker_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
-                die(
-                    f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            info(f"using cached agent image {plan.image!r}")
-            info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
-        else:
-            docker_mod.build_image(
-                plan.image, _REPO_DIR,
-                dockerfile=plan.dockerfile_path,
-            )
 
         internal_network = network_mod.network_name_for_slug(plan.slug)
         egress_network = network_mod.network_egress_name_for_slug(plan.slug)

bot_bottle/backend/macos_container/backend.py

@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -67,14 +67,17 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
-    def _image_stale_checks(self, plan: MacosContainerBottlePlan) -> None:
+    def prelaunch_checks(self, plan: MacosContainerBottlePlan) -> None:
         _launch.stale_checks(plan)
 
+    def _build_or_load_images(self, plan: MacosContainerBottlePlan) -> BottleImages:
+        return _launch.build_or_load_images(plan)
+
     @contextmanager
     def _launch_impl(
-        self, plan: MacosContainerBottlePlan
+        self, plan: MacosContainerBottlePlan, images: BottleImages
     ) -> Generator[MacosContainerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:

bot_bottle/backend/macos_container/launch.py

@@ -34,6 +34,7 @@ from ...git_gate import (
 )
 from ...image_cache import check_stale
 from ...log import die, info, warn
+from .. import BottleImages
 from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
@@ -72,17 +73,51 @@ def sidecar_container_name(slug: str) -> str:
     return f"bot-bottle-sidecars-{slug}"
 
 
+def build_or_load_images(plan: MacosContainerBottlePlan) -> BottleImages:
+    """Resolve agent and sidecar image refs. Builds the sidecar when needed,
+    but never builds the agent when policy is 'cached'."""
+    committed = read_committed_image(plan.slug)
+    if committed and container_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        if plan.spec.image_policy != "cached":
+            container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
+        return BottleImages(agent=committed, sidecar=SIDECAR_BUNDLE_IMAGE)
+    if plan.spec.image_policy == "cached":
+        if not container_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        if not container_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
+            die(
+                f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
+        return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+    container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
+    container_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+
+
 @contextmanager
 def launch(
     plan: MacosContainerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
 ) -> Generator[MacosContainerBottle, None, None]:
-    """Build, run, provision, and yield an Apple Container bottle."""
+    """Run, provision, and yield an Apple Container bottle."""
     stack = ExitStack()
     bottle_for_revoke = plan.manifest.bottle
     git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
 
+    plan = dataclasses.replace(
+        plan,
+        agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
+    )
+
     def teardown() -> None:
         teardown_exc: BaseException | None = None
         try:
@@ -96,7 +131,6 @@ def launch(
 
     try:
         plan = _mint_certs(plan)
-        plan = _build_images(plan)
 
         internal_network = internal_network_name(plan.slug)
         egress_network = egress_network_name(plan.slug)
@@ -149,35 +183,6 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
     return dataclasses.replace(plan, egress_plan=egress_plan)
 
 
-def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
-    cached = plan.spec.image_policy == "cached"
-    committed = read_committed_image(plan.slug)
-    if committed and container_mod.image_exists(committed):
-        info(f"using committed image {committed!r}")
-        if not cached:
-            container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
-        return dataclasses.replace(
-            plan,
-            agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
-        )
-    if cached:
-        if not container_mod.image_exists(plan.image):
-            die(
-                f"cached agent image {plan.image!r} not found; "
-                "run without --cached-images to build it"
-            )
-        if not container_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
-            die(
-                f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
-                "run without --cached-images to build it"
-            )
-        info(f"using cached agent image {plan.image!r}")
-        info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
-        return plan
-    container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
-    container_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
-    return plan
-
 
 def stale_checks(plan: MacosContainerBottlePlan) -> None:
     """Raise StaleImageError if a cached image is older than the configured

bot_bottle/cli/start.py

@@ -556,38 +556,35 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        skip_stale = False
-        while True:
-            try:
-                with backend.launch(plan, skip_stale=skip_stale) as bottle:
-                    agent_provider_template = getattr(plan, "agent_provider_template", "claude")
-                    extra_args: tuple[str, ...] = ()
-                    if headless_prompt_text:
-                        extra_args = tuple(
-                            get_provider(agent_provider_template).headless_prompt(
-                                headless_prompt_text
-                            )
-                        )
-                    exit_code = attach_agent(
-                        bottle,
-                        agent_provider_template=agent_provider_template,
-                        startup_args=plan.agent_provision.startup_args + extra_args,
+        try:
+            backend.prelaunch_checks(plan)
+        except StaleImageError as exc:
+            if assume_yes:
+                die(str(exc))
+            sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
+            sys.stderr.flush()
+            if read_tty_line() not in ("y", "Y", "yes", "YES"):
+                return 0
+        with backend.launch(plan) as bottle:
+            agent_provider_template = getattr(plan, "agent_provider_template", "claude")
+            extra_args: tuple[str, ...] = ()
+            if headless_prompt_text:
+                extra_args = tuple(
+                    get_provider(agent_provider_template).headless_prompt(
+                        headless_prompt_text
                     )
-                    info(
-                        f"session ended (exit {exit_code}); "
-                        f"container {bottle.name} will be removed"
-                    )
-                    if agent_provider_template == "claude":
-                        capture_claude_session_state(identity, exit_code)
-                break
-            except StaleImageError as exc:
-                if assume_yes:
-                    die(str(exc))
-                sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
-                sys.stderr.flush()
-                if read_tty_line() not in ("y", "Y", "yes", "YES"):
-                    break
-                skip_stale = True
+                )
+            exit_code = attach_agent(
+                bottle,
+                agent_provider_template=agent_provider_template,
+                startup_args=plan.agent_provision.startup_args + extra_args,
+            )
+            info(
+                f"session ended (exit {exit_code}); "
+                f"container {bottle.name} will be removed"
+            )
+            if agent_provider_template == "claude":
+                capture_claude_session_state(identity, exit_code)
         return 0
     finally:
         # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount

tests/unit/test_cli_start_stale.py

@@ -1,9 +1,9 @@
 """Unit: _launch_bottle StaleImageError handling.
 
-Exercises the while-True / try-except loop around backend.launch:
+Exercises prelaunch_checks / backend.launch flow:
   - headless mode → die on stale
   - interactive mode, user declines → stop without launching
-  - interactive mode, user confirms → retry with skip_stale=True
+  - interactive mode, user confirms → skip stale check and launch once
 """
 
 from __future__ import annotations
@@ -28,14 +28,6 @@ def _fake_plan() -> Any:
     ))
 
 
-def _stale_cm() -> MagicMock:
-    """Return a context-manager mock whose __enter__ raises StaleImageError."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(side_effect=StaleImageError("image is 5 day(s) old"))
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
 def _ok_cm(bottle: Any) -> MagicMock:
     """Return a context-manager mock that yields `bottle`."""
     cm = MagicMock()
@@ -78,23 +70,25 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
             )
 
     def test_headless_stale_calls_die(self) -> None:
-        """In headless mode (assume_yes=True), a StaleImageError must call die()."""
+        """In headless mode (assume_yes=True), a StaleImageError from prelaunch_checks must call die()."""
         import bot_bottle.cli.start as start_mod
 
         backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
 
         with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
              patch.object(start_mod, "die", side_effect=Die()):
             with self.assertRaises(Die):
                 self._run_launch(assume_yes=True)
 
-    def test_interactive_user_declines_stops_loop(self) -> None:
-        """Interactive user answering 'n' → loop exits without a second launch."""
+        backend_mock.launch.assert_not_called()
+
+    def test_interactive_user_declines_stops_before_launch(self) -> None:
+        """Interactive user answering 'n' → launch is never called."""
         import bot_bottle.cli.start as start_mod
 
         backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
 
         with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
              patch.object(start_mod, "read_tty_line", return_value="n"), \
@@ -102,23 +96,18 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
             rc = self._run_launch(assume_yes=False)
 
         self.assertEqual(0, rc)
-        # launch was called once (stale), then the user declined → no second call.
-        backend_mock.launch.assert_called_once()
+        backend_mock.launch.assert_not_called()
 
-    def test_interactive_user_confirms_retries_with_skip_stale(self) -> None:
-        """Interactive user answering 'y' → second launch with skip_stale=True."""
+    def test_interactive_user_confirms_launches_once(self) -> None:
+        """Interactive user answering 'y' → prelaunch stale error is bypassed; launch called once."""
         import bot_bottle.cli.start as start_mod
 
         bottle_mock = MagicMock()
         bottle_mock.name = "dev-abc"
 
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
         backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+        backend_mock.launch.return_value = _ok_cm(bottle_mock)
 
         with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
              patch.object(start_mod, "read_tty_line", return_value="y"), \
@@ -128,11 +117,8 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
             rc = self._run_launch(assume_yes=False)
 
         self.assertEqual(0, rc)
-        # First call: skip_stale=False → stale. Second call: skip_stale=True → ok.
-        self.assertEqual(2, backend_mock.launch.call_count)
-        calls = backend_mock.launch.call_args_list
-        self.assertFalse(calls[0].kwargs.get("skip_stale", False))
-        self.assertTrue(calls[1].kwargs.get("skip_stale", False))
+        backend_mock.prelaunch_checks.assert_called_once()
+        backend_mock.launch.assert_called_once()
 
     def test_interactive_yes_uppercase_also_accepted(self) -> None:
         """'Y' or 'YES' should also be accepted as confirmation."""
@@ -141,13 +127,9 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
         bottle_mock = MagicMock()
         bottle_mock.name = "dev-abc"
 
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
         backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+        backend_mock.launch.return_value = _ok_cm(bottle_mock)
 
         with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
              patch.object(start_mod, "read_tty_line", return_value="YES"), \
@@ -157,7 +139,7 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
             rc = self._run_launch(assume_yes=False)
 
         self.assertEqual(0, rc)
-        self.assertEqual(2, backend_mock.launch.call_count)
+        backend_mock.launch.assert_called_once()
 
 
 if __name__ == "__main__":

tests/unit/test_docker_launch_committed_image.py

@@ -12,7 +12,7 @@ from typing import Any
 from unittest import mock
 
 from bot_bottle.agent_provider import AgentProvisionPlan
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import BottleImages, BottleSpec
 from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
@@ -90,7 +90,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
         committed_tag: str | None = None,
         image_present: bool = True,
     ) -> list[str]:
-        """Drive launch() through its full sequence with the committed-image
+        """Drive build_or_load_images() + launch() with the committed-image
         behaviour controlled by the arguments. Returns the images that were
         passed to `build_image` (empty list if it was never called)."""
         built: list[str] = []
@@ -124,8 +124,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
+            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         return built
@@ -170,8 +171,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
+            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         self.assertEqual(1, len(captured_plans))

tests/unit/test_docker_launch_teardown.py

@@ -16,7 +16,7 @@ from pathlib import Path
 from unittest import mock
 
 from bot_bottle.agent_provider import AgentProvisionPlan
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import BottleImages, BottleSpec
 from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
@@ -86,6 +86,8 @@ class TestTeardownWarning(unittest.TestCase):
         plan = _plan(self._tmp)
         buf = io.StringIO()
 
+        images = BottleImages(agent="bot-bottle-claude:latest", sidecar="bot-bottle-sidecars:latest")
+
         with mock.patch.object(launch_mod.docker_mod, "build_image"), \
              mock.patch.object(
                  launch_mod, "egress_tls_init",
@@ -115,7 +117,7 @@ class TestTeardownWarning(unittest.TestCase):
              ), \
              contextlib.redirect_stderr(buf):
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         output = buf.getvalue()

tests/unit/test_macos_container_launch.py

@@ -11,7 +11,7 @@ from typing import cast
 from unittest.mock import patch
 
 from bot_bottle.agent_provider import AgentProvisionPlan
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import BottleImages, BottleSpec
 from bot_bottle.backend.macos_container import launch
 from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
 from bot_bottle.egress import EgressPlan
@@ -339,9 +339,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ), patch.object(launch, "info"):
-            updated = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", updated.image)
+        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
         self.assertEqual(1, len(calls))
         self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0])
 
@@ -360,9 +360,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ):
-            updated = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
-        self.assertEqual("bot-bottle-agent:latest", updated.image)
+        self.assertEqual("bot-bottle-agent:latest", images.agent)
         self.assertEqual(2, len(calls))
         self.assertEqual("bot-bottle-agent:latest", calls[1][0])
 
@@ -395,7 +395,7 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
             launch, "die", side_effect=Die(),
         ):
             with self.assertRaises(Die):
-                launch._build_images(plan)
+                launch.build_or_load_images(plan)
 
     def test_cached_mode_dies_when_sidecar_image_missing(self) -> None:
         plan = self._cached_plan()
@@ -411,7 +411,7 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
             launch, "die", side_effect=Die(),
         ):
             with self.assertRaises(Die):
-                launch._build_images(plan)
+                launch.build_or_load_images(plan)
 
     def test_cached_mode_both_present_returns_unchanged_plan(self) -> None:
         plan = self._cached_plan()
@@ -422,10 +422,10 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image",
         ) as build, patch.object(launch, "info"):
-            result = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
         build.assert_not_called()
-        self.assertEqual(plan.image, result.image)
+        self.assertEqual(plan.image, images.agent)
 
     def test_committed_image_plus_cached_skips_sidecar_build(self) -> None:
         plan = self._cached_plan()
@@ -437,11 +437,11 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image",
         ) as build, patch.object(launch, "info"):
-            result = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
         # In cached mode with a committed image, no builds should fire.
         build.assert_not_called()
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", result.image)
+        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
 
 
 if __name__ == "__main__":

tests/unit/test_stale_checks.py

@@ -1,12 +1,11 @@
 """Unit: stale-image check functions across backends, and the
-BottleBackend.launch template method (skip_stale flag).
+BottleBackend.launch template method (prelaunch_checks + build_or_load_images).
 
 No real images or containers are used — all Docker/container/smolmachine
 calls are mocked at the module boundary."""
 
 from __future__ import annotations
 
-import tempfile
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
@@ -23,51 +22,41 @@ def _bottle_cm(bottle: Any) -> MagicMock:
 
 
 # ---------------------------------------------------------------------------
-# BottleBackend.launch template — _image_stale_checks + skip_stale
+# BottleBackend.launch template — prelaunch_checks + _build_or_load_images
 # ---------------------------------------------------------------------------
 
 class TestBottleBackendLaunchTemplate(unittest.TestCase):
     """Verify the concrete launch() method on BottleBackend calls
-    _image_stale_checks unless skip_stale=True, then delegates to _launch_impl."""
+    _build_or_load_images and _launch_impl, and that prelaunch_checks is a no-op
+    on the base class."""
 
     def _make_backend(self) -> Any:
         from bot_bottle.backend.docker.backend import DockerBottleBackend
         return DockerBottleBackend()
 
-    def test_stale_checks_called_by_default(self) -> None:
+    def test_launch_delegates_to_build_or_load_and_launch_impl(self) -> None:
+        from bot_bottle.backend import BottleImages
         backend = self._make_backend()
         plan = cast(Any, SimpleNamespace())
         bottle = MagicMock()
+        images = BottleImages(agent="agent:latest", sidecar="sidecar:latest")
         with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
+            backend, "_build_or_load_images", return_value=images,
+        ) as build_mock, patch.object(
             backend, "_launch_impl",
             return_value=_bottle_cm(bottle),
-        ):
+        ) as impl_mock:
             with backend.launch(plan):
                 pass
-        stale_mock.assert_called_once_with(plan)
+        build_mock.assert_called_once_with(plan)
+        impl_mock.assert_called_once_with(plan, images)
 
-    def test_skip_stale_bypasses_stale_checks(self) -> None:
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ):
-            with backend.launch(plan, skip_stale=True):
-                pass
-        stale_mock.assert_not_called()
-
-    def test_noop_default_image_stale_checks(self) -> None:
+    def test_noop_default_prelaunch_checks(self) -> None:
         from bot_bottle.backend.docker.backend import DockerBottleBackend
         from bot_bottle.backend import BottleBackend
         backend = DockerBottleBackend()
-        # Call the base-class _image_stale_checks directly to verify it's a no-op.
-        BottleBackend._image_stale_checks(backend, cast(Any, SimpleNamespace()))  # type: ignore[arg-type]
+        # Base-class prelaunch_checks is a no-op — must not raise.
+        BottleBackend.prelaunch_checks(backend, cast(Any, SimpleNamespace()))  # type: ignore[arg-type]
 
 
 # ---------------------------------------------------------------------------
@@ -142,98 +131,13 @@ class TestDockerStaleChecks(unittest.TestCase):
         self.assertEqual(1, cs.call_count)
         self.assertIn(plan.image, cs.call_args.args[0])
 
-    def test_backend_image_stale_checks_delegates(self) -> None:
+    def test_backend_prelaunch_checks_delegates(self) -> None:
         from bot_bottle.backend.docker.backend import DockerBottleBackend
         from bot_bottle.backend.docker import launch as mod
         backend = DockerBottleBackend()
         plan = self._plan()
         with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# smolmachines backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestSmolmachinesStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached") -> Any:
-        return cast(Any, SimpleNamespace(
-            spec=SimpleNamespace(image_policy=policy),
-            slug="dev-abc",
-            agent_image="bot-bottle-claude:latest",
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale_path") as csp:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        csp.assert_not_called()
-
-    def test_committed_artifact_found_checks_it(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f:
-            committed_path = f.name
-            with patch.object(mod, "read_committed_image", return_value=committed_path), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=False), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        csp.assert_called_once()
-        self.assertIn("agent smolmachine", csp.call_args.args[0])
-
-    def test_no_committed_agent_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        digest = "abcdef0123456789"
-        with tempfile.TemporaryDirectory() as tmp:
-            artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
-            artifact.write_text("")
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 patch.object(mod, "read_committed_image", return_value=""), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=True), \
-                 patch.object(
-                     mod.docker_mod, "image_id",
-                     return_value=f"sha256:{digest}ffffffffffffffff",
-                 ), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        # At least the agent artifact check ran.
-        self.assertGreaterEqual(csp.call_count, 1)
-
-    def test_sidecar_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        sidecar_digest = "fedcba9876543210"
-        with tempfile.TemporaryDirectory() as tmp:
-            sidecar_artifact = (
-                Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine"
-            )
-            sidecar_artifact.write_text("")
-
-            def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
-                return True
-
-            def image_id(ref: str) -> str:
-                if ref == mod._bundle.SIDECAR_BUNDLE_IMAGE:  # type: ignore[attr-defined]
-                    return f"sha256:{sidecar_digest}ffffffff"
-                return "sha256:0000000000000000ffffffff"
-
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 patch.object(mod, "read_committed_image", return_value=""), \
-                 patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \
-                 patch.object(mod.docker_mod, "image_id", side_effect=image_id), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        sidecar_calls = [c for c in csp.call_args_list if "sidecar" in c.args[0]]
-        self.assertGreaterEqual(len(sidecar_calls), 1)
-
-    def test_backend_image_stale_checks_delegates(self) -> None:
-        from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend
-        from bot_bottle.backend.smolmachines import launch as mod
-        backend = SmolmachinesBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
+            backend.prelaunch_checks(plan)  # type: ignore[arg-type]
         sc.assert_called_once_with(plan)
 
 
@@ -288,13 +192,13 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
             mod.stale_checks(self._plan())
         cs.assert_not_called()
 
-    def test_backend_image_stale_checks_delegates(self) -> None:
+    def test_backend_prelaunch_checks_delegates(self) -> None:
         from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
         from bot_bottle.backend.macos_container import launch as mod
         backend = MacosContainerBottleBackend()
         plan = self._plan()
         with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
+            backend.prelaunch_checks(plan)  # type: ignore[arg-type]
         sc.assert_called_once_with(plan)