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

@@ -37,10 +37,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 TYPE_CHECKING, Any, Generic, Sequence, TypeVar
+from typing import TYPE_CHECKING, Any, Generator, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -439,8 +439,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

@@ -100,8 +100,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

@@ -42,7 +42,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 util as docker_mod
 from .bottle import DockerBottle
@@ -104,11 +104,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),
@@ -120,10 +115,6 @@ def launch(
                     "run without --cached-images to build it"
                 )
             info(f"using cached agent image {plan.image!r}")
-            warn_if_stale(
-                f"agent image {plan.image!r}",
-                docker_mod.image_created_at(plan.image),
-            )
         else:
             docker_mod.build_image(
                 plan.image, _REPO_DIR,
@@ -225,3 +216,17 @@ 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
+    if docker_mod.image_exists(plan.image):
+        check_stale(f"agent image {plan.image!r}", docker_mod.image_created_at(plan.image))

bot_bottle/backend/macos_container/backend.py

@@ -82,8 +82,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

@@ -53,6 +53,7 @@ from ...git_gate import (
     revoke_git_gate_provisioned_keys,
 )
 from ...git_http_backend import DEFAULT_PORT as _GIT_HTTP_PORT
+from ...image_cache import check_stale
 from ...log import die, info, warn
 from ...supervise import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
@@ -180,6 +181,7 @@ def launch(
 def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
     """Build the agent image. The gateway's own image is built by
     `ensure_gateway` — it belongs to the shared singleton, not to a bottle."""
+    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}")
@@ -189,12 +191,34 @@ def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
                 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"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        return plan
     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
+    if container_mod.image_exists(plan.image):
+        check_stale(f"agent image {plan.image!r}", container_mod.image_created_at(plan.image))
+
+
 def _provision_git_gate_keys(
     plan: MacosContainerBottlePlan,
 ) -> MacosContainerBottlePlan:

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
@@ -611,6 +612,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/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
@@ -571,30 +572,38 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        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
+        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,
                     )
-                )
-            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"
-            )
-            # While the container is still alive: always snapshot the
-            # transcript and — if the agent exited non-zero — mark
-            # the state for preservation. This picks up crashes /
-            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
-            if agent_provider_template == "claude":
-                capture_claude_session_state(identity, exit_code)
+                    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
         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:
-    warn_if_stale(
-        label,
-        datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc),
-    )
+def check_stale_path(label: str, path: Path) -> None:
+    """Raise StaleImageError if `path`'s mtime exceeds the staleness threshold."""
+    check_stale(label, 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

@@ -95,7 +95,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
                 mock.patch.object(launch_mod.docker_mod, "build_image", side_effect=_build), \
                 mock.patch.object(launch_mod.docker_mod, "verify_agent_image"), \
                 mock.patch.object(launch_mod.docker_mod, "image_created_at"), \
-                mock.patch.object(launch_mod, "warn_if_stale"), \
                 mock.patch.object(launch_mod, "launch_consolidated", return_value=_CTX), \
                 mock.patch.object(launch_mod, "teardown_consolidated"), \
                 mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \