refactor(image-cache): replace warn_if_stale with StaleImageError; add launch template

- image_cache: StaleImageError exception + check_stale/check_stale_path (raise instead of warn)
- BottleBackend.launch: template method (skip_stale flag) that calls _image_stale_checks then _launch_impl
- Each backend: _image_stale_checks delegates to a stale_checks() function in its launch module; _launch_impl replaces launch override
- macos_container: adds image_created_at to util, cached-image support in _build_images, stale_checks
- cli/start.py: catches StaleImageError, prompts interactively, retries with skip_stale=True; headless mode dies on it

bot_bottle/backend/__init__.py

@@ -36,10 +36,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager
+from contextlib import AbstractContextManager, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generic, Sequence, TypeVar
+from typing import Any, Generator, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -435,8 +435,25 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
+    @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:
+            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 launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
+    def _launch_impl(self, plan: PlanT) -> AbstractContextManager[Bottle]:
         """Build/run the bottle and yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:

bot_bottle/backend/docker/backend.py

@@ -85,8 +85,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: DockerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
+    def _launch_impl(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 

bot_bottle/backend/docker/launch.py

@@ -41,7 +41,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import warn_if_stale
+from ...image_cache import check_stale
 from ...log import die, info, warn
 from . import network as network_mod
 from . import util as docker_mod
@@ -105,11 +105,6 @@ def launch(
         cached_policy = plan.spec.image_policy == "cached"
         if committed and docker_mod.image_exists(committed):
             info(f"using committed image {committed!r}")
-            if cached_policy:
-                warn_if_stale(
-                    f"agent image {committed!r}",
-                    docker_mod.image_created_at(committed),
-                )
             plan = dataclasses.replace(
                 plan,
                 agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
@@ -127,14 +122,6 @@ def launch(
                 )
             info(f"using cached agent image {plan.image!r}")
             info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
-            warn_if_stale(
-                f"agent image {plan.image!r}",
-                docker_mod.image_created_at(plan.image),
-            )
-            warn_if_stale(
-                f"sidecar image {SIDECAR_BUNDLE_IMAGE!r}",
-                docker_mod.image_created_at(SIDECAR_BUNDLE_IMAGE),
-            )
         else:
             docker_mod.build_image(
                 plan.image, _REPO_DIR,
@@ -236,3 +223,21 @@ def launch(
         yield bottle
     finally:
         teardown()
+
+
+def stale_checks(plan: DockerBottlePlan) -> None:
+    """Raise StaleImageError if a cached image is older than the configured
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        check_stale(f"agent image {committed!r}", docker_mod.image_created_at(committed))
+        return
+    for label, ref in [
+        ("agent image", plan.image),
+        ("sidecar image", SIDECAR_BUNDLE_IMAGE),
+    ]:
+        if docker_mod.image_exists(ref):
+            check_stale(f"{label} {ref!r}", docker_mod.image_created_at(ref))

bot_bottle/backend/macos_container/backend.py

@@ -67,8 +67,11 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: MacosContainerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
+    def _launch_impl(
         self, plan: MacosContainerBottlePlan
     ) -> Generator[MacosContainerBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/macos_container/launch.py

@@ -32,6 +32,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
+from ...image_cache import check_stale
 from ...log import die, info, warn
 from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
@@ -149,29 +150,53 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
 
 
 def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
-    container_mod.build_image(
+    cached = plan.spec.image_policy == "cached"
-        SIDECAR_BUNDLE_IMAGE,
-        _REPO_DIR,
-        dockerfile=SIDECAR_BUNDLE_DOCKERFILE,
-    )
     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(
+            agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
-                plan.agent_provision,
-                image=committed,
-            ),
         )
-    container_mod.build_image(
+    if cached:
-        plan.image,
+        if not container_mod.image_exists(plan.image):
-        _REPO_DIR,
+            die(
-        dockerfile=plan.dockerfile_path,
+                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
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed and container_mod.image_exists(committed):
+        check_stale(f"agent image {committed!r}", container_mod.image_created_at(committed))
+        return
+    for label, ref in [
+        ("agent image", plan.image),
+        ("sidecar image", SIDECAR_BUNDLE_IMAGE),
+    ]:
+        if container_mod.image_exists(ref):
+            check_stale(f"{label} {ref!r}", container_mod.image_created_at(ref))
+
+
 def _create_networks(
     internal_network: str,
     egress_network: str,

bot_bottle/backend/macos_container/util.py

@@ -10,6 +10,7 @@ import shutil
 import subprocess
 import tempfile
 import time
+from datetime import datetime, timezone
 from typing import Iterable
 
 from ...log import die, info
@@ -458,6 +459,39 @@ def image_id(ref: str) -> str:
     raise AssertionError("unreachable")
 
 
+def image_created_at(ref: str) -> datetime:
+    """Return the image creation timestamp as an aware UTC datetime.
+
+    Parses the `created` field from `container image inspect` JSON output."""
+    result = subprocess.run(
+        [_CONTAINER, "image", "inspect", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        die(
+            f"container image inspect for {ref!r} failed: "
+            f"{(result.stderr or '').strip() or '<no stderr>'}"
+        )
+    try:
+        data = json.loads(result.stdout or "{}")
+    except json.JSONDecodeError as exc:
+        die(f"container image inspect for {ref!r} returned malformed JSON: {exc}")
+    if isinstance(data, list) and data:
+        data = data[0]
+    if isinstance(data, dict):
+        value = data.get("created") or data.get("Created")
+        if isinstance(value, str) and value:
+            try:
+                ts = value.rstrip("Z")
+                return datetime.fromisoformat(ts).replace(tzinfo=timezone.utc)
+            except ValueError:
+                pass
+    die(f"container image inspect for {ref!r} did not include a creation timestamp")
+    raise AssertionError("unreachable")
+
+
 def save(ref: str, output: str) -> None:
     subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True)
 

bot_bottle/backend/smolmachines/backend.py

@@ -77,8 +77,11 @@ class SmolmachinesBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: SmolmachinesBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
+    def _launch_impl(
         self, plan: SmolmachinesBottlePlan
     ) -> Generator[SmolmachinesBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/smolmachines/launch.py

@@ -45,7 +45,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import warn_if_stale_path
+from ...image_cache import check_stale_path
 from ...log import die, info, warn
 from ...bottle_state import (
     egress_state_dir,
@@ -471,8 +471,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
         committed_path = Path(committed)
         if committed_path.is_file():
             info(f"using committed smolmachine {str(committed_path)!r}")
-            if _image_policy(plan) == "cached":
-                warn_if_stale_path("agent smolmachine artifact", committed_path)
             return committed_path
 
     if _image_policy(plan) == "cached":
@@ -488,6 +486,32 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
     )
 
 
+def stale_checks(plan: SmolmachinesBottlePlan) -> None:
+    """Raise StaleImageError if a cached smolmachine artifact is older than the
+    configured threshold. Only runs when image_policy is 'cached'. Checks the
+    committed agent artifact first, then the cache-keyed agent and sidecar
+    artifacts. Called by the backend class's _image_stale_checks before any
+    resources are allocated."""
+    if _image_policy(plan) != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed:
+        committed_path = Path(committed)
+        if committed_path.is_file():
+            check_stale_path("agent smolmachine artifact", committed_path)
+    elif docker_mod.image_exists(plan.agent_image):
+        digest = docker_mod.image_id(plan.agent_image).split(":", 1)[-1][:16]
+        artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
+        if artifact.is_file():
+            check_stale_path("agent smolmachine artifact", artifact)
+    sidecar_image = _bundle.SIDECAR_BUNDLE_IMAGE
+    if docker_mod.image_exists(sidecar_image):
+        digest = docker_mod.image_id(sidecar_image).split(":", 1)[-1][:16]
+        sidecar_artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
+        if sidecar_artifact.is_file():
+            check_stale_path("sidecar smolmachine artifact", sidecar_artifact)
+
+
 def _image_policy(plan: object) -> str:
     spec = getattr(plan, "spec", None)
     return str(getattr(spec, "image_policy", "fresh"))
@@ -513,7 +537,6 @@ def _cached_smolmachine(image_ref: str, *, label: str) -> Path:
             "run without --cached-images to build it"
         )
     info(f"using cached {label} smolmachine {str(sidecar)!r}")
-    warn_if_stale_path(f"{label} smolmachine artifact", sidecar)
     return sidecar
 
 

bot_bottle/cli/start.py

@@ -36,6 +36,7 @@ from ..bottle_state import (
     is_preserved,
     mark_preserved,
 )
+from ..image_cache import StaleImageError
 from ..log import info, die
 from ..manifest import Manifest, ManifestIndex
 from ._common import PROG, USER_CWD, read_tty_line
@@ -555,30 +556,38 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        with backend.launch(plan) as bottle:
+        skip_stale = False
-            agent_provider_template = getattr(plan, "agent_provider_template", "claude")
+        while True:
-            extra_args: tuple[str, ...] = ()
+            try:
-            if headless_prompt_text:
+                with backend.launch(plan, skip_stale=skip_stale) as bottle:
-                extra_args = tuple(
+                    agent_provider_template = getattr(plan, "agent_provider_template", "claude")
-                    get_provider(agent_provider_template).headless_prompt(
+                    extra_args: tuple[str, ...] = ()
-                        headless_prompt_text
+                    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,
                     )
-                )
+                    info(
-            exit_code = attach_agent(
+                        f"session ended (exit {exit_code}); "
-                bottle,
+                        f"container {bottle.name} will be removed"
-                agent_provider_template=agent_provider_template,
+                    )
-                startup_args=plan.agent_provision.startup_args + extra_args,
+                    if agent_provider_template == "claude":
-            )
+                        capture_claude_session_state(identity, exit_code)
-            info(
+                break
-                f"session ended (exit {exit_code}); "
+            except StaleImageError as exc:
-                f"container {bottle.name} will be removed"
+                if assume_yes:
-            )
+                    die(str(exc))
-            # While the container is still alive: always snapshot the
+                sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
-            # transcript and — if the agent exited non-zero — mark
+                sys.stderr.flush()
-            # the state for preservation. This picks up crashes /
+                if read_tty_line() not in ("y", "Y", "yes", "YES"):
-            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
+                    break
-            if agent_provider_template == "claude":
+                skip_stale = True
-                capture_claude_session_state(identity, exit_code)
         return 0
     finally:
         # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount

bot_bottle/image_cache.py

@@ -1,4 +1,4 @@
-"""Shared helpers for cached-image quickstart warnings."""
+"""Shared helpers for cached-image quickstart stale checks."""
 
 from __future__ import annotations
 
@@ -7,13 +7,19 @@ from pathlib import Path
 
 try:
     from .config_store import ConfigStore
-    from .log import warn
 except ImportError:
     from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
-    from log import warn  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 
 
-def warn_if_stale(label: str, created_at: datetime) -> None:
+class StaleImageError(Exception):
+    """Raised when a cached image or artifact exceeds the configured staleness
+    threshold. Callers can catch this to prompt interactively; headless paths
+    let it propagate as a fatal error."""
+
+
+def check_stale(label: str, created_at: datetime) -> None:
+    """Raise StaleImageError if `created_at` is older than the configured
+    stale-warning threshold. Negative threshold disables the check."""
     threshold_days = ConfigStore().cached_image_stale_warning_days()
     if threshold_days < 0:
         return
@@ -22,17 +28,15 @@ def warn_if_stale(label: str, created_at: datetime) -> None:
     age = now - created
     if age.total_seconds() <= threshold_days * 86400:
         return
-    warn(
+    raise StaleImageError(
         f"cached {label} is {age.days} day(s) old; "
         "quickstart does not verify it matches the current Dockerfile/context"
     )
 
 
-def warn_if_stale_path(label: str, path: Path) -> None:
+def check_stale_path(label: str, path: Path) -> None:
-    warn_if_stale(
+    """Raise StaleImageError if `path`'s mtime exceeds the staleness threshold."""
-        label,
+    check_stale(label, datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc))
-        datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc),
-    )
 
 
-__all__ = ["warn_if_stale", "warn_if_stale_path"]
+__all__ = ["StaleImageError", "check_stale", "check_stale_path"]

tests/unit/test_docker_launch_committed_image.py

@@ -105,10 +105,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
             launch_mod.docker_mod, "image_exists", return_value=image_present,
         ), mock.patch.object(
             launch_mod.docker_mod, "build_image", side_effect=fake_build,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "image_created_at",
-        ), mock.patch.object(
-            launch_mod, "warn_if_stale",
         ), mock.patch.object(
             launch_mod, "egress_tls_init",
             return_value=(Path("/egress_ca"), Path("/egress_cert")),

tests/unit/test_macos_container_launch.py

@@ -81,7 +81,7 @@ def _plan(
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(),
+        spec=SimpleNamespace(image_policy="fresh"),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
@@ -292,7 +292,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
 
 def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
     return MacosContainerBottlePlan(
-        spec=cast(BottleSpec, SimpleNamespace()),
+        spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),

tests/unit/test_smolmachines_launch_image.py

@@ -206,8 +206,6 @@ class TestAgentFromPath(unittest.TestCase):
             ), patch.object(
                 _launch_mod.docker_mod, "image_id",
                 return_value=f"sha256:{digest}fffffffffffffffff",
-            ), patch.object(
-                _launch_mod, "warn_if_stale_path",
             ), patch.object(
                 _launch_mod, "_ensure_smolmachine",
             ) as ensure: