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)