fix(lint): remove unused BottleImages imports from two test files

pyright strict reportUnusedImport flagged BottleImages in
test_docker_launch_committed_image.py and test_macos_container_launch.py;
neither file references the type by name (they only use the returned
value's attributes).

bot_bottle/backend/__init__.py

@@ -36,10 +36,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager
+from contextlib import AbstractContextManager, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generic, Sequence, TypeVar
+from typing import Any, Generator, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -75,6 +75,12 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
+    # 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)
@@ -270,6 +276,18 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
 CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan)
 
 
+@dataclass(frozen=True)
+class BottleImages:
+    """Resolved image references (or artifact paths) for a bottle launch.
+
+    For Docker/macOS-container backends, `agent` and `sidecar` are string
+    image refs. For the smolmachines backend they are Path objects pointing
+    to pre-built `.smolmachine` artifacts."""
+
+    agent: str | Path
+    sidecar: str | Path
+
+
 class BottleBackend(ABC, Generic[PlanT, CleanupT]):
     """Abstract base for selectable bottle backends. Concrete subclasses
     (e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -300,6 +318,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
 
         self._preflight()
 
+        from ..git_gate_host_key import preflight_host_keys
+        manifest = preflight_host_keys(
+            manifest,
+            headless=spec.headless,
+            home_md=spec.manifest.home_md,
+        )
+
         manifest_bottle = manifest.bottle
         manifest_agent_provider = manifest_bottle.agent_provider
         agent_provider = get_provider(manifest_agent_provider.template)
@@ -422,9 +447,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
+    def prelaunch_checks(self, plan: PlanT) -> None:
+        """Raise StaleImageError if any cached image used by this plan is stale.
+        No-op default; backends override to call the shared check_stale*
+        helpers on their image/artifact timestamps. Called by the CLI before
+        launch so the operator can be prompted outside the launch context."""
+
+    @contextmanager
+    def launch(self, plan: PlanT) -> 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]:
-        """Build/run the bottle and yield a handle; tear down on exit."""
+    def _build_or_load_images(self, plan: PlanT) -> BottleImages:
+        """Return the agent and sidecar image references (or artifact paths)
+        for this plan, building fresh images when the policy requires it."""
+
+    @abstractmethod
+    def _launch_impl(self, plan: PlanT, images: BottleImages) -> AbstractContextManager[Bottle]:
+        """Bring up the bottle using pre-resolved images; yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy host-side files (CA cert, prompt, skills, .git) into

bot_bottle/backend/docker/backend.py

@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -85,9 +85,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]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+    def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:

bot_bottle/backend/docker/compose.py

@@ -180,10 +180,6 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
 
     service: dict[str, Any] = {
         "image": SIDECAR_BUNDLE_IMAGE,
-        "build": {
-            "context": _REPO_DIR,
-            "dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
-        },
         "container_name": sidecar_bundle_container_name(plan.slug),
         "networks": {
             "internal": {"aliases": internal_aliases},
@@ -192,6 +188,11 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
         "environment": env,
         "volumes": volumes,
     }
+    if plan.spec.image_policy != "cached":
+        service["build"] = {
+            "context": _REPO_DIR,
+            "dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
+        }
     return service
 
 

bot_bottle/backend/docker/launch.py

@@ -41,7 +41,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 network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -63,22 +65,57 @@ from .compose import (
     write_compose_file,
 )
 from .egress import egress_tls_init
+from .sidecar_bundle import SIDECAR_BUNDLE_IMAGE
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 
 
+def build_or_load_images(plan: DockerBottlePlan) -> BottleImages:
+    """Resolve the agent and sidecar image refs for this plan.
+
+    Returns the committed snapshot if one exists, the local cached images
+    when the policy is 'cached', or builds fresh images and returns those."""
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return BottleImages(agent=committed, sidecar=SIDECAR_BUNDLE_IMAGE)
+    if plan.spec.image_policy == "cached":
+        if not docker_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        if not docker_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
+            die(
+                f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
+        return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+    docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
+
+
 @contextmanager
 def launch(
     plan: DockerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
 ) -> Generator[DockerBottle, None, None]:
-    """Build, launch, and provision a Docker bottle via compose.
-    Teardown on exit."""
+    """Launch and provision a Docker bottle via compose. Teardown on exit."""
     stack = ExitStack()
 
+    # Stamp the resolved agent image ref into the plan so compose rendering
+    # picks up the right image (may be a committed snapshot or cached ref).
+    plan = dataclasses.replace(
+        plan,
+        agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
+    )
+
     _bottle_for_revoke = plan.manifest.bottle
     _git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
 
@@ -95,22 +132,6 @@ def launch(
         )
 
     try:
-        # Step 1: agent image. Use a committed snapshot when one exists
-        # and is present in the local daemon; otherwise build from the
-        # Dockerfile. Sidecar images get built lazily by `docker compose
-        # up` via the renderer's `build:` directives.
-        committed = read_committed_image(plan.slug)
-        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,
-            )
 
         internal_network = network_mod.network_name_for_slug(plan.slug)
         egress_network = network_mod.network_egress_name_for_slug(plan.slug)
@@ -207,3 +228,21 @@ def launch(
         yield bottle
     finally:
         teardown()
+
+
+def stale_checks(plan: DockerBottlePlan) -> None:
+    """Raise StaleImageError if a cached image is older than the configured
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        check_stale(f"agent image {committed!r}", docker_mod.image_created_at(committed))
+        return
+    for label, ref in [
+        ("agent image", plan.image),
+        ("sidecar image", SIDECAR_BUNDLE_IMAGE),
+    ]:
+        if docker_mod.image_exists(ref):
+            check_stale(f"{label} {ref!r}", docker_mod.image_created_at(ref))

bot_bottle/backend/docker/util.py

@@ -4,6 +4,7 @@ existence, and building images."""
 
 from __future__ import annotations
 
+from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -186,6 +187,45 @@ def image_id(ref: str) -> str:
     return r.stdout.strip()
 
 
+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)
+
+
 def save(ref: str, output: str) -> None:
     """`docker save REF -o OUTPUT`. Writes a tarball of the image
     layers + manifest to the host path. Used by smolmachines

bot_bottle/backend/macos_container/backend.py

@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -67,11 +67,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(
-        self, plan: MacosContainerBottlePlan
+    def _launch_impl(
+        self, plan: MacosContainerBottlePlan, images: BottleImages
     ) -> Generator[MacosContainerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:

bot_bottle/backend/macos_container/launch.py

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

bot_bottle/backend/macos_container/util.py

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

bot_bottle/backend/smolmachines/backend.py

@@ -19,7 +19,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
@@ -77,11 +77,14 @@ class SmolmachinesBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _build_or_load_images(self, plan: SmolmachinesBottlePlan) -> BottleImages:
+        return _launch.build_or_load_images(plan)
+
     @contextmanager
-    def launch(
-        self, plan: SmolmachinesBottlePlan
+    def _launch_impl(
+        self, plan: SmolmachinesBottlePlan, images: BottleImages
     ) -> Generator[SmolmachinesBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str:

bot_bottle/backend/smolmachines/launch.py

@@ -52,6 +52,7 @@ from ...bottle_state import (
     git_gate_state_dir,
     read_committed_image,
 )
+from .. import BottleImages
 from . import loopback_alias as _loopback
 from . import sidecar_bundle as _bundle
 from . import smolvm as _smolvm
@@ -79,13 +80,20 @@ _GIT_HTTP_PORT = 9420
 _SUPERVISE_PORT = SUPERVISE_PORT
 
 
+def build_or_load_images(plan: SmolmachinesBottlePlan) -> BottleImages:
+    """Build or resolve the agent smolmachine artifact and sidecar bundle image."""
+    _bundle.ensure_bundle_image()
+    return BottleImages(agent=_agent_from_path(plan), sidecar=_bundle.SIDECAR_BUNDLE_IMAGE)
+
+
 @contextmanager
 def launch(
     plan: SmolmachinesBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
 ) -> Generator[SmolmachinesBottle, None, None]:
-    """Build + run the bottle and yield a handle; tear everything
+    """Run the bottle from pre-built images and yield a handle; tear everything
     down on exit. Errors during bringup unwind any partial state
     via the ExitStack."""
     stack = ExitStack()
@@ -96,9 +104,7 @@ def launch(
         plan = _start_bundle(plan, network, proxy_host, stack)
         plan = _discover_urls(plan, proxy_host)
 
-        agent_from_path = _agent_from_path(plan)
-
-        _launch_vm(plan, agent_from_path, proxy_host, stack)
+        _launch_vm(plan, Path(images.agent), proxy_host, stack)
         _init_vm(plan)
 
         bottle = SmolmachinesBottle(
@@ -184,7 +190,6 @@ def _start_bundle(
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
-    _bundle.ensure_bundle_image(bundle_spec.image)
     _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
     stack.callback(_bundle.stop_bundle, plan.slug)
     return plan

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
@@ -63,6 +64,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",
@@ -95,6 +104,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"
 
@@ -142,6 +153,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,
@@ -150,6 +165,7 @@ def cmd_start(argv: list[str]) -> int:
         label=label,
         color=color,
         bottle_names=bottle_names,
+        image_policy=image_policy,
     )
     return _launch_bottle(
         spec,
@@ -209,6 +225,8 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
+        headless=True,
+        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -389,6 +407,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)
@@ -531,6 +556,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, ...] = ()
@@ -549,10 +583,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 .supervise_types 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 supervise_types 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/git_gate_host_key.py

@@ -0,0 +1,268 @@
+"""Preflight host-key population for git-gate upstreams (issue #333).
+
+When a git-gate repo entry lacks a `host_key`, this module either:
+- headless: dies with a clear config error.
+- interactive: fetches the key via ssh-keyscan, prompts the operator to
+  confirm, and optionally persists it to the bottle config file on disk.
+
+Public entry point: `preflight_host_keys(manifest, headless=..., home_md=...)`.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import subprocess
+import sys
+from pathlib import Path
+from typing import cast
+
+from .log import die, info
+from .manifest import Manifest
+from .yaml_subset import YamlSubsetError, parse_frontmatter, serialize_yaml_subset
+
+
+# Preferred key types, most secure first.
+_KEY_TYPE_PREFERENCE = (
+    "ssh-ed25519",
+    "ecdsa-sha2-nistp256",
+    "ecdsa-sha2-nistp384",
+    "ecdsa-sha2-nistp521",
+    "ssh-rsa",
+)
+
+
+def fetch_host_key(host: str, port: str) -> str:
+    """Return an SSH public key for `host`:`port` via ssh-keyscan.
+
+    Returns the key in `<type> <base64-data>` format (the host prefix is
+    stripped so the result can be stored in `host_key` and later formatted
+    into a known_hosts line by `git_gate_known_hosts_line`).
+
+    Prefers ed25519 > ecdsa > rsa; falls back to the first key type
+    returned if none of the preferred types are present.
+
+    Raises `RuntimeError` on subprocess failure, timeout, or no result.
+    Uses only the Python stdlib (subprocess)."""
+    args = ["ssh-keyscan"]
+    if port and port != "22":
+        args += ["-p", port]
+    args.append(host)
+    try:
+        result = subprocess.run(
+            args, capture_output=True, text=True, timeout=15, check=False,
+        )
+    except OSError as e:
+        raise RuntimeError(
+            f"ssh-keyscan: could not launch for {host}:{port}: {e}"
+        ) from e
+    except subprocess.TimeoutExpired as e:
+        raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from e
+
+    # known_hosts format: "[host]:port type data"  or  "host type data"
+    # Strip the host/port prefix; collect "type -> type data" by type.
+    found: dict[str, str] = {}
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        parts = line.split(None, 2)
+        if len(parts) == 3 and parts[1] not in found:
+            found[parts[1]] = f"{parts[1]} {parts[2]}"
+
+    for preferred in _KEY_TYPE_PREFERENCE:
+        if preferred in found:
+            return found[preferred]
+    if found:
+        return next(iter(found.values()))
+
+    raise RuntimeError(
+        f"ssh-keyscan returned no host key for {host}:{port}."
+        + (f" stderr: {result.stderr.strip()!r}" if result.stderr.strip() else "")
+    )
+
+
+# ---------------------------------------------------------------------------
+# Frontmatter editing
+# ---------------------------------------------------------------------------
+
+
+def add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str:
+    """Return an updated copy of `file_text` with `host_key` set on the
+    named repo entry in the YAML frontmatter.
+
+    Parses the frontmatter into a dict, sets the key, and re-serializes.
+    Returns the original text unchanged when: the file has no frontmatter,
+    the git-gate.repos.<repo_name> entry is absent or already has a
+    host_key, or the frontmatter cannot be parsed."""
+    try:
+        fm, body = parse_frontmatter(file_text)
+    except YamlSubsetError:
+        return file_text
+
+    git_gate = fm.get("git-gate")
+    if not isinstance(git_gate, dict):
+        return file_text
+    repos = git_gate.get("repos")
+    if not isinstance(repos, dict):
+        return file_text
+    repo = repos.get(repo_name)
+    if not isinstance(repo, dict):
+        return file_text
+    if repo.get("host_key"):
+        return file_text
+
+    cast(dict[str, object], repo)["host_key"] = host_key
+    return f"---\n{serialize_yaml_subset(fm)}---\n{body}"
+
+
+def find_repo_bottle_file(bottles_dir: Path, repo_name: str) -> Path | None:
+    """Return the first `bottles_dir/*.md` that declares `repo_name` without
+    a `host_key`, without modifying anything.  Returns None if not found."""
+    if not bottles_dir.is_dir():
+        return None
+    for path in sorted(bottles_dir.glob("*.md")):
+        try:
+            text = path.read_text(encoding="utf-8")
+            fm, _ = parse_frontmatter(text)
+        except (OSError, UnicodeDecodeError, YamlSubsetError):
+            continue
+        git_gate = fm.get("git-gate")
+        if not isinstance(git_gate, dict):
+            continue
+        repos = git_gate.get("repos")
+        if not isinstance(repos, dict):
+            continue
+        repo = repos.get(repo_name)
+        if not isinstance(repo, dict) or repo.get("host_key"):
+            continue
+        return path
+    return None
+
+
+def find_and_update_bottle_file(
+    bottles_dir: Path, repo_name: str, host_key: str,
+) -> bool:
+    """Write `host_key` into the bottle file returned by `find_repo_bottle_file`.
+
+    Returns True on success, False when no suitable file is found or the
+    write fails."""
+    path = find_repo_bottle_file(bottles_dir, repo_name)
+    if path is None:
+        return False
+    try:
+        text = path.read_text(encoding="utf-8")
+    except (OSError, UnicodeDecodeError):
+        return False
+    updated = add_host_key_to_frontmatter(text, repo_name, host_key)
+    if updated == text:
+        return False
+    path.write_text(updated, encoding="utf-8")
+    info(f"wrote host_key for {repo_name!r} to {path}")
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Interactive prompt helper
+# ---------------------------------------------------------------------------
+
+
+def prompt_tty(message: str) -> str:
+    """Write `message` to stderr and read a line from /dev/tty (or stdin)."""
+    sys.stderr.write(message)
+    sys.stderr.flush()
+    try:
+        with open("/dev/tty", "r", encoding="utf-8") as tty:
+            return tty.readline().rstrip("\n")
+    except OSError:
+        return sys.stdin.readline().rstrip("\n")
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def preflight_host_keys(
+    manifest: Manifest,
+    *,
+    headless: bool,
+    home_md: Path | None,
+) -> Manifest:
+    """Ensure every git-gate repo entry has a `host_key` configured.
+
+    For entries whose `KnownHostKey` is empty:
+    - headless: calls `die()` with a clear message naming the repos.
+    - interactive: fetches the key via ssh-keyscan, shows it to the
+      operator, and requests confirmation.  If accepted, optionally
+      persists it to the bottle config file on disk; the key is always
+      applied in memory for this launch regardless of the persistence
+      choice.  Aborted confirmation calls `die()`.
+
+    Returns a (possibly updated) Manifest.  If all entries already have
+    host keys the original manifest is returned unchanged."""
+    bottle = manifest.bottle
+    missing = [e for e in bottle.git if not e.KnownHostKey]
+    if not missing:
+        return manifest
+
+    if headless:
+        names = ", ".join(repr(e.Name) for e in missing)
+        die(
+            f"git-gate: no host_key configured for repo(s) {names}. "
+            f"Add host_key to each bottle git-gate.repos entry, or run "
+            f"interactively once to have it fetched and saved automatically."
+        )
+
+    bottles_dir = (home_md / "bottles") if home_md is not None else None
+    updated_entries = list(bottle.git)
+
+    for entry in missing:
+        host = entry.UpstreamHost
+        port = entry.UpstreamPort
+        label = f"git-gate.repos[{entry.Name!r}]"
+
+        info(f"{label}: no host_key configured; fetching from {host}:{port}")
+        try:
+            key = fetch_host_key(host, port)
+        except RuntimeError as e:
+            die(f"git-gate: {label}: {e}")
+
+        sys.stderr.write(f"\ngit-gate: host key for {label}:\n  {key}\n\n")
+        confirm = prompt_tty("Is this host key correct? [y/N] ")
+        if confirm.strip().lower() not in ("y", "yes"):
+            die(f"git-gate: {label}: host key not confirmed; aborting launch")
+
+        if bottles_dir is not None:
+            target_file = find_repo_bottle_file(bottles_dir, entry.Name)
+            if target_file is not None:
+                save = prompt_tty(
+                    f"Save host_key for {entry.Name!r} to {target_file}? [y/N] "
+                )
+                if save.strip().lower() in ("y", "yes"):
+                    ok = find_and_update_bottle_file(bottles_dir, entry.Name, key)
+                    if not ok:
+                        sys.stderr.write(
+                            f"git-gate: {label}: could not write to {target_file}; "
+                            f"host_key kept in memory for this session only\n"
+                        )
+            else:
+                sys.stderr.write(
+                    f"git-gate: {label}: no bottle config file found for "
+                    f"{entry.Name!r}; host_key kept in memory for this session only\n"
+                )
+
+        idx = next(i for i, e in enumerate(updated_entries) if e.Name == entry.Name)
+        updated_entries[idx] = dataclasses.replace(entry, KnownHostKey=key)
+
+    updated_bottle = dataclasses.replace(bottle, git=tuple(updated_entries))
+    return dataclasses.replace(manifest, bottle=updated_bottle)
+
+
+__all__ = [
+    "fetch_host_key",
+    "preflight_host_keys",
+    "add_host_key_to_frontmatter",
+    "find_repo_bottle_file",
+    "find_and_update_bottle_file",
+    "prompt_tty",
+]

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/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
@@ -50,11 +52,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"]

bot_bottle/yaml_subset.py

@@ -21,6 +21,11 @@ Public API:
         For a Markdown file with YAML frontmatter delimited by `---`
         lines. Returns (frontmatter_dict, body_text).
 
+    serialize_yaml_subset(data) -> str
+        Serialize a dict (as produced by parse_yaml_subset) back to
+        block-style YAML text.  The result ends with a newline and
+        can be parsed back by parse_yaml_subset.
+
 What we accept (block-style):
 
     key: value                # mapping entry, value is inline
@@ -576,3 +581,105 @@ def parse_frontmatter(text: str) -> tuple[dict[str, object], str]:
     fm = parse_yaml_subset(fm_text)
     body = text[body_start:]
     return fm, body
+
+
+# --- Serializer -------------------------------------------------------------
+
+
+def _needs_quoting(s: str) -> bool:
+    """True when the string must be single-quoted to survive a round-trip."""
+    if not s:
+        return True
+    if s in ("true", "false", "null", "~") or s in _RESERVED_BOOL_LIKE:
+        return True
+    if (
+        _INT_RX.match(s)
+        or _DATE_RX.match(s)
+        or _OCTAL_RX.match(s)
+        or _HEX_RX.match(s)
+        or _FLOAT_RX.match(s)
+    ):
+        return True
+    # Characters that have special meaning at the start of a YAML value
+    if s[0] in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">", "%", "@", "`"):
+        return True
+    return False
+
+
+def _yaml_scalar(v: object) -> str:
+    """Serialize a scalar Python value to its YAML text form."""
+    if v is None:
+        return "null"
+    if isinstance(v, bool):
+        return "true" if v else "false"
+    if isinstance(v, int):
+        return str(v)
+    s = str(v)
+    if _needs_quoting(s):
+        return "'" + s.replace("'", "''") + "'"
+    return s
+
+
+def _serialize_node(node: object, indent: int) -> list[str]:
+    """Return lines (without trailing newlines) for `node` at `indent`.
+
+    Called only for non-empty dicts and lists (the caller guards with
+    `isinstance(val, (dict, list)) and val`), plus scalars at the leaf."""
+    prefix = " " * indent
+    if isinstance(node, dict):
+        out: list[str] = []
+        for key, val in node.items():
+            if isinstance(val, (dict, list)) and val:
+                out.append(f"{prefix}{key}:")
+                out.extend(_serialize_node(val, indent + 2))
+            else:
+                scalar = (
+                    "{}" if isinstance(val, dict)
+                    else "[]" if isinstance(val, list)
+                    else _yaml_scalar(val)
+                )
+                out.append(f"{prefix}{key}: {scalar}")
+        return out
+    if isinstance(node, list):
+        out = []
+        for item in node:
+            if isinstance(item, dict) and item:
+                entries = list(item.items())
+                first_key, first_val = entries[0]
+                if isinstance(first_val, (dict, list)) and first_val:
+                    out.append(f"{prefix}- {first_key}:")
+                    out.extend(_serialize_node(first_val, indent + 4))
+                else:
+                    scalar = (
+                        "{}" if isinstance(first_val, dict)
+                        else "[]" if isinstance(first_val, list)
+                        else _yaml_scalar(first_val)
+                    )
+                    out.append(f"{prefix}- {first_key}: {scalar}")
+                cont = prefix + "  "
+                for key, val in entries[1:]:
+                    if isinstance(val, (dict, list)) and val:
+                        out.append(f"{cont}{key}:")
+                        out.extend(_serialize_node(val, indent + 4))
+                    else:
+                        scalar = (
+                            "{}" if isinstance(val, dict)
+                            else "[]" if isinstance(val, list)
+                            else _yaml_scalar(val)
+                        )
+                        out.append(f"{cont}{key}: {scalar}")
+            else:
+                out.append(f"{prefix}- {_yaml_scalar(item)}")
+        return out
+    return [_yaml_scalar(node)]  # pragma: no cover
+
+
+def serialize_yaml_subset(data: dict[str, object]) -> str:
+    """Serialize `data` (as produced by parse_yaml_subset) to YAML text.
+
+    Produces block-style output with 2-space indentation. The result ends
+    with a newline and can be parsed back by parse_yaml_subset. Keys are
+    emitted in iteration order (insertion order in Python 3.7+)."""
+    if not data:
+        return ""
+    return "\n".join(_serialize_node(data, 0)) + "\n"

docs/research/landscape-containerized-claude.md

@@ -1,14 +1,20 @@
-# Landscape: containerized Claude Code agent tools
+# Landscape: containerized AI coding agent tools
 
 Research into whether bot-bottle is redundant with existing projects, and
 whether it's worth publishing.
 
 ## Summary
 
-The "Claude Code in Docker" space is active but not saturated. bot-bottle
-occupies a distinct position: no surveyed project combines all five of its
-defining features. Publishing is likely worthwhile, with the main risk being
-claudebox expanding to absorb the same niche.
+The "AI coding agents in isolated sandboxes" space is active but not saturated.
+bot-bottle occupies a distinct position: no surveyed project combines all five
+of its defining features. Publishing is likely worthwhile, with the main risk
+being claudebox expanding to absorb the same niche.
+
+**Updated 2026-07-09:** bot-bottle now supports three isolation backends
+(Docker, Apple `container`, smolmachines/libkrun microVMs) and three built-in
+agent providers (Claude Code, OpenAI Codex, Pi) with an open plugin system for
+arbitrary providers. This meaningfully strengthens the differentiation against
+all surveyed competitors.
 
 ## Closest competitor: claudebox
 
@@ -43,28 +49,78 @@ manifest merge.
   Still marked early-development.
 - **E2B, Northflank, Cloudflare Sandbox SDK** — cloud-hosted SaaS sandbox
   runtimes; fundamentally different architecture.
+- **superhq.ai / SuperHQ** (v0.4.4, April 2026) — macOS desktop app (Rust/GPUI)
+  that runs Claude Code, Codex, and Pi inside microVMs via Apple's
+  Virtualization.framework (their own shuru-sdk / libkrun). Auth gateway
+  injects API keys on the wire so the sandbox never sees them; tmpfs overlay
+  stages agent writes for diff-and-accept review; mobile remote access via
+  remote.superhq.ai. Early alpha, free on launch, Apple Silicon only.
+
+  Overlap: both projects cover agent isolation, credential proxying, and
+  multi-provider support (Claude Code / Codex / Pi). Differences: SuperHQ is a
+  GUI desktop app with no manifest layer; bot-bottle is a CLI fleet manager with
+  named agents, skills injection, per-agent system prompts, and cross-platform
+  backends (Docker, Apple `container`, smolmachines). SuperHQ's microVM
+  isolation story is now partially matched by bot-bottle's `macos_container` and
+  smolmachines backends. Worth watching — it targets the same security-minded
+  power-user audience and moves fast.
+
+  **Known gap in SuperHQ (user-requested, as of 2026-07-09):** A named user
+  (Brian Cheong, Founder, Dunialabs.io) explicitly called out the absence of
+  per-run audit logging: tool calls and network egress. Bot-bottle covers both:
+  network egress is logged by pipelock/mitmproxy, and per-run op-log/audit state
+  is persisted to SQLite.
 
 ## What no found project does
 
 None combine:
-1. Named-agent JSON manifest with per-agent env resolution (prompt / host-forward / literal)
-2. Claude Code skills directory injection
+1. Named-agent manifest with per-agent env resolution (prompt / host-forward / literal), supporting multiple providers (Claude Code, Codex, Pi, arbitrary plugins)
+2. Skills directory injection
 3. Per-agent system prompts
 4. SSH-agent key forwarding without copying private keys into the container
 5. Home + project manifest merge
+6. Pluggable isolation backends: Docker (Linux/macOS), Apple `container` (macOS microVMs), smolmachines/libkrun microVMs
+7. Per-run audit log: network egress via pipelock/mitmproxy + op-log persisted to SQLite
+
+**In-flight directions (not yet shipped):**
+
+- **Forge-native dispatch (issue #317):** Gitea webhook → orchestrator spins up a bottle
+  with the issue body as prompt → agent works → bottle freezes awaiting review comment →
+  rehydrates on comment → tears down on PR close. The issue-to-PR lifecycle concept is not
+  novel (Devin, Copilot Workspace, SWE-agent all do this as cloud services); what's
+  distinct is doing it self-hosted, manifest-driven, inside bot-bottle's isolation
+  primitives.
+- **Paid web control plane (issue #327):** Browser-based multi-host agent launch and
+  monitoring; account-scoped bottle and agent definitions; secret custody (encrypted at
+  rest, injected into the sidecar at launch, never exposed to the agent or returned by any
+  read API). Monetization model: OSS runtime free, control plane paid — a standard split
+  (HashiCorp, Grafana) applied to a self-hosted agent sandbox. The principled secret
+  custody model (agent never sees real credentials, even via printenv) is more rigorous
+  than most surveyed tools but not unprecedented.
 
 ## Publishing verdict
 
 Worth publishing. Differentiators that matter to the target audience (power
-users running parallel Claude Code sessions with distinct personas/tooling):
+users running parallel AI coding agent sessions with distinct personas/tooling):
 
-- The Python-stdlib-first, low-dependency design — competitors are npm-based or
-  Kubernetes-native.
+- The Python-stdlib-first, low-dependency design — competitors are npm-based,
+  Rust/GUI, or Kubernetes-native.
 - Named agents with distinct skills and system prompts, not just language profiles.
+- Multi-backend isolation: Docker, Apple `container` microVMs, and
+  smolmachines/libkrun — single manifest works across all three.
+- Multi-provider: Claude Code, Codex, Pi, plus an open plugin system for
+  arbitrary providers.
 - SSH forwarding without key copying.
+- Per-run audit log (tool calls + network egress) — an explicitly requested gap
+  in SuperHQ as of 2026-07-09.
+- Forge-native dispatch and a paid control plane (in flight) bring bot-bottle
+  into the same product category as cloud services like Devin and Copilot
+  Workspace — but self-hosted, with stronger isolation guarantees and a
+  manifest-driven fleet model those services don't have.
 
-Main risk: claudebox adds manifest/agent config. The space is moving fast
-enough that publishing sooner is better if establishing prior art matters.
+Main risk: claudebox adds manifest/agent config; SuperHQ is moving fast on the
+GUI / microVM side. The space is moving fast enough that publishing sooner is
+better if establishing prior art matters.
 
 Discovery will be slow without active promotion; an Anthropic Discord post or
 HN "Show HN" would do most of the work.
@@ -73,4 +129,4 @@ HN "Show HN" would do most of the work.
 
 - GitHub search cannot surface private or very new repos comprehensively.
 - Counts (stars, forks) were not confirmed for every project.
-- Research conducted 2026-05-07; the space moves fast.
+- Initial research conducted 2026-05-07; SuperHQ entry added 2026-07-09; the space moves fast.

requirements-dev.txt

@@ -3,5 +3,5 @@
 # These tools are used for code quality checks in CI/CD.
 
 pylint>=3.0.0
-pyright>=1.1.300
+pyright>=1.1.411
 coverage>=7.0.0

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_compose.py

@@ -118,6 +118,7 @@ def _plan(
     with_egress: bool = False,
     supervise: bool = False,
     canary: bool = False,
+    image_policy: str = "fresh",
 ) -> DockerBottlePlan:
     """Build a fully-resolved DockerBottlePlan. Toggles cover the
     matrix the renderer's conditional-service logic branches on."""
@@ -148,6 +149,7 @@ def _plan(
         agent_name="demo",
         copy_cwd=False,
         user_cwd="/tmp/x",
+        image_policy=image_policy,
     )
     return DockerBottlePlan(
         spec=spec,
@@ -265,7 +267,8 @@ class TestAgentAlwaysPresent(unittest.TestCase):
     def test_agent_depends_only_on_sidecars(self):
         # Bundle shape: the init supervisor owns intra-bundle daemon
         # ordering, so the agent waits on the bundle container alone.
-        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
+        cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
+        for kwargs in cases:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
@@ -300,6 +303,11 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
         self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
 
+    def test_cached_policy_omits_bundle_build(self):
+        sc = self._render(image_policy="cached")["services"]["sidecars"]
+        self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
+        self.assertNotIn("build", sc)
+
     def test_bundle_container_name_uses_sidecars_prefix(self):
         sc = self._render()["services"]["sidecars"]
         self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])

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
@@ -16,6 +17,7 @@ from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
+from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 
 
@@ -88,7 +90,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
         committed_tag: str | None = None,
         image_present: bool = True,
     ) -> list[str]:
-        """Drive launch() through its full sequence with the committed-image
+        """Drive build_or_load_images() + launch() with the committed-image
         behaviour controlled by the arguments. Returns the images that were
         passed to `build_image` (empty list if it was never called)."""
         built: list[str] = []
@@ -122,8 +124,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
+            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         return built
@@ -168,8 +171,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
+            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         self.assertEqual(1, len(captured_plans))
@@ -180,6 +184,24 @@ class TestLaunchCommittedImage(unittest.TestCase):
         built = self._run_launch(plan, committed_tag=None)
         self.assertEqual([_DEFAULT_IMAGE], built)
 
+    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:
         plan = _plan(self._tmp)
         built = self._run_launch(

tests/unit/test_docker_launch_teardown.py

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

tests/unit/test_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
@@ -54,6 +55,51 @@ class TestImageId(unittest.TestCase):
             self.assertIn("missing:tag", die.call_args.args[0])
 
 
+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 TestSave(unittest.TestCase):
     def test_save_runs_docker_save(self):
         with patch.object(

tests/unit/test_git_gate_host_key.py

@@ -0,0 +1,698 @@
+"""Unit tests for git_gate_host_key (issue #333)."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.git_gate_host_key import (
+    add_host_key_to_frontmatter,
+    find_and_update_bottle_file,
+    find_repo_bottle_file,
+    prompt_tty,
+    fetch_host_key,
+    preflight_host_keys,
+)
+from bot_bottle.manifest import Manifest, ManifestIndex
+
+
+# ---------------------------------------------------------------------------
+# fetch_host_key
+# ---------------------------------------------------------------------------
+
+
+class TestFetchHostKey(unittest.TestCase):
+    def _run_result(self, stdout: str, returncode: int = 0) -> MagicMock:
+        r = MagicMock()
+        r.stdout = stdout
+        r.stderr = ""
+        r.returncode = returncode
+        return r
+
+    def test_returns_type_and_key(self) -> None:
+        stdout = "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key)
+
+    def test_non_default_port_passes_p_flag(self) -> None:
+        stdout = "[gitea.example.com]:30009 ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run:
+            fetch_host_key("gitea.example.com", "30009")
+        args = mock_run.call_args[0][0]
+        self.assertIn("-p", args)
+        self.assertIn("30009", args)
+
+    def test_default_port_omits_p_flag(self) -> None:
+        stdout = "github.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run:
+            fetch_host_key("github.com", "22")
+        args = mock_run.call_args[0][0]
+        self.assertNotIn("-p", args)
+
+    def test_skips_comment_lines(self) -> None:
+        stdout = "# comment\ngithub.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("github.com", "22")
+        self.assertEqual("ssh-ed25519 AAAA", key)
+
+    def test_skips_lines_without_three_parts(self) -> None:
+        # A line with only two whitespace-separated tokens is not a valid
+        # known_hosts entry; it should be skipped and the function should
+        # raise rather than return a partial result.
+        stdout = "malformed-line\ngithub.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("github.com", "22")
+        self.assertEqual("ssh-ed25519 AAAA", key)
+
+    def test_raises_on_empty_output(self) -> None:
+        with patch("subprocess.run", return_value=self._run_result("")):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("no host key", str(cm.exception))
+
+    def test_prefers_ed25519_over_ecdsa(self) -> None:
+        stdout = (
+            "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n"
+            "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n"
+        )
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key)
+
+    def test_falls_back_to_ecdsa_without_ed25519(self) -> None:
+        stdout = "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ecdsa-sha2-nistp256 BBBBB", key)
+
+    def test_falls_back_to_rsa_without_ed25519_or_ecdsa(self) -> None:
+        stdout = "gitea.example.com ssh-rsa AAAAB3NzaC1yc2EAAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-rsa AAAAB3NzaC1yc2EAAAA", key)
+
+    def test_falls_back_to_first_for_unknown_type(self) -> None:
+        stdout = "gitea.example.com ssh-unknown CCCCC\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-unknown CCCCC", key)
+
+    def test_raises_includes_stderr_when_present(self) -> None:
+        r = self._run_result("")
+        r.stderr = "connection refused"
+        with patch("subprocess.run", return_value=r):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("connection refused", str(cm.exception))
+
+    def test_raises_on_os_error(self) -> None:
+        with patch("subprocess.run", side_effect=OSError("not found")):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("could not launch", str(cm.exception))
+
+    def test_raises_on_timeout(self) -> None:
+        import subprocess
+        with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("ssh-keyscan", 15)):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("timed out", str(cm.exception))
+
+
+# ---------------------------------------------------------------------------
+# prompt_tty
+# ---------------------------------------------------------------------------
+
+
+class TestPromptTty(unittest.TestCase):
+    def test_reads_from_tty_when_available(self) -> None:
+        import io
+        fake_tty = io.StringIO("yes\n")
+        with patch("builtins.open", return_value=fake_tty), patch("sys.stderr"):
+            result = prompt_tty("question: ")
+        self.assertEqual("yes", result)
+
+    def test_falls_back_to_stdin_on_os_error(self) -> None:
+        import io
+        with patch("builtins.open", side_effect=OSError("no tty")), \
+                patch("sys.stdin", io.StringIO("fallback\n")), \
+                patch("sys.stderr"):
+            result = prompt_tty("question: ")
+        self.assertEqual("fallback", result)
+
+
+# ---------------------------------------------------------------------------
+# add_host_key_to_frontmatter
+# ---------------------------------------------------------------------------
+
+
+class TestAddHostKeyToFrontmatter(unittest.TestCase):
+    _FILE = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+# Body text here
+"""
+
+    def test_inserts_host_key_at_repo_level(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        # Must be a sibling of `url` and `key`, not nested inside `key`
+        lines = result.splitlines()
+        hk_line = next(l for l in lines if "host_key" in l)
+        url_line = next(l for l in lines if "url:" in l)
+        self.assertEqual(
+            len(hk_line) - len(hk_line.lstrip()),
+            len(url_line) - len(url_line.lstrip()),
+        )
+
+    def test_preserves_body(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("# Body text here", result)
+
+    def test_preserves_other_repos(self) -> None:
+        text = """\
+---
+git-gate:
+  repos:
+    first:
+      url: ssh://git@host/org/first.git
+      key:
+        provider: static
+        path: /dev/null
+    second:
+      url: ssh://git@host/org/second.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+        result = add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        self.assertIn("second:", result)
+
+    def test_no_change_when_host_key_already_present(self) -> None:
+        text = self._FILE.replace(
+            "        path: /dev/null\n",
+            "        path: /dev/null\n      host_key: ssh-ed25519 EXISTING\n",
+        )
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW")
+        self.assertNotIn("NEW", result)
+
+    def test_no_change_when_repo_not_found(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA")
+        self.assertEqual(self._FILE, result)
+
+    def test_no_change_without_frontmatter(self) -> None:
+        text = "No frontmatter here\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_no_closing_delimiter(self) -> None:
+        text = "---\ngit-gate:\n  repos:\n    myrepo:\n      url: x\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_git_gate_missing(self) -> None:
+        text = "---\nenv:\n  FOO: bar\n---\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_repos_missing(self) -> None:
+        text = "---\ngit-gate:\n  foo: bar\n---\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_repo_entry_not_a_dict(self) -> None:
+        text = "---\ngit-gate:\n  repos:\n    - item\n---\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+
+# ---------------------------------------------------------------------------
+# find_and_update_bottle_file
+# ---------------------------------------------------------------------------
+
+
+class TestFindAndUpdateBottleFile(unittest.TestCase):
+    def _write_bottle(self, d: str, name: str, content: str) -> Path:
+        path = Path(d) / f"{name}.md"
+        path.write_text(content, encoding="utf-8")
+        return path
+
+    def test_updates_file_with_matching_repo(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+            path = self._write_bottle(d, "dev", content)
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertTrue(result)
+            updated = path.read_text()
+            self.assertIn("host_key: ssh-ed25519 AAAA", updated)
+
+    def test_returns_false_when_no_matching_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    other:
+      url: ssh://git@host/org/other.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+            self._write_bottle(d, "dev", content)
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_file_that_already_has_host_key(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+      host_key: ssh-ed25519 EXISTING
+---
+"""
+            path = self._write_bottle(d, "dev", content)
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 NEW")
+            self.assertFalse(result)
+            self.assertNotIn("NEW", path.read_text())
+
+    def test_returns_false_for_nonexistent_dir(self) -> None:
+        result = find_and_update_bottle_file(
+            Path("/nonexistent/dir"), "myrepo", "ssh-ed25519 AAAA"
+        )
+        self.assertFalse(result)
+
+    def test_skips_file_without_git_gate_section(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write_bottle(d, "plain", "---\nenv:\n  FOO: bar\n---\n")
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_file_with_non_dict_repos(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write_bottle(d, "bad", "---\ngit-gate:\n  repos:\n    - item\n---\n")
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_invalid_utf8_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "broken.md"
+            path.write_bytes(b"\xff\xfe")  # invalid UTF-8
+            result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_unreadable_file(self) -> None:
+        import os
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "unreadable.md"
+            path.write_text("---\n---\n")
+            os.chmod(path, 0o000)
+            try:
+                result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+                self.assertFalse(result)
+            finally:
+                os.chmod(path, 0o644)
+
+    def test_returns_false_when_file_unreadable_on_write(self) -> None:
+        # find_repo_bottle_file succeeds, but the second read (for writing)
+        # raises OSError — e.g. file removed or permissions changed.
+        content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+        with tempfile.TemporaryDirectory() as d:
+            path = self._write_bottle(d, "dev", content)
+            with patch(
+                "bot_bottle.git_gate_host_key.find_repo_bottle_file",
+                return_value=path,
+            ), patch.object(Path, "read_text", side_effect=OSError("Permission denied")):
+                result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_when_update_produces_no_change(self) -> None:
+        # add_host_key_to_frontmatter returns the original text unchanged
+        # (e.g. because the entry is absent from the parsed structure).
+        # find_and_update_bottle_file must continue rather than write.
+        content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+        with tempfile.TemporaryDirectory() as d:
+            path = self._write_bottle(d, "dev", content)
+            with patch(
+                "bot_bottle.git_gate_host_key.add_host_key_to_frontmatter",
+                return_value=content,  # no change
+            ):
+                result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+            self.assertNotIn("host_key", path.read_text())
+
+
+# ---------------------------------------------------------------------------
+# find_repo_bottle_file
+# ---------------------------------------------------------------------------
+
+
+class TestFindRepoBottleFile(unittest.TestCase):
+    _CONTENT = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+
+    def _write(self, d: str, name: str, content: str) -> Path:
+        path = Path(d) / f"{name}.md"
+        path.write_text(content, encoding="utf-8")
+        return path
+
+    def test_returns_path_when_repo_found(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            path = self._write(d, "dev", self._CONTENT)
+            result = find_repo_bottle_file(Path(d), "myrepo")
+            self.assertEqual(path, result)
+
+    def test_returns_none_when_no_matching_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write(d, "dev", self._CONTENT)
+            self.assertIsNone(find_repo_bottle_file(Path(d), "other"))
+
+    def test_returns_none_when_host_key_already_set(self) -> None:
+        content = self._CONTENT.replace("        path: /dev/null\n",
+                                        "        path: /dev/null\n      host_key: X\n")
+        with tempfile.TemporaryDirectory() as d:
+            self._write(d, "dev", content)
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_returns_none_for_nonexistent_dir(self) -> None:
+        self.assertIsNone(find_repo_bottle_file(Path("/nonexistent"), "myrepo"))
+
+    def test_skips_unreadable_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "unreadable.md").write_text(self._CONTENT)
+            with patch.object(Path, "read_text", side_effect=OSError("Permission denied")):
+                self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_skips_file_with_non_dict_git_gate(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "dev.md").write_text("---\ngit-gate:\n  - item\n---\n")
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_skips_file_with_non_dict_repos(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "dev.md").write_text("---\ngit-gate:\n  repos:\n    - item\n---\n")
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_save_prompt_includes_filename(self) -> None:
+        """preflight prompt must name the discovered bottle file."""
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            mock_prompt = MagicMock(side_effect=["y", "n"])  # confirm key, skip save
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key.prompt_tty", mock_prompt,
+            ), patch("sys.stderr"):
+                preflight_host_keys(manifest, headless=False, home_md=Path(home))
+
+            save_call = next(
+                c for c in mock_prompt.call_args_list if "Save" in c.args[0]
+            )
+            self.assertIn(str(bottle_file), save_call.args[0])
+
+
+# ---------------------------------------------------------------------------
+# preflight_host_keys
+# ---------------------------------------------------------------------------
+
+
+def _manifest_with_git(host_key: str = "") -> Manifest:
+    """Build a Manifest with one git entry; host_key defaults to empty."""
+    return ManifestIndex.from_json_obj({
+        "bottles": {
+            "dev": {
+                "git-gate": {
+                    "repos": {
+                        "myrepo": {
+                            "url": "ssh://git@gitea.example.com:30009/org/myrepo.git",
+                            "key": {"provider": "static", "path": "/dev/null"},
+                            **({"host_key": host_key} if host_key else {}),
+                        },
+                    },
+                },
+            },
+        },
+        "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+    }).load_for_agent("demo")
+
+
+class TestPreflightHostKeys(unittest.TestCase):
+    def test_no_op_when_all_keys_present(self) -> None:
+        manifest = _manifest_with_git("ssh-ed25519 AAAA")
+        result = preflight_host_keys(manifest, headless=False, home_md=None)
+        self.assertIs(result, manifest)
+
+    def test_headless_dies_when_key_missing(self) -> None:
+        manifest = _manifest_with_git()
+        with self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=True, home_md=None)
+
+    def test_interactive_fetches_and_confirms_key(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            return_value="ssh-ed25519 FETCHED",
+        ), patch(
+            "bot_bottle.git_gate_host_key.prompt_tty",
+            side_effect=["y", "n"],  # confirm key, don't persist
+        ), patch("sys.stderr"):
+            result = preflight_host_keys(manifest, headless=False, home_md=None)
+
+        self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+
+    def test_interactive_dies_when_key_not_confirmed(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            return_value="ssh-ed25519 FETCHED",
+        ), patch(
+            "bot_bottle.git_gate_host_key.prompt_tty",
+            return_value="n",
+        ), patch("sys.stderr"), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=False, home_md=None)
+
+    def test_interactive_persists_when_requested(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key.prompt_tty",
+                side_effect=["y", "y"],  # confirm key, yes persist
+            ), patch("sys.stderr"):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            written = bottle_file.read_text()
+            self.assertIn("host_key: ssh-ed25519 FETCHED", written)
+
+    def test_interactive_skips_persist_when_declined(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key.prompt_tty",
+                side_effect=["y", "n"],  # confirm key, no persist
+            ), patch("sys.stderr"):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            written = bottle_file.read_text()
+            self.assertNotIn("host_key", written)
+
+    def test_interactive_dies_when_fetch_fails(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            side_effect=RuntimeError("connection refused"),
+        ), patch("sys.stderr"), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=False, home_md=None)
+
+    def test_interactive_warns_when_persist_file_not_found(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            # bottles/ dir exists but does not contain the repo entry
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+
+            import io
+            buf = io.StringIO()
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key.prompt_tty",
+                side_effect=["y"],  # confirm key only; no save prompt when no file found
+            ), patch("sys.stderr", buf):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            self.assertIn("kept in memory", buf.getvalue())
+
+    def test_interactive_warns_when_write_fails_after_prompt(self) -> None:
+        # User says yes to save, but find_and_update_bottle_file fails.
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            import io
+            buf = io.StringIO()
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key.prompt_tty",
+                side_effect=["y", "y"],  # confirm key, yes persist
+            ), patch(
+                "bot_bottle.git_gate_host_key.find_and_update_bottle_file",
+                return_value=False,
+            ), patch("sys.stderr", buf):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            self.assertIn("kept in memory", buf.getvalue())
+
+    def test_headless_names_all_missing_repos_in_error(self) -> None:
+        manifest = ManifestIndex.from_json_obj({
+            "bottles": {
+                "dev": {
+                    "git-gate": {
+                        "repos": {
+                            "alpha": {
+                                "url": "ssh://git@host/org/alpha.git",
+                                "key": {"provider": "static", "path": "/dev/null"},
+                            },
+                            "beta": {
+                                "url": "ssh://git@host/org/beta.git",
+                                "key": {"provider": "static", "path": "/dev/null"},
+                            },
+                        },
+                    },
+                },
+            },
+            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+        }).load_for_agent("demo")
+
+        import io
+        buf = io.StringIO()
+        with patch("sys.stderr", buf), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=True, home_md=None)
+
+
+if __name__ == "__main__":
+    unittest.main()

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_launch.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import unittest
 import tempfile
 from pathlib import Path
@@ -15,6 +16,7 @@ from bot_bottle.backend.macos_container import launch
 from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
 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
 
 _MANIFEST = ManifestIndex.from_json_obj({
@@ -81,7 +83,7 @@ def _plan(
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(),
+        spec=SimpleNamespace(image_policy="fresh"),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
@@ -292,7 +294,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
 
 def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
     return MacosContainerBottlePlan(
-        spec=cast(BottleSpec, SimpleNamespace()),
+        spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -337,9 +339,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ), patch.object(launch, "info"):
-            updated = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", updated.image)
+        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
         self.assertEqual(1, len(calls))
         self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0])
 
@@ -358,12 +360,89 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ):
-            updated = launch._build_images(plan)
+            images = launch.build_or_load_images(plan)
 
-        self.assertEqual("bot-bottle-agent:latest", updated.image)
+        self.assertEqual("bot-bottle-agent:latest", images.agent)
         self.assertEqual(2, len(calls))
         self.assertEqual("bot-bottle-agent:latest", calls[1][0])
 
 
+class TestMacosContainerLaunchCachedImages(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.stage_dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _cached_plan(self) -> MacosContainerBottlePlan:
+        return dataclasses.replace(
+            _build_plan(self.stage_dir),
+            spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
+        )
+
+    def test_cached_mode_dies_when_agent_image_missing(self) -> None:
+        plan = self._cached_plan()
+
+        def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
+            return False  # Nothing cached.
+
+        with patch.object(
+            launch, "read_committed_image", return_value="",
+        ), patch.object(
+            launch.container_mod, "image_exists", side_effect=image_exists,
+        ), patch.object(
+            launch, "die", side_effect=Die(),
+        ):
+            with self.assertRaises(Die):
+                launch.build_or_load_images(plan)
+
+    def test_cached_mode_dies_when_sidecar_image_missing(self) -> None:
+        plan = self._cached_plan()
+
+        def image_exists(ref: str) -> bool:
+            return ref == plan.image  # Agent present; sidecar missing.
+
+        with patch.object(
+            launch, "read_committed_image", return_value="",
+        ), patch.object(
+            launch.container_mod, "image_exists", side_effect=image_exists,
+        ), patch.object(
+            launch, "die", side_effect=Die(),
+        ):
+            with self.assertRaises(Die):
+                launch.build_or_load_images(plan)
+
+    def test_cached_mode_both_present_returns_unchanged_plan(self) -> None:
+        plan = self._cached_plan()
+        with patch.object(
+            launch, "read_committed_image", return_value="",
+        ), patch.object(
+            launch.container_mod, "image_exists", return_value=True,
+        ), patch.object(
+            launch.container_mod, "build_image",
+        ) as build, patch.object(launch, "info"):
+            images = launch.build_or_load_images(plan)
+
+        build.assert_not_called()
+        self.assertEqual(plan.image, images.agent)
+
+    def test_committed_image_plus_cached_skips_sidecar_build(self) -> None:
+        plan = self._cached_plan()
+        with patch.object(
+            launch, "read_committed_image",
+            return_value="bot-bottle-committed-dev-abc:latest",
+        ), patch.object(
+            launch.container_mod, "image_exists", return_value=True,
+        ), patch.object(
+            launch.container_mod, "build_image",
+        ) as build, patch.object(launch, "info"):
+            images = launch.build_or_load_images(plan)
+
+        # In cached mode with a committed image, no builds should fire.
+        build.assert_not_called()
+        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_container_util.py

@@ -273,5 +273,71 @@ resolver #2
             )
 
 
+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,206 @@
+"""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_and_sidecar(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=""), \
+             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())
+        # Should have been called for both agent and sidecar images.
+        self.assertEqual(2, cs.call_count)
+
+    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_and_sidecar(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=""), \
+             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())
+        self.assertEqual(2, cs.call_count)
+
+    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)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_yaml_subset.py

@@ -6,7 +6,7 @@ import textwrap
 import unittest
 
 from bot_bottle.yaml_subset import YamlSubsetError
-from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
+from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset, serialize_yaml_subset
 
 
 def _y(s: str):
@@ -457,5 +457,141 @@ class TestEdgeAndErrorBranches(unittest.TestCase):
         self.assertEqual(({}, ""), parse_frontmatter(""))
 
 
+# ---------------------------------------------------------------------------
+# serialize_yaml_subset
+# ---------------------------------------------------------------------------
+
+
+class TestSerializeYamlSubset(unittest.TestCase):
+    def _roundtrip(self, text: str) -> None:
+        """Parse then serialize then re-parse; result must equal original parse."""
+        parsed = parse_yaml_subset(text)
+        serialized = serialize_yaml_subset(parsed)
+        self.assertEqual(parsed, parse_yaml_subset(serialized))
+
+    # --- scalar values ---------------------------------------------------------
+
+    def test_string_bare(self) -> None:
+        self.assertEqual("key: value\n", serialize_yaml_subset({"key": "value"}))
+
+    def test_string_empty_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": ""})
+        self.assertIn("key: ''", out)
+        self._roundtrip("key: ''\n")
+
+    def test_string_true_like_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": "true"})
+        self.assertIn("'true'", out)
+        self.assertEqual({"key": "true"}, parse_yaml_subset(out))
+
+    def test_string_int_like_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": "42"})
+        self.assertIn("'42'", out)
+        self.assertEqual({"key": "42"}, parse_yaml_subset(out))
+
+    def test_string_special_start_char_is_quoted(self) -> None:
+        for ch in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">"):
+            out = serialize_yaml_subset({"key": ch + "rest"})
+            self.assertIn("'", out, msg=f"expected quoting for {ch!r}")
+
+    def test_none_emits_null(self) -> None:
+        self.assertEqual("key: null\n", serialize_yaml_subset({"key": None}))
+
+    def test_bool_true(self) -> None:
+        self.assertEqual("key: true\n", serialize_yaml_subset({"key": True}))
+
+    def test_bool_false(self) -> None:
+        self.assertEqual("key: false\n", serialize_yaml_subset({"key": False}))
+
+    def test_int(self) -> None:
+        self.assertEqual("key: 42\n", serialize_yaml_subset({"key": 42}))
+
+    # --- nested dict -----------------------------------------------------------
+
+    def test_nested_dict(self) -> None:
+        data: dict[str, object] = {"outer": {"inner": "val"}}
+        out = serialize_yaml_subset(data)
+        self.assertIn("outer:", out)
+        self.assertIn("  inner: val", out)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_empty_dict_value_inline(self) -> None:
+        out = serialize_yaml_subset({"key": {}})
+        self.assertIn("key: {}", out)
+
+    # --- lists -----------------------------------------------------------------
+
+    def test_list_of_scalars(self) -> None:
+        data: dict[str, object] = {"items": ["a", "b", "c"]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_empty_list_value_inline(self) -> None:
+        out = serialize_yaml_subset({"key": []})
+        self.assertIn("key: []", out)
+
+    def test_list_of_mappings(self) -> None:
+        data: dict[str, object] = {"items": [{"name": "alpha", "val": 1}, {"name": "beta", "val": 2}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_mapping_with_nested_dict(self) -> None:
+        data: dict[str, object] = {"items": [{"name": "x", "sub": {"k": "v"}}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_first_key_is_nested_dict(self) -> None:
+        # Covers the branch where the FIRST key of a mapping list item
+        # has a nested dict/list value (not a scalar).
+        data: dict[str, object] = {"items": [{"cfg": {"a": 1, "b": 2}, "name": "x"}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_mapping_continuation_keys(self) -> None:
+        data: dict[str, object] = {"items": [{"a": 1, "b": 2, "c": 3}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    # --- empty / trivial -------------------------------------------------------
+
+    def test_empty_dict(self) -> None:
+        self.assertEqual("", serialize_yaml_subset({}))
+
+    # --- realistic bottle frontmatter round-trip --------------------------------
+
+    def test_bottle_frontmatter_round_trip(self) -> None:
+        text = textwrap.dedent("""\
+            git-gate:
+              repos:
+                myrepo:
+                  url: ssh://git@host/org/repo.git
+                  key:
+                    provider: static
+                    path: /dev/null
+        """)
+        self._roundtrip(text)
+
+    def test_host_key_inserted_at_repo_level(self) -> None:
+        """The canonical use-case: host_key lands as a sibling of url/key."""
+        from typing import cast
+        text = textwrap.dedent("""\
+            git-gate:
+              repos:
+                myrepo:
+                  url: ssh://git@host/org/repo.git
+                  key:
+                    provider: static
+                    path: /dev/null
+        """)
+        data = parse_yaml_subset(text)
+        repos = cast(dict[str, object], cast(dict[str, object], data["git-gate"])["repos"])
+        cast(dict[str, object], repos["myrepo"])["host_key"] = "ssh-ed25519 AAAA"
+        out = serialize_yaml_subset(data)
+        parsed_back = parse_yaml_subset(out)
+        repo = cast(dict[str, object], cast(dict[str, object], cast(dict[str, object], parsed_back["git-gate"])["repos"])["myrepo"])
+        self.assertEqual("ssh-ed25519 AAAA", repo["host_key"])
+        self.assertIn("provider", cast(dict[str, object], repo["key"]))
+
+
 if __name__ == "__main__":
     unittest.main()