fix(firecracker): honor cached image policy

- config_store.py imported host_db_path from supervise_types (wrong);
  it lives in paths.py, matching all other stores (audit, queue, etc.)
- test_stale_checks: two tests expected check_stale called twice
  (agent + sidecar) — consolidated arch has no sidecar, so once is
  correct; update assertions and remove unused Path import

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitea/workflows/test.yml

@@ -4,15 +4,16 @@
 # dependencies are required to execute it. Tests are split by directory:
 #
 #   tests/unit/         — pure unit tests; always run
-#   tests/integration/  — need a reachable backend; skip cleanly when
+#   tests/integration/  — need a reachable Docker daemon; skip cleanly
-#                         the backend isn't available on the runner
+#                         (via tests/_docker.py:skip_unless_docker) when
+#                         Docker isn't available on the runner
 #   tests/canaries/     — upstream regression canaries; run on a separate
 #                         schedule (see canaries.yml), not here
 #
-# Integration tests run once per backend in separate jobs. Each job sets
+# This workflow assumes the Gitea Actions runner exposes the host Docker
-# BOT_BOTTLE_BACKEND explicitly so the test suite uses the right backend.
+# socket to the job container so `docker` commands inside the job can
-# Backends that aren't available on the runner fail the preflight step
+# reach the daemon. If that's not yet configured on the runner the
-# rather than silently skipping inside the test output.
+# integration tests will skip rather than fail.
 
 name: test
 
@@ -22,16 +23,9 @@ on:
       - main
     paths:
       - '**.py'
-      - '.gitea/workflows/**.yml'
-      - 'scripts/**'
-      - 'README.md'
   pull_request:
     paths:
       - '**.py'
-      - '.gitea/workflows/**.yml'
-      - 'scripts/**'
-      - 'README.md'
-  workflow_dispatch:
 
 jobs:
   unit:
@@ -54,7 +48,7 @@ jobs:
       - name: Report unit coverage
         run: python3 -m coverage report -m
 
-  integration-docker:
+  integration:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
@@ -71,96 +65,33 @@ jobs:
             echo "docker not on PATH — integration tests will skip"
           fi
 
-      - name: Run integration tests (docker)
+      - name: Run integration tests
-        env:
-          BOT_BOTTLE_BACKEND: docker
         run: python3 -m unittest discover -t . -s tests/integration -v
 
-  # Integration tests against the Firecracker backend. Runs on a self-hosted
+  # Combined unit+integration coverage report (informational). See
-  # KVM runner (label `kvm`) where /dev/kvm and the TAP/nft pool are available.
+  # docs/decisions/0004-coverage-policy.md.
   #
-  # Restricted to same-repo PRs, push to main, and workflow_dispatch — fork
+  # The hard diff-coverage gate (changed lines >= 90%) is DEFERRED: the
-  # PRs don't execute untrusted code on the privileged runner.
+  # Firecracker backend's VM/SSH orchestration is covered by the integration
-  #
+  # suite, which needs /dev/kvm + the provisioned TAP/nft pool — a
-  # Runner prerequisites (provision once; see README "Firecracker on Linux"):
+  # container-based runner skips it and those lines read uncovered, so the
-  # `firecracker` on PATH, `/dev/kvm` accessible, Docker, cached kernel +
+  # gate can't pass here. Re-enabling it on a self-hosted KVM runner is
-  # static dropbear, and the pool as a persistent systemd unit.
+  # tracked separately (see PRD 0069 / #348 and the ci-runner branch).
-  integration-firecracker:
-    runs-on: [self-hosted, kvm]
-    if: >-
-      github.event_name == 'push' ||
-      github.event_name == 'workflow_dispatch' ||
-      (github.event_name == 'pull_request' &&
-       github.event.pull_request.head.repo.full_name == github.repository)
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v4
-
-      - name: Preflight — Firecracker host is ready
-        run: |
-          command -v firecracker >/dev/null || {
-            echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
-          test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
-          # `backend status` exits non-zero unless the TAP pool is up + no
-          # range overlap; it prints the exact `backend setup` fix.
-          python3 cli.py backend status --backend=firecracker
-
-      # No dev-requirements install: the integration suite runs on stdlib
-      # `unittest` (pylint/pyright are lint.yml's concern, not this job's),
-      # and the self-hosted runner's Nix python env has no `pip` module
-      # (`python3 -m pip` → "No module named pip"). Nothing to install.
-      - name: Run integration tests (firecracker)
-        env:
-          BOT_BOTTLE_BACKEND: firecracker
-        run: python3 -m unittest discover -t . -s tests/integration -v
-
-  # Combined unit+integration coverage + the diff-coverage gate (the hard
-  # gate: new/changed lines >= 90%). See docs/decisions/0004-coverage-policy.md.
-  #
-  # This runs on a self-hosted KVM runner (label `kvm`), NOT ubuntu-latest,
-  # because the Firecracker backend's subprocess/VM orchestration
-  # (launch/boot/SSH/isolation-probe) is covered by the integration suite,
-  # and that suite needs `/dev/kvm` + the provisioned TAP/nft pool — which a
-  # container-based runner doesn't have. On such a runner the firecracker
-  # integration test skips and its ~230 orchestration lines read as
-  # uncovered, so the gate can't pass there.
-  #
-  # Restricted to the same events as integration-firecracker (same-repo PRs,
-  # push, workflow_dispatch) for the same security reason.
-  #
-  # See #414 for the planned follow-up: artifact-based coverage combination
-  # (run tests once in their respective jobs, combine .coverage files here).
   coverage:
-    timeout-minutes: 15
+    runs-on: ubuntu-latest
-    runs-on: [self-hosted, kvm]
-    if: >-
-      github.event_name == 'push' ||
-      github.event_name == 'workflow_dispatch' ||
-      (github.event_name == 'pull_request' &&
-       github.event.pull_request.head.repo.full_name == github.repository)
     steps:
       - name: Checkout
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
-      - name: Preflight — Firecracker host is ready
+      # No actions/setup-python: the runner image already ships Python 3.12,
-        run: |
+      # and older act_runner engines mishandle setup-python's PATH (coverage
-          command -v firecracker >/dev/null || {
+      # lands in one interpreter, `python3` resolves to another). Install
-            echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
+      # straight into the ephemeral job container's system Python —
-          test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
+      # --break-system-packages is safe because the container is disposable.
-          # `backend status` exits non-zero unless the TAP pool is up + no
+      - name: Install dev requirements
-          # range overlap; it prints the exact `backend setup` fix.
+        run: python3 -m pip install --break-system-packages -r requirements-dev.txt
-          python3 cli.py backend status --backend=firecracker
 
-      # No dev-requirements install: `coverage` is already provided by the
+      - name: Combined coverage report (unit + integration)
-      # self-hosted runner's Nix python env, and that env has no `pip`
-      # module to install into anyway. `scripts/coverage.sh` +
-      # `diff_coverage.py` need only `coverage` (not pylint/pyright).
-      - name: Combined coverage (unit + integration, incl. firecracker)
         run: PYTHON=python3 bash scripts/coverage.sh critical
-
-      - name: Diff-coverage gate (changed lines >= 90%)
-        run: |
-          git fetch --no-tags origin main:refs/remotes/origin/main
-          python3 scripts/diff_coverage.py --base origin/main --min 90

README.md

@@ -90,8 +90,6 @@ BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
 
 > **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`.
 
-> **CI:** the coverage gate (`.gitea/workflows/test.yml` → `coverage` job) runs on a self-hosted runner labelled `kvm`, because the Firecracker backend's VM/SSH orchestration is exercised only by the integration suite, which needs `/dev/kvm` + the provisioned pool (a container runner would skip it and read as uncovered). Provision that runner exactly like a normal Firecracker host — `firecracker` on `PATH`, `/dev/kvm`, Docker, the cached guest kernel + static dropbear, and the pool installed as the persistent systemd unit — then register it with the `kvm` label. The unit/lint jobs still run on `ubuntu-latest`.
-
 ```sh
 ./cli.py start <agent>   # builds the image on first run, drops you into claude
 ```

bot_bottle/audit_store.py

@@ -42,7 +42,7 @@ class AuditStore(DbStore):
         super().__init__(db_path or host_db_path(), migrations)
 
     def write_audit_entry(self, entry: AuditEntry) -> Path:
-        with self._connection() as conn:
+        with self._connect() as conn:
             conn.execute(
                 """
                 INSERT INTO supervise_audit_entries (
@@ -66,7 +66,7 @@ class AuditStore(DbStore):
     def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]:
         if not self.db_path.is_file():
             return []
-        with self._connection() as conn:
+        with self._connect() as conn:
             rows = conn.execute(
                 """
                 SELECT * FROM supervise_audit_entries

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
@@ -82,6 +82,9 @@ class BottleSpec:
     # True when launched via --headless (no TTY, no interactive prompts).
     # The git-gate host-key preflight uses this to error rather than prompt.
     headless: bool = False
+    # Image startup policy. "fresh" preserves the normal build path;
+    # "cached" reuses the current local image/artifact without rebuilding.
+    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -277,6 +280,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.
@@ -436,9 +451,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) -> 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
+
     @abstractmethod
-    def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
+    def _build_or_load_images(self, plan: PlanT) -> BottleImages:
-        """Build/run the bottle and yield a handle; tear down on exit."""
+        """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, 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
@@ -100,9 +100,15 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
+    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(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
+    def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def ensure_orchestrator(self) -> str:

bot_bottle/backend/docker/launch.py

@@ -42,7 +42,9 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale
+from ...log import die, info, warn
+from .. import BottleImages
 from . import util as docker_mod
 from .bottle import DockerBottle
 from .bottle_plan import DockerBottlePlan
@@ -70,16 +72,47 @@ from ...orchestrator.gateway import DockerGateway
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 
 
+def build_or_load_images(plan: DockerBottlePlan) -> BottleImages:
+    """Resolve the agent image ref for this plan.
+
+    Returns the committed snapshot if one exists, the cached image when the
+    policy is 'cached', or builds a fresh image and returns that."""
+    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)
+    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"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        return BottleImages(agent=plan.image)
+    docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    docker_mod.verify_agent_image(
+        plan.image, runtime_for(plan.agent_provider_template).smoke_test,
+    )
+    return BottleImages(agent=plan.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.
+    """Launch and provision a Docker bottle via compose. Teardown on exit."""
-    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)
 
@@ -96,25 +129,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. (The gateway image is built by the orchestrator.)
-        committed = read_committed_image(plan.slug)
-        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),
-            )
-        else:
-            docker_mod.build_image(
-                plan.image, _REPO_DIR,
-                dockerfile=plan.dockerfile_path,
-            )
-            docker_mod.verify_agent_image(
-                plan.image, runtime_for(plan.agent_provider_template).smoke_test,
-            )
-
         # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
         # provisioning the bottle's repos into the shared gateway.
         git_gate_plan = plan.git_gate_plan
@@ -207,3 +221,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/docker/util.py

@@ -5,6 +5,7 @@ existence, and building images."""
 from __future__ import annotations
 
 import os
+from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -197,3 +198,42 @@ def commit_container(container_name: str, image_tag: str) -> None:
             f"{(result.stderr or '').strip() or '<no stderr>'}"
         )
     info(f"committed {container_name!r} → {image_tag!r}")
+
+
+def image_created_at(ref: str) -> datetime:
+    """Return Docker's image Created timestamp as an aware UTC datetime."""
+    r = subprocess.run(
+        ["docker", "image", "inspect", "--format", "{{.Created}}", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if r.returncode != 0:
+        die(
+            f"docker image inspect for {ref!r} failed: "
+            f"{(r.stderr or '').strip() or '<no stderr>'}"
+        )
+    raw = r.stdout.strip()
+    try:
+        return _parse_docker_timestamp(raw)
+    except ValueError:
+        die(f"docker image inspect for {ref!r} returned invalid Created timestamp: {raw!r}")
+
+
+def _parse_docker_timestamp(raw: str) -> datetime:
+    text = raw.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    dot = text.find(".")
+    if dot != -1:
+        tz_plus = text.find("+", dot)
+        tz_minus = text.find("-", dot)
+        tz_candidates = [pos for pos in (tz_plus, tz_minus) if pos != -1]
+        if tz_candidates:
+            tz_pos = min(tz_candidates)
+            frac = text[dot + 1:tz_pos]
+            text = text[:dot + 1] + frac[:6].ljust(6, "0") + text[tz_pos:]
+    dt = datetime.fromisoformat(text)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)

bot_bottle/backend/firecracker/backend.py

@@ -18,7 +18,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...manifest import Manifest
 from ...supervise import SupervisePlan
-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
@@ -92,11 +92,18 @@ class FirecrackerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _build_or_load_images(self, plan: FirecrackerBottlePlan) -> BottleImages:
+        return BottleImages(agent=_launch.build_or_load_agent_base(plan))
+
+    def prelaunch_checks(self, plan: FirecrackerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
+    def _launch_impl(
-        self, plan: FirecrackerBottlePlan
+        self, plan: FirecrackerBottlePlan, images: BottleImages,
     ) -> Generator[FirecrackerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        assert isinstance(images.agent, Path)
+        with _launch.launch(plan, images.agent, provision=self.provision) as bottle:
             yield bottle
 
     def prepare_cleanup(self) -> FirecrackerBottleCleanupPlan:

bot_bottle/backend/firecracker/freezer.py

@@ -5,7 +5,7 @@ backend — we stream the guest root filesystem out over the control
 channel (SSH here). Unlike the other backends this needs no Docker: the
 tar *is* the resumable artifact. `resume` extracts it and rebuilds a
 fresh per-bottle ext4 with `mke2fs -d` (see `util.build_committed_rootfs_dir`
-and `launch._build_agent_base`). The bottle keeps running after the
+and `launch.build_or_load_agent_base`). The bottle keeps running after the
 snapshot.
 """
 

bot_bottle/backend/firecracker/image_builder.py

@@ -45,6 +45,13 @@ def _dockerfile_hash(dockerfile: Path) -> str:
     return hashlib.sha256(dockerfile.read_bytes()).hexdigest()[:16]
 
 
+def cached_agent_rootfs_dir(dockerfile: Path) -> Path | None:
+    """Return the ready cached rootfs for ``dockerfile``, if one exists."""
+    digest = _dockerfile_hash(dockerfile)
+    base = util.cache_dir() / "rootfs" / f"agent-{digest}"
+    return base if (base / ".bb-ready").is_file() else None
+
+
 def build_agent_rootfs_dir(
     dockerfile: Path, *, image_tag: str, smoke_test: tuple[str, ...] = (),
 ) -> Path:
@@ -58,7 +65,7 @@ def build_agent_rootfs_dir(
     silent-failure image at build time rather than at first agent use."""
     digest = _dockerfile_hash(dockerfile)
     base = util.cache_dir() / "rootfs" / f"agent-{digest}"
-    if (base / ".bb-ready").is_file():
+    if cached_agent_rootfs_dir(dockerfile) is not None:
         info(f"using cached agent rootfs {base.name}")
         return base
 

bot_bottle/backend/firecracker/launch.py

@@ -45,7 +45,8 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale_path
+from ...log import die, info, warn
 from ...supervise import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
@@ -64,6 +65,7 @@ _GIT_HTTP_PORT = 9420
 @contextmanager
 def launch(
     plan: FirecrackerBottlePlan,
+    agent_base: Path,
     *,
     provision: Callable[[FirecrackerBottlePlan, "FirecrackerBottle"], str | None],
 ) -> Generator[FirecrackerBottle, None, None]:
@@ -85,11 +87,9 @@ def launch(
             raise teardown_exc
 
     try:
-        # Step 1: agent rootfs. Built from the Dockerfile inside a Firecracker
+        # Step 1 (rootfs resolution/build) runs in BottleBackend.launch before
-        # builder VM (buildah, no host docker); a committed snapshot is reused
+        # this context starts resources. ``agent_base`` is the selected cache,
-        # when present. Returns the base dir the per-bottle ext4 is made from.
+        # fresh build, or committed snapshot.
-        plan, agent_base = _build_agent_base(plan)
-
         # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any.
         git_gate_plan = plan.git_gate_plan
         if git_gate_plan.upstreams:
@@ -206,9 +206,7 @@ def launch(
         teardown()
 
 
-def _build_agent_base(
+def build_or_load_agent_base(plan: FirecrackerBottlePlan) -> Path:
-    plan: FirecrackerBottlePlan,
-) -> tuple[FirecrackerBottlePlan, Path]:
     """Produce the agent's base rootfs dir. Primary path: build the Dockerfile
     inside a Firecracker builder VM (buildah, no host docker), smoke-testing
     the image before export. A committed snapshot (freeze/migrate) is resumed
@@ -217,13 +215,36 @@ def _build_agent_base(
     committed_tar = committed_rootfs_path(plan.slug)
     if committed and committed_tar.is_file():
         info(f"resuming from committed rootfs {committed_tar}")
-        return plan, util.build_committed_rootfs_dir(committed_tar)
+        return util.build_committed_rootfs_dir(committed_tar)
-    base = image_builder.build_agent_rootfs_dir(
+    dockerfile = Path(plan.dockerfile_path)
-        Path(plan.dockerfile_path),
+    if plan.spec.image_policy == "cached":
+        cached = image_builder.cached_agent_rootfs_dir(dockerfile)
+        if cached is None:
+            die(
+                f"cached agent rootfs for {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent rootfs {cached.name}")
+        return cached
+    return image_builder.build_agent_rootfs_dir(
+        dockerfile,
         image_tag=plan.image,
         smoke_test=runtime_for(plan.agent_provider_template).smoke_test,
     )
-    return plan, base
+
+
+def stale_checks(plan: FirecrackerBottlePlan) -> None:
+    """Raise when the cached rootfs selected by this plan is stale."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    committed_tar = committed_rootfs_path(plan.slug)
+    if committed and committed_tar.is_file():
+        check_stale_path(f"agent rootfs {committed_tar}", committed_tar)
+        return
+    cached = image_builder.cached_agent_rootfs_dir(Path(plan.dockerfile_path))
+    if cached is not None:
+        check_stale_path(f"agent rootfs {cached}", cached / ".bb-ready")
 
 
 # --- agent guest env -------------------------------------------------

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
@@ -82,11 +82,17 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    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(
+    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 ensure_orchestrator(self) -> str:

bot_bottle/backend/macos_container/launch.py

@@ -53,7 +53,9 @@ 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 .. import BottleImages
 from ...supervise import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
@@ -71,18 +73,43 @@ _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 _AGENT_SLEEP_SECONDS = "2147483647"
 
 
+def build_or_load_images(plan: MacosContainerBottlePlan) -> BottleImages:
+    """Resolve the agent image ref for this plan. The gateway's own image is
+    built by `ensure_gateway` — it belongs to the shared singleton."""
+    committed = read_committed_image(plan.slug)
+    if committed and container_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return BottleImages(agent=committed)
+    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"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        return BottleImages(agent=plan.image)
+    container_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    return BottleImages(agent=plan.image)
+
+
 @contextmanager
 def launch(
     plan: MacosContainerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
 ) -> Generator[MacosContainerBottle, None, None]:
-    """Build, run, register, provision, and yield an Apple Container bottle on
+    """Run, register, provision, and yield an Apple Container bottle on the
-    the shared per-host gateway."""
+    shared per-host gateway."""
     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:
@@ -95,8 +122,6 @@ def launch(
             raise teardown_exc
 
     try:
-        plan = _build_images(plan)
-
         # Step 1: the per-host singletons. Must precede the agent run — its
         # proxy env needs the gateway's address at `container run` time.
         endpoint = ensure_gateway()
@@ -177,22 +202,19 @@ def launch(
         teardown()
 
 
-def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
+
-    """Build the agent image. The gateway's own image is built by
+def stale_checks(plan: MacosContainerBottlePlan) -> None:
-    `ensure_gateway` — it belongs to the shared singleton, not to a bottle."""
+    """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):
-        info(f"using committed image {committed!r}")
+        check_stale(f"agent image {committed!r}", container_mod.image_created_at(committed))
-        return dataclasses.replace(
+        return
-            plan,
+    if container_mod.image_exists(plan.image):
-            agent_provision=dataclasses.replace(
+        check_stale(f"agent image {plan.image!r}", container_mod.image_created_at(plan.image))
-                plan.agent_provision, image=committed,
-            ),
-        )
-    container_mod.build_image(
-        plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path,
-    )
-    return plan
 
 
 def _provision_git_gate_keys(

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
@@ -74,6 +75,14 @@ def cmd_start(argv: list[str]) -> int:
             "skip all prompts. For orchestrators, CI, and webhooks."
         ),
     )
+    parser.add_argument(
+        "--cached-images",
+        action="store_true",
+        help=(
+            "quickstart with existing local agent and sidecar images; "
+            "only valid with --headless"
+        ),
+    )
     parser.add_argument(
         "--bottle",
         action="append",
@@ -106,6 +115,8 @@ def cmd_start(argv: list[str]) -> int:
         help="agent name defined in bot-bottle.json (omit to pick interactively)",
     )
     args = parser.parse_args(argv)
+    if args.cached_images and not args.headless:
+        die("--cached-images is only supported with --headless")
 
     dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
     if args.no_cache or os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
@@ -158,6 +169,10 @@ def cmd_start(argv: list[str]) -> int:
     label, color = tui.name_color_modal(default_label=agent_name)
     label, color = _resolve_unique_label(label, color)
 
+    image_policy = _select_image_policy()
+    if image_policy is None:
+        return 0
+
     spec = BottleSpec(
         manifest=manifest,
         agent_name=agent_name,
@@ -166,6 +181,7 @@ def cmd_start(argv: list[str]) -> int:
         label=label,
         color=color,
         bottle_names=bottle_names,
+        image_policy=image_policy,
     )
     return _launch_bottle(
         spec,
@@ -226,6 +242,7 @@ def _start_headless(
         color=args.color or "",
         bottle_names=bottle_names,
         headless=True,
+        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -406,6 +423,13 @@ def _text_prompt_yes() -> bool:
     return reply in ("y", "Y", "yes", "YES")
 
 
+def _select_image_policy() -> str | None:
+    return tui.filter_select(
+        ["fresh", "cached"],
+        title="Select image startup mode",
+    )
+
+
 def _text_render_preflight():
     def _render(plan: DockerBottlePlan, backend_name: str) -> None:
         print(file=sys.stderr)
@@ -548,6 +572,15 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
+        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, ...] = ()
@@ -566,10 +599,6 @@ def _launch_bottle(
                 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)
         return 0

bot_bottle/config_store.py

@@ -0,0 +1,71 @@
+"""SQLite-backed bot-bottle configuration store."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+try:
+    from .db_store import DbStore
+    from .migrations import TableMigrations
+    from .paths import host_db_path
+except ImportError:
+    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from paths import host_db_path  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS = 1
+
+
+class ConfigStore(DbStore):
+    """SQLite configuration for host-side bot-bottle settings."""
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        migrations = TableMigrations("config_store", [
+            # v1 — host-side bot-bottle settings
+            """
+            CREATE TABLE IF NOT EXISTS bot_bottle_config (
+                id INTEGER PRIMARY KEY CHECK (id = 1),
+                cached_image_stale_warning_days INTEGER NOT NULL DEFAULT 1
+            )
+            """,
+        ])
+        super().__init__(db_path or host_db_path(), migrations)
+
+    def cached_image_stale_warning_days(self) -> int:
+        if not self.db_path.is_file():
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+        with self._connect() as conn:
+            row = conn.execute(
+                """
+                SELECT cached_image_stale_warning_days
+                FROM bot_bottle_config
+                WHERE id = 1
+                """,
+            ).fetchone()
+        if row is None:
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+        try:
+            return int(row["cached_image_stale_warning_days"])
+        except (TypeError, ValueError):
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+
+    def set_cached_image_stale_warning_days(self, days: int) -> Path:
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO bot_bottle_config (id, cached_image_stale_warning_days)
+                VALUES (1, ?)
+                ON CONFLICT(id) DO UPDATE SET
+                    cached_image_stale_warning_days = excluded.cached_image_stale_warning_days
+                """,
+                (days,),
+            )
+        self._chmod()
+        return self.db_path
+
+
+__all__ = [
+    "DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS",
+    "ConfigStore",
+]

bot_bottle/db_store.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import sqlite3
-from contextlib import contextmanager
 from pathlib import Path
 
 try:
@@ -29,21 +28,12 @@ class DbStore:
         conn.row_factory = sqlite3.Row
         return conn
 
-    @contextmanager
-    def _connection(self):
-        conn = self._connect()
-        try:
-            with conn:
-                yield conn
-        finally:
-            conn.close()
-
     def is_migrated(self) -> bool:
         """Return True if the DB is fully up-to-date, False if migration is needed."""
         if not self.db_path.exists():
             return False
         try:
-            with self._connection() as conn:
+            with self._connect() as conn:
                 row = conn.execute(
                     "SELECT version FROM schema_versions WHERE module = ?",
                     (self._migrations.schema_key,),
@@ -55,7 +45,7 @@ class DbStore:
 
     def migrate(self) -> None:
         """Apply any pending migrations and set permissions on the DB file."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             self._migrations.apply(conn)
         self._chmod()
 

bot_bottle/image_cache.py

@@ -0,0 +1,42 @@
+"""Shared helpers for cached-image quickstart stale checks."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+try:
+    from .config_store import ConfigStore
+except ImportError:
+    from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+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
+    now = datetime.now(timezone.utc)
+    created = created_at.astimezone(timezone.utc)
+    age = now - created
+    if age.total_seconds() <= threshold_days * 86400:
+        return
+    raise StaleImageError(
+        f"cached {label} is {age.days} day(s) old; "
+        "quickstart does not verify it matches the current Dockerfile/context"
+    )
+
+
+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__ = ["StaleImageError", "check_stale", "check_stale_path"]

bot_bottle/orchestrator/registry.py

@@ -167,7 +167,7 @@ class RegistryStore(DbStore):
             metadata=metadata,
             policy=policy,
         )
-        with self._connection() as conn:
+        with self._connect() as conn:
             conn.execute(
                 "DELETE FROM orchestrator_bottles "
                 "WHERE source_ip = ? AND state = 'active' AND bottle_id != ?",
@@ -193,7 +193,7 @@ class RegistryStore(DbStore):
     def set_policy(self, bottle_id: str, policy: str) -> bool:
         """Update a bottle's policy in place (live reload). Returns True if
         the bottle exists."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             cur = conn.execute(
                 "UPDATE orchestrator_bottles SET policy = ? WHERE bottle_id = ?",
                 (policy, bottle_id),
@@ -203,7 +203,7 @@ class RegistryStore(DbStore):
 
     def deregister(self, bottle_id: str) -> bool:
         """Remove a bottle. Returns True if a row was deleted."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             cur = conn.execute(
                 "DELETE FROM orchestrator_bottles WHERE bottle_id = ?", (bottle_id,)
             )
@@ -211,7 +211,7 @@ class RegistryStore(DbStore):
 
     def get(self, bottle_id: str) -> BottleRecord | None:
         """Return the bottle by id, or None if absent."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             row = conn.execute(
                 "SELECT * FROM orchestrator_bottles WHERE bottle_id = ?", (bottle_id,)
             ).fetchone()
@@ -219,7 +219,7 @@ class RegistryStore(DbStore):
 
     def all(self) -> list[BottleRecord]:
         """Every registered bottle, oldest first."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             rows = conn.execute(
                 "SELECT * FROM orchestrator_bottles ORDER BY created_at"
             ).fetchall()
@@ -232,7 +232,7 @@ class RegistryStore(DbStore):
         source IP is unspoofable (Firecracker `/31` + nft) and the control
         plane is reachable only by the trusted gateway; pair with the
         identity token (`attribute`) elsewhere."""
-        with self._connection() as conn:
+        with self._connect() as conn:
             rows = conn.execute(
                 "SELECT * FROM orchestrator_bottles "
                 "WHERE source_ip = ? AND state = 'active'",

bot_bottle/queue_store.py

@@ -66,7 +66,7 @@ class QueueStore(DbStore):
         super().__init__(resolved, migrations)
 
     def write_proposal(self, proposal: Proposal) -> Path:
-        with self._connection() as conn:
+        with self._connect() as conn:
             conn.execute(
                 """
                 INSERT OR REPLACE INTO supervise_proposals (
@@ -89,7 +89,7 @@ class QueueStore(DbStore):
         return self.db_path
 
     def read_proposal(self, proposal_id: str) -> Proposal:
-        with self._connection() as conn:
+        with self._connect() as conn:
             row = conn.execute(
                 """
                 SELECT * FROM supervise_proposals
@@ -104,7 +104,7 @@ class QueueStore(DbStore):
     def list_pending_proposals(self) -> list[Proposal]:
         if not self.db_path.is_file():
             return []
-        with self._connection() as conn:
+        with self._connect() as conn:
             rows = conn.execute(
                 """
                 SELECT p.* FROM supervise_proposals p
@@ -125,7 +125,7 @@ class QueueStore(DbStore):
     def list_all_pending_proposals(self) -> list[Proposal]:
         if not self.db_path.is_file():
             return []
-        with self._connection() as conn:
+        with self._connect() as conn:
             rows = conn.execute(
                 """
                 SELECT p.* FROM supervise_proposals p
@@ -142,7 +142,7 @@ class QueueStore(DbStore):
         return [self._row_to_proposal(row) for row in rows]
 
     def write_response(self, response: Response) -> Path:
-        with self._connection() as conn:
+        with self._connect() as conn:
             conn.execute(
                 """
                 INSERT OR REPLACE INTO supervise_responses (
@@ -161,7 +161,7 @@ class QueueStore(DbStore):
         return self.db_path
 
     def read_response(self, proposal_id: str) -> Response:
-        with self._connection() as conn:
+        with self._connect() as conn:
             row = conn.execute(
                 """
                 SELECT * FROM supervise_responses
@@ -176,7 +176,7 @@ class QueueStore(DbStore):
     def archive_proposal(self, proposal_id: str) -> None:
         if not self.db_path.is_file():
             return
-        with self._connection() as conn:
+        with self._connect() as conn:
             conn.execute(
                 """
                 UPDATE supervise_proposals SET archived = 1

bot_bottle/store_manager.py

@@ -6,9 +6,11 @@ from pathlib import Path
 
 try:
     from .audit_store import AuditStore
+    from .config_store import ConfigStore
     from .queue_store import QueueStore
 except ImportError:
     from audit_store import AuditStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
     from queue_store import QueueStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 
 _instance: StoreManager | None = None
@@ -47,11 +49,13 @@ class StoreManager:
         return (
             QueueStore("", self.db_path).is_migrated()
             and AuditStore(self.db_path).is_migrated()
+            and ConfigStore(self.db_path).is_migrated()
         )
 
     def migrate(self) -> None:
         QueueStore("", self.db_path).migrate()
         AuditStore(self.db_path).migrate()
+        ConfigStore(self.db_path).migrate()
 
 
 __all__ = ["StoreManager"]

scripts/coverage.sh

@@ -27,8 +27,7 @@ echo "== unit ==" >&2
 "$PY" -m coverage run -m unittest discover -t . -s tests/unit
 
 echo "== integration (skips without Docker) ==" >&2
-BOT_BOTTLE_BACKEND=firecracker \
+"$PY" -m coverage run --append -m unittest discover -t . -s tests/integration
-    "$PY" -m coverage run --append -m unittest discover -t . -s tests/integration
 
 echo "== combined report ==" >&2
 "$PY" -m coverage report -m

tests/integration/test_sandbox_escape.py

@@ -69,12 +69,11 @@ _DUMMY_HOST_KEY = (
 
 @skip_unless_docker()
 @unittest.skipIf(
-    os.environ.get("GITEA_ACTIONS") == "true"
+    os.environ.get("GITEA_ACTIONS") == "true",
-    and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
+    "skipped under act_runner: egress_tls_init uses a host bind mount "
-    "skipped under act_runner unless BOT_BOTTLE_BACKEND=firecracker: "
+    "the runner container can't see, and the network topology hides "
-    "egress_tls_init uses a host bind mount the runner container can't "
+    "sibling-gateway visibility — same constraint as the other "
-    "see, and the network topology hides sibling-gateway visibility — "
+    "bottle-bringup integration tests",
-    "these constraints don't apply on the self-hosted KVM runner",
 )
 class TestSandboxEscape(unittest.TestCase):
     """End-to-end attacks against a real bottle. The bottle stays

tests/unit/test_cli_start_headless.py

@@ -184,6 +184,18 @@ class TestCmdStartHeadless(unittest.TestCase):
         )
         self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"])
 
+    def test_cached_images_sets_cached_policy(self):
+        start_mod.cmd_start(
+            ["--headless", "--cached-images", "researcher", "--bottle", "claude",
+             "--prompt", "Do it"]
+        )
+        self.assertEqual("cached", self._spec().image_policy)
+
+    def test_cached_images_requires_headless(self):
+        with self.assertRaises(Die):
+            start_mod.cmd_start(["--cached-images", "researcher"])
+        self._launch_mock.assert_not_called()
+
 
 class TestPrepareWithPreflight(unittest.TestCase):
     """prepare_with_preflight calls render_preflight with the plan and backend name."""

tests/unit/test_cli_start_selector.py

@@ -57,6 +57,12 @@ class TestCmdStartSelector(unittest.TestCase):
         self._bottle_picker_mock = self._bottle_picker_patch.start()
         self._bottle_picker_mock.return_value = ["claude"]  # default: one bottle selected
 
+        self._image_policy_patch = patch(
+            "bot_bottle.cli.start._select_image_policy",
+            return_value="fresh",
+        )
+        self._image_policy_patch.start()
+
         self._env_patch = patch.dict(os.environ, {}, clear=False)
         self._env_patch.start()
         os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -66,6 +72,7 @@ class TestCmdStartSelector(unittest.TestCase):
         self._launch_patch.stop()
         self._agent_picker_patch.stop()
         self._bottle_picker_patch.stop()
+        self._image_policy_patch.stop()
         self._env_patch.stop()
 
     # ------------------------------------------------------------------
@@ -124,6 +131,19 @@ class TestCmdStartSelector(unittest.TestCase):
         spec = self._launch_mock.call_args[0][0]
         self.assertEqual(("claude", "dev"), spec.bottle_names)
 
+    def test_image_policy_forwarded_to_spec(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value="cached"):
+            start_mod.cmd_start(["researcher"])
+        self._launch_mock.assert_called_once()
+        spec = self._launch_mock.call_args[0][0]
+        self.assertEqual("cached", spec.image_policy)
+
+    def test_image_policy_cancel_returns_0(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value=None):
+            rc = start_mod.cmd_start(["researcher"])
+        self.assertEqual(0, rc)
+        self._launch_mock.assert_not_called()
+
     def test_empty_bottle_selection_forwarded(self):
         self._bottle_picker_mock.return_value = []
         start_mod.cmd_start(["researcher"])
@@ -215,6 +235,7 @@ class TestCmdStartLabelCollision(unittest.TestCase):
         ).start()
         # Stub the bottle picker to always return a selection.
         patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start()
+        patch("bot_bottle.cli.start._select_image_policy", return_value="fresh").start()
         self.addCleanup(patch.stopall)
 
     def test_no_collision_proceeds_without_reprompt(self):

tests/unit/test_cli_start_stale.py

@@ -0,0 +1,146 @@
+"""Unit: _launch_bottle StaleImageError handling.
+
+Exercises prelaunch_checks / backend.launch flow:
+  - headless mode → die on stale
+  - interactive mode, user declines → stop without launching
+  - interactive mode, user confirms → skip stale check and launch once
+"""
+
+from __future__ import annotations
+
+import io
+import tempfile
+import unittest
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.image_cache import StaleImageError
+from bot_bottle.log import Die
+
+
+def _fake_plan() -> Any:
+    provision = SimpleNamespace(startup_args=())
+    return cast(Any, SimpleNamespace(
+        agent_provision=provision,
+        agent_provider_template="claude",
+        slug="dev-abc",
+    ))
+
+
+def _ok_cm(bottle: Any) -> MagicMock:
+    """Return a context-manager mock that yields `bottle`."""
+    cm = MagicMock()
+    cm.__enter__ = MagicMock(return_value=bottle)
+    cm.__exit__ = MagicMock(return_value=False)
+    return cm
+
+
+class TestLaunchBottleStaleHandling(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.")
+
+    def _spec(self) -> Any:
+        from bot_bottle.backend import BottleSpec
+        from bot_bottle.manifest import ManifestIndex
+        idx = ManifestIndex.from_json_obj({
+            "bottles": {"dev": {}},
+            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+        })
+        return BottleSpec(
+            manifest=idx,
+            agent_name="demo",
+            copy_cwd=False,
+            user_cwd=self._tmp,
+            identity="dev-abc",
+        )
+
+    def _run_launch(self, **patch_kwargs: Any) -> int:
+        import bot_bottle.cli.start as start_mod
+        spec = self._spec()
+        with patch.object(start_mod, "prepare_with_preflight",
+                          return_value=(_fake_plan(), "dev-abc")), \
+             patch.object(start_mod, "settle_state"), \
+             patch.object(start_mod, "info"):
+            return start_mod._launch_bottle(
+                spec,
+                dry_run=False,
+                backend_name="docker",
+                **patch_kwargs,
+            )
+
+    def test_headless_stale_calls_die(self) -> None:
+        """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.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)
+
+        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.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"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        backend_mock.launch.assert_not_called()
+
+    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"
+
+        backend_mock = MagicMock()
+        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"), \
+             patch.object(start_mod, "attach_agent", return_value=0), \
+             patch.object(start_mod, "capture_claude_session_state"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        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."""
+        import bot_bottle.cli.start as start_mod
+
+        bottle_mock = MagicMock()
+        bottle_mock.name = "dev-abc"
+
+        backend_mock = MagicMock()
+        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"), \
+             patch.object(start_mod, "attach_agent", return_value=0), \
+             patch.object(start_mod, "capture_claude_session_state"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        backend_mock.launch.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_config_store.py

@@ -0,0 +1,86 @@
+"""Unit tests for the host-side configuration store."""
+
+from __future__ import annotations
+
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from bot_bottle.config_store import (
+    DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+    ConfigStore,
+)
+from bot_bottle.store_manager import StoreManager
+
+
+class TestConfigStore(unittest.TestCase):
+    def test_cached_image_warning_days_defaults_to_one(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+    def test_cached_image_warning_days_reads_value(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            store.set_cached_image_stale_warning_days(7)
+            self.assertEqual(7, store.cached_image_stale_warning_days())
+
+    def test_config_schema_uses_explicit_settings_columns(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            with sqlite3.connect(store.db_path) as conn:
+                conn.row_factory = sqlite3.Row
+                columns = [
+                    row["name"]
+                    for row in conn.execute("PRAGMA table_info(bot_bottle_config)")
+                ]
+            self.assertEqual([
+                "id",
+                "cached_image_stale_warning_days",
+            ], columns)
+
+    def test_store_manager_includes_config_store(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            db = Path(tmp) / "bot-bottle.db"
+            manager = StoreManager(db)
+            self.assertFalse(manager.is_migrated())
+            manager.migrate()
+            self.assertTrue(manager.is_migrated())
+
+    def test_cached_image_warning_days_returns_default_when_db_missing(self) -> None:
+        # When the db file doesn't exist yet (parent exists, file doesn't),
+        # the store returns the default without touching the file.
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "missing.db")
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+    def test_cached_image_warning_days_returns_default_on_null_value(self) -> None:
+        # If the row exists but the value is NULL (or not castable to int),
+        # the store falls back to the default.
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            db_path = Path(tmp) / "bot-bottle.db"
+            store = ConfigStore(db_path)
+            store.migrate()
+            # Write a NULL value directly.
+            with sqlite3.connect(db_path) as conn:
+                conn.execute(
+                    "UPDATE bot_bottle_config SET cached_image_stale_warning_days = NULL WHERE id = 1"
+                )
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_docker_launch_committed_image.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import contextlib
+import dataclasses
 import io
 import tempfile
 import unittest
@@ -18,6 +19,7 @@ from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.docker.consolidated_launch import LaunchContext
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
+from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 from tests.unit import use_bottle_root
 
@@ -92,6 +94,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
                 mock.patch.object(launch_mod.docker_mod, "image_exists", return_value=image_present), \
                 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, "launch_consolidated", return_value=_CTX), \
                 mock.patch.object(launch_mod, "teardown_consolidated"), \
                 mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
@@ -112,7 +115,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
         with self._patched(
             committed_tag=committed_tag, image_present=image_present, compose=compose,
         ) as built:
-            with launch_mod.launch(plan, provision=mock.Mock(return_value=None)):
+            images = launch_mod.build_or_load_images(plan)
+            with launch_mod.launch(plan, images, provision=mock.Mock(return_value=None)):
                 pass
         return built
 
@@ -127,14 +131,34 @@ class TestLaunchCommittedImage(unittest.TestCase):
             captured.append(p)
             return {"services": {"agent": {}}}
 
-        with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose):
+        with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose) as _:
-            with launch_mod.launch(_plan(self._tmp), provision=mock.Mock(return_value=None)):
+            plan = _plan(self._tmp)
+            images = launch_mod.build_or_load_images(plan)
+            with launch_mod.launch(plan, images, provision=mock.Mock(return_value=None)):
                 pass
         self.assertEqual(_COMMITTED_TAG, captured[0].image)
 
     def test_falls_back_to_build_when_no_committed_image(self) -> None:
         self.assertEqual([_DEFAULT_IMAGE], self._run_launch(_plan(self._tmp), committed_tag=None))
 
+    def test_cached_images_skip_build_when_present(self) -> None:
+        base = _plan(self._tmp)
+        plan = dataclasses.replace(
+            base,
+            spec=dataclasses.replace(base.spec, image_policy="cached"),
+        )
+        built = self._run_launch(plan, committed_tag=None, image_present=True)
+        self.assertEqual([], built)
+
+    def test_cached_images_die_when_agent_missing(self) -> None:
+        base = _plan(self._tmp)
+        plan = dataclasses.replace(
+            base,
+            spec=dataclasses.replace(base.spec, image_policy="cached"),
+        )
+        with self.assertRaises(Die):
+            self._run_launch(plan, committed_tag=None, image_present=False)
+
     def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
         built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=False)
         self.assertEqual([_DEFAULT_IMAGE], built)

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.backend.docker.consolidated_launch import LaunchContext
@@ -93,6 +93,8 @@ class TestTeardownWarning(unittest.TestCase):
             orchestrator_url="http://orch:8099",
         )
 
+        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.docker_mod, "verify_agent_image"), \
              mock.patch.object(launch_mod, "launch_consolidated", return_value=ctx), \
@@ -113,7 +115,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_docker_util_image.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import subprocess
 import unittest
+from datetime import timezone
 from unittest.mock import patch
 
 from bot_bottle.backend.docker import util as docker_mod
@@ -26,6 +27,53 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
     )
 
 
+
+class TestImageCreatedAt(unittest.TestCase):
+    def test_parses_docker_timestamp_with_nanoseconds(self):
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout="2026-07-06T15:33:47.123456789Z\n"),
+        ) as run:
+            created = docker_mod.image_created_at("bot-bottle-claude:latest")
+        self.assertEqual(2026, created.year)
+        self.assertEqual(123456, created.microsecond)
+        self.assertEqual(timezone.utc, created.tzinfo)
+        self.assertEqual(
+            ["docker", "image", "inspect", "--format", "{{.Created}}", "bot-bottle-claude:latest"],
+            run.call_args.args[0],
+        )
+
+    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_created_at("missing:tag")
+        die.assert_called_once()
+        self.assertIn("missing:tag", die.call_args.args[0])
+
+    def test_dies_on_invalid_timestamp(self):
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout="not-a-timestamp\n"),
+        ), patch.object(
+            docker_mod, "die", side_effect=SystemExit("die"),
+        ) as die:
+            with self.assertRaises(SystemExit):
+                docker_mod.image_created_at("some:tag")
+        die.assert_called_once()
+        self.assertIn("some:tag", die.call_args.args[0])
+
+    def test_parse_docker_timestamp_no_tzinfo_defaults_to_utc(self):
+        # A bare datetime with no tz offset should be treated as UTC.
+        dt = docker_mod._parse_docker_timestamp("2024-05-01T10:00:00.000000")
+        self.assertIsNotNone(dt.tzinfo)
+        self.assertEqual(timezone.utc, dt.tzinfo)
+
+
+
 class TestCommitContainer(unittest.TestCase):
     def test_runs_docker_commit(self):
         with patch.object(

tests/unit/test_firecracker_image_builder.py

@@ -35,6 +35,15 @@ class TestBuildAgentRootfsDir(unittest.TestCase):
         build.assert_not_called()
         self.assertEqual(base, out)
 
+    def test_cached_lookup_requires_ready_marker(self):
+        digest = image_builder._dockerfile_hash(self.dockerfile)
+        base = self.cache / "rootfs" / f"agent-{digest}"
+        base.mkdir(parents=True)
+        with patch.object(image_builder.util, "cache_dir", return_value=self.cache):
+            self.assertIsNone(image_builder.cached_agent_rootfs_dir(self.dockerfile))
+            (base / ".bb-ready").write_text("ok\n")
+            self.assertEqual(base, image_builder.cached_agent_rootfs_dir(self.dockerfile))
+
     def test_cache_miss_builds_injects_and_marks_ready(self):
         with patch.object(image_builder.util, "cache_dir", return_value=self.cache), \
              patch.object(image_builder, "_build_in_infra") as build, \

tests/unit/test_image_cache.py

@@ -0,0 +1,81 @@
+"""Unit: image_cache.py — check_stale / check_stale_path."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path
+
+
+class TestCheckStale(unittest.TestCase):
+    def _run(self, threshold: int, age_days: float) -> None:
+        created = datetime.now(tz=timezone.utc) - timedelta(days=age_days)
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = threshold
+            check_stale("test image", created)
+
+    def test_negative_threshold_never_raises(self):
+        # Threshold < 0 means the check is disabled — always passes.
+        self._run(threshold=-1, age_days=9999)
+
+    def test_zero_threshold_raises_immediately(self):
+        # threshold=0 means any image is stale the moment it exists.
+        with self.assertRaises(StaleImageError):
+            self._run(threshold=0, age_days=0.1)
+
+    def test_within_threshold_does_not_raise(self):
+        # Age well under threshold — should pass silently.
+        self._run(threshold=7, age_days=2)
+
+    def test_at_threshold_does_not_raise(self):
+        # Exactly at the boundary is fine (<=, not <).
+        self._run(threshold=1, age_days=0.9999)
+
+    def test_exceeds_threshold_raises(self):
+        created = datetime.now(tz=timezone.utc) - timedelta(days=3)
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = 1
+            with self.assertRaises(StaleImageError) as ctx:
+                check_stale("agent image 'bot-bottle:latest'", created)
+        self.assertIn("agent image", str(ctx.exception))
+        self.assertIn("day(s) old", str(ctx.exception))
+
+    def test_naive_datetime_treated_as_utc(self):
+        # check_stale calls .astimezone(utc) on the input; naive datetimes
+        # that would be interpreted as local time should still work.
+        # We can't control the local tz in a unit test, so just ensure
+        # no exception is thrown for a very recent naive datetime.
+        naive_now = datetime(2099, 1, 1)  # far future, always "fresh"
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = 1
+            # Should not raise — the image is brand new.
+            check_stale("test image", naive_now)
+
+
+class TestCheckStalePath(unittest.TestCase):
+    def test_delegates_to_check_stale_with_mtime(self):
+        with tempfile.NamedTemporaryFile() as f:
+            path = Path(f.name)
+            with patch("bot_bottle.image_cache.check_stale") as mock_check:
+                check_stale_path("some artifact", path)
+            mock_check.assert_called_once()
+            label, dt = mock_check.call_args.args
+            self.assertEqual("some artifact", label)
+            self.assertIsInstance(dt, datetime)
+            self.assertIsNotNone(dt.tzinfo)
+
+    def test_raises_stale_for_old_file(self):
+        with tempfile.NamedTemporaryFile() as f:
+            path = Path(f.name)
+            with patch("bot_bottle.image_cache.ConfigStore") as cs:
+                cs.return_value.cached_image_stale_warning_days.return_value = 0
+                with self.assertRaises(StaleImageError):
+                    check_stale_path("cached artifact", path)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_macos_container_util.py

@@ -323,5 +323,71 @@ class TestWaitContainerIpv4(unittest.TestCase):
             self.assertEqual("", util.wait_container_ipv4_on_network("c", "net", timeout=-1))
 
 
+class TestMacosContainerImageCreatedAt(unittest.TestCase):
+    def _ok(self, stdout: str) -> "util.subprocess.CompletedProcess":  # type: ignore
+        return util.subprocess.CompletedProcess(
+            args=[], returncode=0, stdout=stdout, stderr="",
+        )
+
+    def _fail(self, stderr: str = "no such image") -> "util.subprocess.CompletedProcess":  # type: ignore
+        return util.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr=stderr,
+        )
+
+    def test_parses_iso_timestamp_from_dict(self):
+        payload = '[{"created": "2025-06-01T12:00:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("bot-bottle-agent:latest")
+        self.assertEqual(2025, dt.year)
+        self.assertEqual(6, dt.month)
+        self.assertEqual(1, dt.day)
+
+    def test_accepts_list_or_dict_input(self):
+        # Container CLI may return a list; we take the first element.
+        payload = '[{"created": "2024-01-15T08:30:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("some-image:latest")
+        self.assertEqual(2024, dt.year)
+
+    def test_accepts_uppercase_Created_field(self):
+        payload = '[{"Created": "2024-03-20T10:00:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("some-image:latest")
+        self.assertEqual(2024, dt.year)
+        self.assertEqual(3, dt.month)
+
+    def test_dies_on_nonzero_returncode(self):
+        with patch.object(util.subprocess, "run", return_value=self._fail("not found")), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("missing:tag")
+        die.assert_called_once()
+        self.assertIn("missing:tag", die.call_args.args[0])
+
+    def test_dies_on_malformed_json(self):
+        with patch.object(util.subprocess, "run", return_value=self._ok("not-json {")), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("some:tag")
+        die.assert_called_once()
+
+    def test_dies_when_no_created_field(self):
+        payload = '[{"id": "sha256:abc123"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("some:tag")
+        die.assert_called_once()
+        self.assertIn("creation timestamp", die.call_args.args[0])
+
+    def test_dies_on_invalid_timestamp_format(self):
+        payload = '[{"created": "not-a-date"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("some:tag")
+        die.assert_called_once()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_stale_checks.py

@@ -0,0 +1,265 @@
+"""Unit: stale-image check functions across backends, and the
+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 unittest
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock, patch
+
+
+def _bottle_cm(bottle: Any) -> MagicMock:
+    """Return a mock context manager that yields `bottle`."""
+    cm = MagicMock()
+    cm.__enter__ = MagicMock(return_value=bottle)
+    cm.__exit__ = MagicMock(return_value=False)
+    return cm
+
+
+# ---------------------------------------------------------------------------
+# BottleBackend.launch template — prelaunch_checks + _build_or_load_images
+# ---------------------------------------------------------------------------
+
+class TestBottleBackendLaunchTemplate(unittest.TestCase):
+    """Verify the concrete launch() method on BottleBackend calls
+    _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_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, "_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
+        build_mock.assert_called_once_with(plan)
+        impl_mock.assert_called_once_with(plan, images)
+
+    def test_noop_default_prelaunch_checks(self) -> None:
+        from bot_bottle.backend.docker.backend import DockerBottleBackend
+        from bot_bottle.backend import BottleBackend
+        backend = DockerBottleBackend()
+        # Base-class prelaunch_checks is a no-op — must not raise.
+        BottleBackend.prelaunch_checks(backend, cast(Any, SimpleNamespace()))  # type: ignore[arg-type]
+
+
+# ---------------------------------------------------------------------------
+# Docker backend stale_checks
+# ---------------------------------------------------------------------------
+
+class TestDockerStaleChecks(unittest.TestCase):
+    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
+        spec = SimpleNamespace(image_policy=policy)
+        provision = SimpleNamespace(image="bot-bottle-agent:latest")
+        return cast(Any, SimpleNamespace(
+            spec=spec,
+            slug=slug,
+            image="bot-bottle-agent:latest",
+            agent_provision=provision,
+        ))
+
+    def test_fresh_policy_is_noop(self) -> None:
+        from bot_bottle.backend.docker import launch as mod
+        with patch.object(mod, "read_committed_image") as rci, \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan("fresh"))
+        rci.assert_not_called()
+        cs.assert_not_called()
+
+    def test_committed_image_present_checks_only_committed(self) -> None:
+        from bot_bottle.backend.docker import launch as mod
+        from datetime import datetime, timezone
+        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
+             patch.object(mod.docker_mod, "image_exists", return_value=True), \
+             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan())
+        cs.assert_called_once()
+        self.assertIn("committed:latest", cs.call_args.args[0])
+
+    def test_no_committed_image_checks_agent(self) -> None:
+        from bot_bottle.backend.docker import launch as mod
+        from datetime import datetime, timezone
+        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        plan = self._plan()
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(mod.docker_mod, "image_exists", return_value=True), \
+             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(plan)
+        cs.assert_called_once()
+        self.assertIn(plan.image, cs.call_args.args[0])
+
+    def test_image_not_present_skips_check(self) -> None:
+        from bot_bottle.backend.docker import launch as mod
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(mod.docker_mod, "image_exists", return_value=False), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan())
+        cs.assert_not_called()
+
+    def test_only_sidecar_missing_checks_only_agent(self) -> None:
+        from bot_bottle.backend.docker import launch as mod
+        from datetime import datetime, timezone
+        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        plan = self._plan()
+
+        def image_exists(ref: str) -> bool:
+            return ref == plan.image  # Only agent present; sidecar missing.
+
+        with 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_created_at", return_value=ts), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(plan)
+        self.assertEqual(1, cs.call_count)
+        self.assertIn(plan.image, cs.call_args.args[0])
+
+    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.prelaunch_checks(plan)  # type: ignore[arg-type]
+        sc.assert_called_once_with(plan)
+
+
+# ---------------------------------------------------------------------------
+# macOS container backend stale_checks
+# ---------------------------------------------------------------------------
+
+class TestMacosContainerStaleChecks(unittest.TestCase):
+    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
+        return cast(Any, SimpleNamespace(
+            spec=SimpleNamespace(image_policy=policy),
+            slug=slug,
+            image="bot-bottle-agent:latest",
+        ))
+
+    def test_fresh_policy_is_noop(self) -> None:
+        from bot_bottle.backend.macos_container import launch as mod
+        with patch.object(mod, "read_committed_image") as rci, \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan("fresh"))
+        rci.assert_not_called()
+        cs.assert_not_called()
+
+    def test_committed_image_present_checks_only_committed(self) -> None:
+        from bot_bottle.backend.macos_container import launch as mod
+        from datetime import datetime, timezone
+        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
+             patch.object(mod.container_mod, "image_exists", return_value=True), \
+             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan())
+        cs.assert_called_once()
+        self.assertIn("committed:latest", cs.call_args.args[0])
+
+    def test_no_committed_image_checks_agent(self) -> None:
+        from bot_bottle.backend.macos_container import launch as mod
+        from datetime import datetime, timezone
+        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        plan = self._plan()
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(mod.container_mod, "image_exists", return_value=True), \
+             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(plan)
+        cs.assert_called_once()
+        self.assertIn(plan.image, cs.call_args.args[0])
+
+    def test_image_not_present_skips_check(self) -> None:
+        from bot_bottle.backend.macos_container import launch as mod
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(mod.container_mod, "image_exists", return_value=False), \
+             patch.object(mod, "check_stale") as cs:
+            mod.stale_checks(self._plan())
+        cs.assert_not_called()
+
+    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.prelaunch_checks(plan)  # type: ignore[arg-type]
+        sc.assert_called_once_with(plan)
+
+
+# ---------------------------------------------------------------------------
+# Firecracker cached rootfs selection and stale checks
+# ---------------------------------------------------------------------------
+
+class TestFirecrackerCachedRootfs(unittest.TestCase):
+    def _plan(self, policy: str = "cached") -> Any:
+        return cast(Any, SimpleNamespace(
+            spec=SimpleNamespace(image_policy=policy),
+            slug="dev-abc",
+            image="bot-bottle-agent:latest",
+            dockerfile_path="/repo/Dockerfile",
+            agent_provider_template="claude",
+        ))
+
+    def test_cached_policy_reuses_ready_rootfs_without_building(self) -> None:
+        from bot_bottle.backend.firecracker import launch as mod
+        cached = Path("/cache/rootfs/agent-deadbeef")
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(
+                 mod.image_builder, "cached_agent_rootfs_dir", return_value=cached,
+             ), \
+             patch.object(mod.image_builder, "build_agent_rootfs_dir") as build:
+            self.assertEqual(cached, mod.build_or_load_agent_base(self._plan()))
+        build.assert_not_called()
+
+    def test_cached_policy_fails_when_rootfs_is_missing(self) -> None:
+        from bot_bottle.backend.firecracker import launch as mod
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(
+                 mod.image_builder, "cached_agent_rootfs_dir", return_value=None,
+             ), \
+             patch.object(mod.image_builder, "build_agent_rootfs_dir") as build, \
+             self.assertRaises(SystemExit):
+            mod.build_or_load_agent_base(self._plan())
+        build.assert_not_called()
+
+    def test_stale_checks_use_ready_marker_timestamp(self) -> None:
+        from bot_bottle.backend.firecracker import launch as mod
+        cached = Path("/cache/rootfs/agent-deadbeef")
+        with patch.object(mod, "read_committed_image", return_value=""), \
+             patch.object(
+                 mod.image_builder, "cached_agent_rootfs_dir", return_value=cached,
+             ), \
+             patch.object(mod, "check_stale_path") as check:
+            mod.stale_checks(self._plan())
+        check.assert_called_once_with(f"agent rootfs {cached}", cached / ".bb-ready")
+
+    def test_backend_prelaunch_checks_delegates(self) -> None:
+        from bot_bottle.backend.firecracker.backend import FirecrackerBottleBackend
+        from bot_bottle.backend.firecracker import launch as mod
+        plan = self._plan()
+        with patch.object(mod, "stale_checks") as checks:
+            FirecrackerBottleBackend().prelaunch_checks(plan)
+        checks.assert_called_once_with(plan)
+
+
+if __name__ == "__main__":
+    unittest.main()