fix(smolmachines): enforce per-bottle loopback isolation on Linux via iptables

On Linux, smolvm's TSI allowlist DB patch crashes boot, so
force_allowlist is a no-op. This left the agent VM with full
host loopback access — a security regression from the macOS
behavior.

Fix: install guest-side iptables rules in _init_vm that ACCEPT
traffic to the bottle's allocated loopback alias (proxy_host/32)
and DROP all other 127.0.0.0/8 destinations. The agent runs as
non-root (node) and cannot flush the rules.

- Add iptables to claude and codex agent Dockerfiles
- Use DROP (not REJECT) — libkrun kernel lacks the REJECT module
- Skip on macOS where force_allowlist handles it natively

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

Dockerfile.sidecars

@@ -14,9 +14,10 @@
 #   /app/supervise_server.py + .py       supervise MCP server
 #   /app/sidecar_init.py                 PID 1 supervisor
 #   /etc/egress/routes.yaml              bind-mounted at run time
-#   /etc/git-gate/pre-receive            docker-cp'd at start time
-#   /git-gate-entrypoint.sh              docker-cp'd at start time
-#   /git-gate/creds/*                    docker-cp'd at start time
+#   /etc/git-gate/entrypoint.sh          per-bottle (docker-cp or virtiofs mount)
+#   /etc/git-gate/pre-receive            per-bottle (docker-cp or virtiofs mount)
+#   /git-gate-entrypoint.sh              static wrapper → /etc/git-gate/entrypoint.sh
+#   /git-gate/creds/*                    per-bottle (docker-cp or virtiofs mount)
 #   /git/*                               bare repos, populated at runtime
 #   /run/supervise/bot-bottle.db         bind-mounted at run time
 #   /home/mitmproxy/.mitmproxy/          mitmproxy CA dir
@@ -88,7 +89,21 @@ RUN mkdir -p \
       /git-gate/creds \
       /git \
       /run/supervise \
-      /home/mitmproxy/.mitmproxy
+      /home/mitmproxy/.mitmproxy \
+      /bot-bottle-data/egress \
+      /bot-bottle-data/git-gate
+
+# Static wrapper for the git-gate entrypoint. The per-bottle
+# entrypoint script is either:
+#   - docker-cp'd to /git-gate-entrypoint.sh (docker/macOS backends),
+#     which overwrites this wrapper; or
+#   - virtiofs-mounted at /bot-bottle-data/git-gate/entrypoint.sh
+#     (smolmachines), where this wrapper delegates to it at runtime.
+#     Fallback to /etc/git-gate/ for backwards compatibility.
+# Either way sidecar_init.py calls `/bin/sh /git-gate-entrypoint.sh`.
+RUN printf '#!/bin/sh\nif [ -x /bot-bottle-data/git-gate/entrypoint.sh ]; then\n  exec /bot-bottle-data/git-gate/entrypoint.sh "$@"\nfi\nexec /etc/git-gate/entrypoint.sh "$@"\n' \
+      > /git-gate-entrypoint.sh \
+    && chmod 755 /git-gate-entrypoint.sh
 
 # Documentation only — the compose renderer publishes whichever
 # subset the bottle uses.

bot_bottle/backend/__init__.py

@@ -36,10 +36,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager, contextmanager
+from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generator, Generic, Sequence, TypeVar
+from typing import Any, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -78,9 +78,6 @@ class BottleSpec:
     # True when launched via --headless (no TTY, no interactive prompts).
     # The git-gate host-key preflight uses this to error rather than prompt.
     headless: bool = False
-    # Image startup policy. "fresh" preserves the normal build path;
-    # "cached" reuses the current local image/artifact without rebuilding.
-    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -435,25 +432,8 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
-    @contextmanager
-    def launch(
-        self, plan: PlanT, *, skip_stale: bool = False
-    ) -> Generator[Bottle, None, None]:
-        """Template: optionally check for stale cached images, then delegate
-        to `_launch_impl`. Pass `skip_stale=True` to bypass the stale check
-        (used by the interactive CLI after the operator confirms)."""
-        if not skip_stale:
-            self._image_stale_checks(plan)
-        with self._launch_impl(plan) as bottle:
-            yield bottle
-
-    def _image_stale_checks(self, plan: PlanT) -> None:
-        """Raise StaleImageError if any cached image used by this plan is stale.
-        No-op default; backends override to call the shared `check_stale*`
-        helpers on their image/artifact timestamps."""
-
     @abstractmethod
-    def _launch_impl(self, plan: PlanT) -> AbstractContextManager[Bottle]:
+    def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
         """Build/run the bottle and yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:

bot_bottle/backend/docker/backend.py

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

bot_bottle/backend/docker/compose.py

@@ -180,6 +180,10 @@ 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},
@@ -188,11 +192,6 @@ 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,8 +41,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import check_stale
-from ...log import die, info, warn
+from ...log import info, warn
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -64,7 +63,6 @@ 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.
@@ -102,26 +100,12 @@ def launch(
         # Dockerfile. Sidecar images get built lazily by `docker compose
         # up` via the renderer's `build:` directives.
         committed = read_committed_image(plan.slug)
-        cached_policy = plan.spec.image_policy == "cached"
         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),
             )
-        elif cached_policy:
-            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}")
         else:
             docker_mod.build_image(
                 plan.image, _REPO_DIR,
@@ -223,21 +207,3 @@ 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,7 +4,6 @@ existence, and building images."""
 
 from __future__ import annotations
 
-from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -187,45 +186,6 @@ 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

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

bot_bottle/backend/macos_container/launch.py

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

bot_bottle/backend/macos_container/util.py

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

bot_bottle/backend/smolmachines/backend.py

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

bot_bottle/backend/smolmachines/egress_apply.py

@@ -7,16 +7,23 @@ exec`` instead of Docker.
 
 from __future__ import annotations
 
-from ...egress import EGRESS_ROUTES_IN_CONTAINER
+from pathlib import Path
+
+from ...bottle_state import egress_state_dir
 from ...log import warn
 from ..egress_apply import EgressApplicator, EgressApplyError
 from . import sidecar_bundle as _bundle
 from . import smolvm as _smolvm
 
+# Routes file path inside the sidecar VM. Set via EGRESS_ROUTES env var
+# at launch so the addon reads from the virtiofs-mounted confdir instead
+# of the default /etc/egress/routes.yaml.
+_EGRESS_ROUTES_IN_SIDECAR_VM = "/bot-bottle-data/egress/routes.yaml"
+
 
 def fetch_current_routes(slug: str) -> str:
     machine = _bundle.bundle_machine_name(slug)
-    result = _smolvm.machine_exec(machine, ["cat", EGRESS_ROUTES_IN_CONTAINER])
+    result = _smolvm.machine_exec(machine, ["cat", _EGRESS_ROUTES_IN_SIDECAR_VM])
     if result.returncode != 0:
         raise EgressApplyError(
             f"could not read routes.yaml from {machine}: "
@@ -26,6 +33,14 @@ def fetch_current_routes(slug: str) -> str:
 
 
 class SmolmachinesEgressApplicator(EgressApplicator):
+    @staticmethod
+    def _routes_path(slug: str) -> Path:
+        # Routes live in the smolvm-sidecar-data staging dir, which is
+        # virtiofs-mounted at /bot-bottle-data/ in the sidecar VM. Writes
+        # here are visible inside the VM immediately; a SIGHUP causes the
+        # addon to reload from _EGRESS_ROUTES_IN_SIDECAR_VM.
+        return egress_state_dir(slug) / "smolvm-sidecar-data" / "egress" / "routes.yaml"
+
     def _signal_bundle_reload(self, slug: str) -> None:
         machine = _bundle.bundle_machine_name(slug)
         result = _smolvm.machine_exec(machine, ["sh", "-c", "kill -HUP 1"])

bot_bottle/backend/smolmachines/launch.py

@@ -14,12 +14,12 @@ from __future__ import annotations
 
 import dataclasses
 import os
+import shutil
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
 from ...egress import (
-    EGRESS_ROUTES_IN_CONTAINER,
     egress_agent_env_entries,
     egress_resolve_token_values,
     egress_sidecar_env_entries,
@@ -28,22 +28,14 @@ from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker import util as docker_mod
 from ..docker.egress import (
-    EGRESS_CA_IN_CONTAINER,
     EGRESS_PORT as _EGRESS_PORT,
     egress_tls_init,
 )
-from ..docker.git_gate import (
-    GIT_GATE_ACCESS_HOOK_IN_CONTAINER,
-    GIT_GATE_CREDS_DIR_IN_CONTAINER,
-    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
-    GIT_GATE_HOOK_IN_CONTAINER,
-)
 from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import check_stale_path
-from ...log import die, info, warn
+from ...log import info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
@@ -62,6 +54,18 @@ from .local_registry import crane_push_tarball, ephemeral_registry
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 
 
+# Single virtiofs mount for egress + git-gate files. libkrun limits
+# the total of mounts + port-mappings to 5; with 3 daemon ports the
+# sidecar VM can carry at most 2 mounts. Egress CA/routes and
+# git-gate scripts/creds are staged into subdirectories of one host
+# dir and mounted here. Env vars (EGRESS_CONFDIR, EGRESS_ROUTES,
+# and the Dockerfile's git-gate wrapper) point each daemon at its
+# subdirectory.
+_SIDECAR_DATA_DIR_IN_VM = "/bot-bottle-data"
+_EGRESS_CONFDIR_IN_VM = f"{_SIDECAR_DATA_DIR_IN_VM}/egress"
+_GIT_GATE_SCRIPTS_DIR_IN_VM = f"{_SIDECAR_DATA_DIR_IN_VM}/git-gate"
+
+
 # Per-host cache for `smolvm pack create` outputs. Keyed by the
 # docker image ID so a Dockerfile change automatically invalidates
 # the cache. `pack create` is idempotent on the smolvm side but
@@ -95,7 +99,7 @@ def launch(
         agent_from_path = _agent_from_path(plan)
 
         _launch_vm(plan, agent_from_path, proxy_host, stack)
-        _init_vm(plan)
+        _init_vm(plan, proxy_host)
 
         bottle = SmolmachinesBottle(
             plan.machine_name,
@@ -179,13 +183,10 @@ 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))
-    if _image_policy(plan) == "cached":
-        artifact = _cached_smolmachine(bundle_spec.image, label="sidecar")
-    else:
-        artifact = _ensure_smolmachine(
-            bundle_spec.image,
-            dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-        )
+    artifact = _ensure_smolmachine(
+        bundle_spec.image,
+        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
+    )
     launch = _bundle.start_bundle_vm(
         bundle_spec,
         from_path=artifact,
@@ -295,6 +296,16 @@ def _launch_vm(
     fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
     makes --from and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
+    # Destroy any leftover machine from a previous run that didn't
+    # clean up (e.g. crash, interrupted teardown).
+    try:
+        _smolvm.machine_stop(plan.machine_name)
+    except _smolvm.SmolvmError:
+        pass
+    try:
+        _smolvm.machine_delete(plan.machine_name)
+    except _smolvm.SmolvmError:
+        pass
     _smolvm.machine_create(
         plan.machine_name,
         from_path=agent_from_path,
@@ -311,17 +322,24 @@ def _launch_vm(
     stack.callback(_smolvm.machine_stop, plan.machine_name)
 
 
-def _init_vm(plan: SmolmachinesBottlePlan) -> None:
-    """Repair filesystem ownership and wait for exec channel readiness.
+def _init_vm(plan: SmolmachinesBottlePlan, proxy_host: str) -> None:
+    """Repair filesystem ownership, enforce loopback isolation, and
+    wait for exec channel readiness.
 
     Ownership repair: smolvm's pack process remaps files to the host
     invoker's uid (e.g. 501 on macOS, 1000 on Linux). The chowns use
     names not numbers so they're correct on either. /home/node must
-    be node:node so
-    Claude Code can write ~/.claude.json; /tmp + /var/tmp need root
-    mode 1777 so non-root processes can create per-uid scratch dirs.
-    All folded into one sh -c to avoid back-to-back exec calls
-    immediately after machine_start (libkrun exec-channel race).
+    be node:node so Claude Code can write ~/.claude.json; /tmp +
+    /var/tmp need root mode 1777 so non-root processes can create
+    per-uid scratch dirs.
+
+    Loopback isolation (Linux only): on macOS, smolvm's TSI allowlist
+    restricts which host loopback IPs the guest can reach. On Linux,
+    the allowlist DB patch crashes TSI boot, so we enforce the same
+    /32 scope with guest-side iptables instead. The rules allow
+    outbound to proxy_host/32, then reject all other 127.0.0.0/8.
+    Since the agent runs as non-root (node), it cannot flush these
+    rules.
 
     mkdir -p guards: when booting from a committed snapshot, /tmp and
     /var/tmp are excluded from the archive (they're ephemeral and their
@@ -337,9 +355,41 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
         "chown root:root /tmp /var/tmp && "
         "chmod 1777 /tmp /var/tmp",
     ])
+    if not _loopback._is_macos():
+        _enforce_loopback_isolation(plan.machine_name, proxy_host)
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
+def _enforce_loopback_isolation(machine_name: str, proxy_host: str) -> None:
+    """Install iptables rules restricting the guest to proxy_host/32.
+
+    On Linux, smolvm's TSI bridges the full host loopback into the
+    guest, and the allow-cidr DB patch is incompatible with TSI.
+    Guest-side iptables achieves the same per-bottle isolation:
+    only the allocated loopback alias is reachable, so the agent
+    can't probe other bottles' ports or other host services.
+
+    Runs as root (exec default); the agent process runs as `node`
+    (non-root) and cannot modify iptables rules."""
+    result = _smolvm.machine_exec(machine_name, [
+        "sh", "-c",
+        # Allow the bottle's allocated loopback alias.
+        f"iptables -A OUTPUT -d {proxy_host}/32 -j ACCEPT && "
+        # Drop all other loopback destinations. DROP (not REJECT)
+        # because the libkrun kernel lacks the REJECT target module.
+        # Connections to blocked addresses time out rather than fail
+        # fast, which is acceptable — the agent shouldn't be probing
+        # non-allowed loopback addresses.
+        "iptables -A OUTPUT -d 127.0.0.0/8 -j DROP",
+    ])
+    if result.returncode != 0:
+        warn(
+            f"guest iptables setup failed (exit {result.returncode}): "
+            f"{(result.stderr or '').strip() or '<no stderr>'}. "
+            f"Per-bottle loopback isolation is not enforced."
+        )
+
+
 def _label_for_port(port: int) -> str:
     if port == _EGRESS_PORT:
         return "egress"
@@ -362,6 +412,62 @@ def _port_for_label(label: str) -> int:
     raise ValueError(f"unknown sidecar forward label: {label}")
 
 
+def _stage_sidecar_data(plan: SmolmachinesBottlePlan) -> Path:
+    """Stage egress + git-gate files into one virtiofs-mountable dir.
+
+    libkrun limits total mounts + port-mappings to 5. With 3 daemon
+    ports the sidecar VM can carry at most 2 mounts (the second is
+    the supervise DB). Egress and git-gate share a single mount:
+
+        <staging>/egress/     → _EGRESS_CONFDIR_IN_VM
+        <staging>/git-gate/   → _GIT_GATE_SCRIPTS_DIR_IN_VM
+
+    The mount is writable so mitmproxy can write combined-trust.pem
+    and cache per-host certs under the egress subdir."""
+    staging = egress_state_dir(plan.slug) / "smolvm-sidecar-data"
+
+    # --- egress subdir ---
+    confdir = staging / "egress"
+    confdir.mkdir(parents=True, exist_ok=True)
+    ep = plan.egress_plan
+    shutil.copy2(str(ep.mitmproxy_ca_host_path), str(confdir / "mitmproxy-ca.pem"))
+    if ep.routes:
+        shutil.copy2(str(ep.routes_path), str(confdir / "routes.yaml"))
+
+    # --- git-gate subdir (only when upstreams are configured) ---
+    gp = plan.git_gate_plan
+    if gp.upstreams:
+        scripts_dir = staging / "git-gate"
+        scripts_dir.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(str(gp.entrypoint_script), str(scripts_dir / "entrypoint.sh"))
+        shutil.copy2(str(gp.hook_script), str(scripts_dir / "pre-receive"))
+        shutil.copy2(str(gp.access_hook_script), str(scripts_dir / "access-hook"))
+        for name in ("entrypoint.sh", "pre-receive", "access-hook"):
+            (scripts_dir / name).chmod(0o755)
+
+        # Patch paths: the rendered entrypoint hardcodes /git-gate/creds/
+        # and /etc/git-gate/ for hooks; rewrite both to the in-VM subdir.
+        ep_path = scripts_dir / "entrypoint.sh"
+        text = ep_path.read_text()
+        text = text.replace("/git-gate/creds/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/creds/")
+        text = text.replace("/etc/git-gate/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/")
+        ep_path.write_text(text)
+
+        creds_dir = scripts_dir / "creds"
+        creds_dir.mkdir(exist_ok=True)
+        for u in gp.upstreams:
+            keypath = Path(expand_tilde(u.identity_file))
+            dest_key = creds_dir / f"{u.name}-key"
+            shutil.copy2(str(keypath), str(dest_key))
+            dest_key.chmod(0o600)
+            if u.known_hosts_file:
+                dest_kh = creds_dir / f"{u.name}-known_hosts"
+                shutil.copy2(str(u.known_hosts_file), str(dest_kh))
+                dest_kh.chmod(0o600)
+
+    return staging
+
+
 def _bundle_launch_spec(
     plan: SmolmachinesBottlePlan, network: str, proxy_host: str,
 ) -> _bundle.BundleLaunchSpec:
@@ -380,35 +486,26 @@ def _bundle_launch_spec(
     env: list[str] = []
     volumes: list[tuple[str, str, bool]] = []
 
-    # --- egress -----------------------------------------------
+    # --- egress + git-gate (single mount) ---------------------
+    # Stage both into one dir and mount it at _SIDECAR_DATA_DIR_IN_VM.
+    # libkrun limits mounts + port-mappings to 5; with 3 daemon ports
+    # we can carry at most 2 mounts (this one + supervise DB).
+    # Writable so egress_entrypoint.sh can write combined-trust.pem
+    # and mitmproxy can create its per-host cert cache.
     ep = plan.egress_plan
-    volumes.append((str(ep.mitmproxy_ca_host_path), EGRESS_CA_IN_CONTAINER, True))
-    if ep.routes:
-        volumes.append((str(ep.routes_path.parent), str(Path(EGRESS_ROUTES_IN_CONTAINER).parent), True))
+    gp = plan.git_gate_plan
+    staging = _stage_sidecar_data(plan)
+    volumes.append((str(staging), _SIDECAR_DATA_DIR_IN_VM, False))
+    # Tell the egress entrypoint where to find its CA + routes.
+    env.append(f"EGRESS_CONFDIR={_EGRESS_CONFDIR_IN_VM}")
+    # Always set EGRESS_ROUTES so the addon reads from the confdir path
+    # even when no routes were configured at launch (apply_routes_change
+    # writes here and a SIGHUP causes the addon to pick them up).
+    env.append(f"EGRESS_ROUTES={_EGRESS_CONFDIR_IN_VM}/routes.yaml")
     env.extend(egress_sidecar_env_entries(ep))
 
-    # --- git-gate ---------------------------------------------
-    gp = plan.git_gate_plan
     if gp.upstreams:
         daemons += ["git-gate", "git-http"]
-        volumes += [
-            (str(gp.entrypoint_script), GIT_GATE_ENTRYPOINT_IN_CONTAINER, True),
-            (str(gp.hook_script), GIT_GATE_HOOK_IN_CONTAINER, True),
-            (str(gp.access_hook_script), GIT_GATE_ACCESS_HOOK_IN_CONTAINER, True),
-        ]
-        for u in gp.upstreams:
-            keypath = expand_tilde(u.identity_file)
-            volumes.append((
-                keypath,
-                f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-key",
-                True,
-            ))
-            if u.known_hosts_file:
-                volumes.append((
-                    str(u.known_hosts_file),
-                    f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-known_hosts",
-                    True,
-                ))
 
     # --- supervise --------------------------------------------
     sp = plan.supervise_plan
@@ -419,7 +516,13 @@ def _bundle_launch_spec(
             f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
             f"SUPERVISE_PORT={SUPERVISE_PORT}",
         ]
-        volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
+        # virtiofs requires directory mount — mount the DB's parent
+        # dir so bot-bottle.db lands at the right in-VM path.
+        volumes.append((
+            str(sp.db_path.parent),
+            str(Path(DB_PATH_IN_CONTAINER).parent),
+            False,
+        ))
 
     # Container ports the agent reaches from the smolvm guest —
     # published on `proxy_host` so the TSI allowlist and the docker
@@ -469,9 +572,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
             info(f"using committed smolmachine {str(committed_path)!r}")
             return committed_path
 
-    if _image_policy(plan) == "cached":
-        return _cached_smolmachine(plan.agent_image, label="agent")
-
     # Build the agent image and pack it into a `.smolmachine`
     # artifact (or hit the per-Dockerfile-digest cache). Runs here,
     # not in prepare, so the docker-build output doesn't garble the
@@ -482,60 +582,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
     )
 
 
-def stale_checks(plan: SmolmachinesBottlePlan) -> None:
-    """Raise StaleImageError if a cached smolmachine artifact is older than the
-    configured threshold. Only runs when image_policy is 'cached'. Checks the
-    committed agent artifact first, then the cache-keyed agent and sidecar
-    artifacts. Called by the backend class's _image_stale_checks before any
-    resources are allocated."""
-    if _image_policy(plan) != "cached":
-        return
-    committed = read_committed_image(plan.slug)
-    if committed:
-        committed_path = Path(committed)
-        if committed_path.is_file():
-            check_stale_path("agent smolmachine artifact", committed_path)
-    elif docker_mod.image_exists(plan.agent_image):
-        digest = docker_mod.image_id(plan.agent_image).split(":", 1)[-1][:16]
-        artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-        if artifact.is_file():
-            check_stale_path("agent smolmachine artifact", artifact)
-    sidecar_image = _bundle.SIDECAR_BUNDLE_IMAGE
-    if docker_mod.image_exists(sidecar_image):
-        digest = docker_mod.image_id(sidecar_image).split(":", 1)[-1][:16]
-        sidecar_artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-        if sidecar_artifact.is_file():
-            check_stale_path("sidecar smolmachine artifact", sidecar_artifact)
-
-
-def _image_policy(plan: object) -> str:
-    spec = getattr(plan, "spec", None)
-    return str(getattr(spec, "image_policy", "fresh"))
-
-
-def _cached_smolmachine(image_ref: str, *, label: str) -> Path:
-    """Return the cached smolmachine artifact for the current local image.
-
-    This is intentionally buildless: it inspects the existing local Docker
-    image ID and looks for the artifact keyed by that ID. If either side is
-    missing, the caller must use the fresh path to build/pack it.
-    """
-    if not docker_mod.image_exists(image_ref):
-        die(
-            f"cached {label} image {image_ref!r} not found; "
-            "run without --cached-images to build it"
-        )
-    digest = docker_mod.image_id(image_ref).split(":", 1)[-1][:16]
-    sidecar = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-    if not sidecar.is_file():
-        die(
-            f"cached {label} smolmachine artifact for {image_ref!r} not found; "
-            "run without --cached-images to build it"
-        )
-    info(f"using cached {label} smolmachine {str(sidecar)!r}")
-    return sidecar
-
-
 def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:
     """Build the agent docker image and convert it into a
     `.smolmachine` artifact, caching the result under

bot_bottle/backend/smolmachines/loopback_alias.py

@@ -152,23 +152,31 @@ def force_allowlist(machine_name: str, allowed_cidrs: list[str]) -> None:
     """Ensure the machine's persisted TSI allowlist equals
     `allowed_cidrs`, failing **closed** if that can't be confirmed.
 
-    Runs on both macOS and Linux. It exists because smolvm 0.8.0
-    silently drops `--allow-cidr` when combined with `--from`, so
-    the allowlist has to be written into smolvm's persistent state
-    DB before `machine start`. Rather than assume the flag was
-    dropped, we read the persisted row and only patch when it
-    doesn't already match — so a newer smolvm that honors the flag
-    is left untouched.
+    macOS only. On Linux, smolvm's TSI defaults to full loopback
+    access and patching `allowed_cidrs` into the state DB crashes
+    TSI boot (the VM fails with "boot process exited (code 1)").
+    Per-bottle CIDR isolation is therefore not enforced on Linux
+    until smolvm fixes the `--allow-cidr` + TSI interaction. This
+    is a known limitation — the agent VM can reach all of host
+    loopback, not just its bottle's forwarder ports.
+
+    On macOS, smolvm 0.8.0 silently drops `--allow-cidr` when
+    combined with `--from`, so the allowlist has to be written
+    into smolvm's persistent state DB before `machine start`.
+    Rather than assume the flag was dropped, we read the persisted
+    row and only patch when it doesn't already match — so a newer
+    smolvm that honors the flag is left untouched.
 
     Must run AFTER `smolvm machine create` (the row has to exist)
     and BEFORE `smolvm machine start` (smolvm reads the row on
     start; in-flight VMs don't pick up changes).
 
-    Fail-closed: if the state DB is missing, the row is missing, or
-    the allowlist still doesn't match after patching, we `die()`
-    rather than boot a VM whose egress confinement we can't verify
-    — an unconfirmed allowlist is a sandbox-escape risk (the agent
-    VM could reach all of host loopback)."""
+    Fail-closed (macOS): if the state DB is missing, the row is
+    missing, or the allowlist still doesn't match after patching,
+    we `die()` rather than boot a VM whose egress confinement we
+    can't verify."""
+    if not _is_macos():
+        return
     want = list(allowed_cidrs)
     if not _SMOLVM_DB_PATH.is_file():
         die(

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -135,6 +135,16 @@ def start_bundle_vm(
         elif entry in effective_host_env:
             env[entry] = effective_host_env[entry]
     name = bundle_machine_name(spec.slug)
+    # Destroy any leftover machine from a previous run that didn't
+    # clean up (e.g. crash, interrupted teardown).
+    try:
+        _smolvm.machine_stop(name)
+    except _smolvm.SmolvmError:
+        pass
+    try:
+        _smolvm.machine_delete(name)
+    except _smolvm.SmolvmError:
+        pass
     _smolvm.machine_create(
         name,
         from_path=from_path,

bot_bottle/cli/start.py

@@ -36,7 +36,6 @@ 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
@@ -64,14 +63,6 @@ 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",
@@ -104,8 +95,6 @@ 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"
 
@@ -153,10 +142,6 @@ 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,
@@ -165,7 +150,6 @@ def cmd_start(argv: list[str]) -> int:
         label=label,
         color=color,
         bottle_names=bottle_names,
-        image_policy=image_policy,
     )
     return _launch_bottle(
         spec,
@@ -226,7 +210,6 @@ def _start_headless(
         color=args.color or "",
         bottle_names=bottle_names,
         headless=True,
-        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -407,13 +390,6 @@ 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)
@@ -556,38 +532,30 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        skip_stale = False
-        while True:
-            try:
-                with backend.launch(plan, skip_stale=skip_stale) as bottle:
-                    agent_provider_template = getattr(plan, "agent_provider_template", "claude")
-                    extra_args: tuple[str, ...] = ()
-                    if headless_prompt_text:
-                        extra_args = tuple(
-                            get_provider(agent_provider_template).headless_prompt(
-                                headless_prompt_text
-                            )
-                        )
-                    exit_code = attach_agent(
-                        bottle,
-                        agent_provider_template=agent_provider_template,
-                        startup_args=plan.agent_provision.startup_args + extra_args,
+        with backend.launch(plan) as bottle:
+            agent_provider_template = getattr(plan, "agent_provider_template", "claude")
+            extra_args: tuple[str, ...] = ()
+            if headless_prompt_text:
+                extra_args = tuple(
+                    get_provider(agent_provider_template).headless_prompt(
+                        headless_prompt_text
                     )
-                    info(
-                        f"session ended (exit {exit_code}); "
-                        f"container {bottle.name} will be removed"
-                    )
-                    if agent_provider_template == "claude":
-                        capture_claude_session_state(identity, exit_code)
-                break
-            except StaleImageError as exc:
-                if assume_yes:
-                    die(str(exc))
-                sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
-                sys.stderr.flush()
-                if read_tty_line() not in ("y", "Y", "yes", "YES"):
-                    break
-                skip_stale = True
+                )
+            exit_code = attach_agent(
+                bottle,
+                agent_provider_template=agent_provider_template,
+                startup_args=plan.agent_provision.startup_args + extra_args,
+            )
+            info(
+                f"session ended (exit {exit_code}); "
+                f"container {bottle.name} will be removed"
+            )
+            # While the container is still alive: always snapshot the
+            # transcript and — if the agent exited non-zero — mark
+            # the state for preservation. This picks up crashes /
+            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
+            if agent_provider_template == "claude":
+                capture_claude_session_state(identity, exit_code)
         return 0
     finally:
         # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount

bot_bottle/config_store.py

@@ -1,71 +0,0 @@
-"""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/contrib/claude/Dockerfile

@@ -21,7 +21,7 @@ FROM node:22-slim
 # to it) works against egress's bumped TLS without the agent needing
 # local DNS.
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils \
+  && apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils iptables \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by claude-code itself

bot_bottle/contrib/codex/Dockerfile

@@ -6,7 +6,7 @@
 FROM node:22-slim
 
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep \
+  && apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep iptables \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by codex itself

bot_bottle/egress_entrypoint.sh

@@ -28,7 +28,7 @@ set -e
 # flag mitmdump would generate a fresh CA on the wrong path and
 # the agent's installed trust anchor would no longer match the
 # bumped leaf certs.
-CONFDIR=/home/mitmproxy/.mitmproxy
+CONFDIR="${EGRESS_CONFDIR:-/home/mitmproxy/.mitmproxy}"
 CONFDIR_FLAG="--set confdir=$CONFDIR"
 
 MODE="--mode regular@9099"

bot_bottle/image_cache.py

@@ -1,42 +0,0 @@
-"""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,11 +6,9 @@ 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
@@ -52,13 +50,11 @@ 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"]

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.

tests/unit/test_cli_start_headless.py

@@ -184,18 +184,6 @@ 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,12 +57,6 @@ 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)
@@ -72,7 +66,6 @@ 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()
 
     # ------------------------------------------------------------------
@@ -131,19 +124,6 @@ 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"])
@@ -235,7 +215,6 @@ 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

@@ -1,164 +0,0 @@
-"""Unit: _launch_bottle StaleImageError handling.
-
-Exercises the while-True / try-except loop around backend.launch:
-  - headless mode → die on stale
-  - interactive mode, user declines → stop without launching
-  - interactive mode, user confirms → retry with skip_stale=True
-"""
-
-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 _stale_cm() -> MagicMock:
-    """Return a context-manager mock whose __enter__ raises StaleImageError."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(side_effect=StaleImageError("image is 5 day(s) old"))
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
-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 must call die()."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
-
-        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)
-
-    def test_interactive_user_declines_stops_loop(self) -> None:
-        """Interactive user answering 'n' → loop exits without a second launch."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
-
-        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)
-        # launch was called once (stale), then the user declined → no second call.
-        backend_mock.launch.assert_called_once()
-
-    def test_interactive_user_confirms_retries_with_skip_stale(self) -> None:
-        """Interactive user answering 'y' → second launch with skip_stale=True."""
-        import bot_bottle.cli.start as start_mod
-
-        bottle_mock = MagicMock()
-        bottle_mock.name = "dev-abc"
-
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
-        backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
-
-        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)
-        # First call: skip_stale=False → stale. Second call: skip_stale=True → ok.
-        self.assertEqual(2, backend_mock.launch.call_count)
-        calls = backend_mock.launch.call_args_list
-        self.assertFalse(calls[0].kwargs.get("skip_stale", False))
-        self.assertTrue(calls[1].kwargs.get("skip_stale", False))
-
-    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"
-
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
-        backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
-
-        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)
-        self.assertEqual(2, backend_mock.launch.call_count)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_compose.py

@@ -118,7 +118,6 @@ 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."""
@@ -149,7 +148,6 @@ def _plan(
         agent_name="demo",
         copy_cwd=False,
         user_cwd="/tmp/x",
-        image_policy=image_policy,
     )
     return DockerBottlePlan(
         spec=spec,
@@ -267,8 +265,7 @@ 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.
-        cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
-        for kwargs in cases:
+        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
@@ -303,11 +300,6 @@ 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

@@ -1,86 +0,0 @@
-"""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,7 +3,6 @@
 from __future__ import annotations
 
 import contextlib
-import dataclasses
 import io
 import tempfile
 import unittest
@@ -17,7 +16,6 @@ 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
 
 
@@ -182,24 +180,6 @@ 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_util_image.py

@@ -9,7 +9,6 @@ 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
@@ -55,51 +54,6 @@ 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_image_cache.py

@@ -1,81 +0,0 @@
-"""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,7 +2,6 @@
 
 from __future__ import annotations
 
-import dataclasses
 import unittest
 import tempfile
 from pathlib import Path
@@ -16,7 +15,6 @@ 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({
@@ -83,7 +81,7 @@ def _plan(
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(image_policy="fresh"),
+        spec=SimpleNamespace(),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
@@ -294,7 +292,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
 
 def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
     return MacosContainerBottlePlan(
-        spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
+        spec=cast(BottleSpec, SimpleNamespace()),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -367,82 +365,5 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         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_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_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"):
-            result = launch._build_images(plan)
-
-        build.assert_not_called()
-        self.assertEqual(plan.image, result.image)
-
-    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"):
-            result = launch._build_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", result.image)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_container_util.py

@@ -273,71 +273,5 @@ 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_smolmachines_egress_apply.py

@@ -18,7 +18,7 @@ class TestFetchCurrentRoutes(unittest.TestCase):
             self.assertEqual("routes", egress_apply.fetch_current_routes("dev-abc"))
         exec_.assert_called_once_with(
             "bot-bottle-sidecars-dev-abc",
-            ["cat", "/etc/egress/routes.yaml"],
+            ["cat", "/bot-bottle-data/egress/routes.yaml"],
         )
 
     def test_read_failure_raises_apply_error(self):

tests/unit/test_smolmachines_launch_image.py

@@ -187,54 +187,6 @@ class TestAgentFromPath(unittest.TestCase):
             dockerfile="/repo/Dockerfile",
         )
 
-    def test_cached_policy_uses_existing_artifact_without_build(self):
-        with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
-            digest = "abcdef0123456789"
-            artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
-            artifact.write_text("")
-            plan = SimpleNamespace(
-                slug="dev-abc12",
-                agent_image="bot-bottle-claude:latest",
-                spec=SimpleNamespace(image_policy="cached"),
-            )
-            with patch.object(
-                _launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
-            ), patch.object(
-                _launch_mod, "read_committed_image", return_value="",
-            ), patch.object(
-                _launch_mod.docker_mod, "image_exists", return_value=True,
-            ), patch.object(
-                _launch_mod.docker_mod, "image_id",
-                return_value=f"sha256:{digest}fffffffffffffffff",
-            ), patch.object(
-                _launch_mod, "_ensure_smolmachine",
-            ) as ensure:
-                result = _launch_mod._agent_from_path(cast(Any, plan))
-
-        self.assertEqual(artifact, result)
-        ensure.assert_not_called()
-
-    def test_cached_policy_dies_when_artifact_missing(self):
-        with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
-            plan = SimpleNamespace(
-                slug="dev-abc12",
-                agent_image="bot-bottle-claude:latest",
-                spec=SimpleNamespace(image_policy="cached"),
-            )
-            with patch.object(
-                _launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
-            ), patch.object(
-                _launch_mod, "read_committed_image", return_value="",
-            ), patch.object(
-                _launch_mod.docker_mod, "image_exists", return_value=True,
-            ), patch.object(
-                _launch_mod.docker_mod, "image_id",
-                return_value="sha256:abcdef0123456789fffffffffffffffff",
-            ):
-                from bot_bottle.log import Die
-                with self.assertRaises(Die):
-                    _launch_mod._agent_from_path(cast(Any, plan))
-
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_loopback_alias.py

@@ -313,9 +313,9 @@ class TestForceAllowlist(unittest.TestCase):
         self.assertEqual(4, cfg["cpus"])
         self.assertTrue(cfg["network"])
 
-    def test_patches_on_linux_too(self):
-        # force_allowlist no longer no-ops on Linux — the TSI
-        # allowlist must be enforced there as well.
+    def test_skips_patch_on_linux(self):
+        # On Linux, patching allowed_cidrs into the smolvm state DB
+        # crashes TSI boot. force_allowlist is a no-op on Linux.
         with patch.object(loopback_alias, "_is_macos", return_value=False), \
              patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db):
             loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"])
@@ -324,7 +324,7 @@ class TestForceAllowlist(unittest.TestCase):
             "SELECT data FROM vms WHERE name='demo-vm'",
         ).fetchone()[0])
         con.close()
-        self.assertEqual(["127.0.0.16/32"], cfg["allowed_cidrs"])
+        self.assertIsNone(cfg["allowed_cidrs"])
 
     def test_skips_write_when_already_matching(self):
         # A newer smolvm that honors --allow-cidr at create leaves the

tests/unit/test_smolmachines_provision.py

@@ -404,7 +404,11 @@ class TestBundleLaunchSpec(unittest.TestCase):
             ),
         )
 
-        spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._stage_sidecar_data",
+            return_value=Path("/tmp/smolvm-sidecar-data"),
+        ):
+            spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
 
         self.assertEqual(
             "egress,git-gate,git-http",
@@ -416,7 +420,11 @@ class TestBundleLaunchSpec(unittest.TestCase):
     def test_canary_env_registered_as_sensitive_in_bundle(self):
         plan = _plan(canary=True)
 
-        spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._stage_sidecar_data",
+            return_value=Path("/tmp/smolvm-sidecar-data"),
+        ):
+            spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
 
         self.assertIn("CANON_ALPHA_SECRET=fake-canary-value", spec.environment)
         self.assertIn(
@@ -427,10 +435,16 @@ class TestBundleLaunchSpec(unittest.TestCase):
     def test_supervise_adds_daemon_volume_and_env(self):
         from bot_bottle.supervise import DB_PATH_IN_CONTAINER
         plan = _plan(supervise=True)
-        spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._stage_sidecar_data",
+            return_value=Path("/tmp/smolvm-sidecar-data"),
+        ):
+            spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
         self.assertIn("supervise", spec.daemons_csv)
         self.assertIn(f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", spec.environment)
-        self.assertIn(("/tmp/bot-bottle.db", DB_PATH_IN_CONTAINER, False), spec.volumes)
+        # virtiofs requires directory mounts; the DB's parent dir is
+        # mounted so bot-bottle.db lands at the right in-VM path.
+        self.assertIn(("/tmp", str(Path(DB_PATH_IN_CONTAINER).parent), False), spec.volumes)
 
     def test_canary_env_visible_to_smolvm_guest(self):
         plan = _plan(canary=True)
@@ -556,6 +570,9 @@ class TestLaunchResourceWiring(unittest.TestCase):
             "bot_bottle.backend.smolmachines.launch._bundle.start_bundle_vm",
             return_value=raw_launch,
         ) as start_vm, patch(
+            "bot_bottle.backend.smolmachines.launch._stage_sidecar_data",
+            return_value=Path("/tmp/smolvm-sidecar-data"),
+        ), patch(
             "bot_bottle.backend.smolmachines.launch._forward.start_forwarder",
             return_value=handle,
         ) as start_forwarder:
@@ -584,8 +601,11 @@ class TestLaunchResourceWiring(unittest.TestCase):
             "bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
         ) as machine_exec, patch(
             "bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
-        ) as wait_exec_ready:
-            _launch._init_vm(plan)
+        ) as wait_exec_ready, patch(
+            "bot_bottle.backend.smolmachines.launch._loopback._is_macos",
+            return_value=True,
+        ):
+            _launch._init_vm(plan, "127.0.0.16")
 
         machine_exec.assert_called_once()
         argv = machine_exec.call_args.args[1]
@@ -593,6 +613,27 @@ class TestLaunchResourceWiring(unittest.TestCase):
         self.assertIn("chown -R node:node /home/node", argv[2])
         wait_exec_ready.assert_called_once_with(plan.machine_name)
 
+    def test_init_vm_installs_iptables_on_linux(self):
+        plan = _plan()
+        from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
+            return_value=SmolvmRunResult(0, "", ""),
+        ) as machine_exec, patch(
+            "bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
+        ), patch(
+            "bot_bottle.backend.smolmachines.launch._loopback._is_macos",
+            return_value=False,
+        ):
+            _launch._init_vm(plan, "127.0.0.16")
+
+        # Two calls: ownership repair + iptables
+        self.assertEqual(2, machine_exec.call_count)
+        iptables_argv = machine_exec.call_args_list[1].args[1]
+        self.assertIn("iptables", iptables_argv[2])
+        self.assertIn("127.0.0.16/32", iptables_argv[2])
+        self.assertIn("127.0.0.0/8", iptables_argv[2])
+
 
 class TestPortLabels(unittest.TestCase):
     def test_known_and_dynamic_port_labels_round_trip(self):

tests/unit/test_stale_checks.py

@@ -1,302 +0,0 @@
-"""Unit: stale-image check functions across backends, and the
-BottleBackend.launch template method (skip_stale flag).
-
-No real images or containers are used — all Docker/container/smolmachine
-calls are mocked at the module boundary."""
-
-from __future__ import annotations
-
-import tempfile
-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 — _image_stale_checks + skip_stale
-# ---------------------------------------------------------------------------
-
-class TestBottleBackendLaunchTemplate(unittest.TestCase):
-    """Verify the concrete launch() method on BottleBackend calls
-    _image_stale_checks unless skip_stale=True, then delegates to _launch_impl."""
-
-    def _make_backend(self) -> Any:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        return DockerBottleBackend()
-
-    def test_stale_checks_called_by_default(self) -> None:
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ):
-            with backend.launch(plan):
-                pass
-        stale_mock.assert_called_once_with(plan)
-
-    def test_skip_stale_bypasses_stale_checks(self) -> None:
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ):
-            with backend.launch(plan, skip_stale=True):
-                pass
-        stale_mock.assert_not_called()
-
-    def test_noop_default_image_stale_checks(self) -> None:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        from bot_bottle.backend import BottleBackend
-        backend = DockerBottleBackend()
-        # Call the base-class _image_stale_checks directly to verify it's a no-op.
-        BottleBackend._image_stale_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_image_stale_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._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# smolmachines backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestSmolmachinesStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached") -> Any:
-        return cast(Any, SimpleNamespace(
-            spec=SimpleNamespace(image_policy=policy),
-            slug="dev-abc",
-            agent_image="bot-bottle-claude:latest",
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale_path") as csp:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        csp.assert_not_called()
-
-    def test_committed_artifact_found_checks_it(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f:
-            committed_path = f.name
-            with patch.object(mod, "read_committed_image", return_value=committed_path), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=False), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        csp.assert_called_once()
-        self.assertIn("agent smolmachine", csp.call_args.args[0])
-
-    def test_no_committed_agent_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        digest = "abcdef0123456789"
-        with tempfile.TemporaryDirectory() as tmp:
-            artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
-            artifact.write_text("")
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 patch.object(mod, "read_committed_image", return_value=""), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=True), \
-                 patch.object(
-                     mod.docker_mod, "image_id",
-                     return_value=f"sha256:{digest}ffffffffffffffff",
-                 ), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        # At least the agent artifact check ran.
-        self.assertGreaterEqual(csp.call_count, 1)
-
-    def test_sidecar_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        sidecar_digest = "fedcba9876543210"
-        with tempfile.TemporaryDirectory() as tmp:
-            sidecar_artifact = (
-                Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine"
-            )
-            sidecar_artifact.write_text("")
-
-            def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
-                return True
-
-            def image_id(ref: str) -> str:
-                if ref == mod._bundle.SIDECAR_BUNDLE_IMAGE:  # type: ignore[attr-defined]
-                    return f"sha256:{sidecar_digest}ffffffff"
-                return "sha256:0000000000000000ffffffff"
-
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 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_id", side_effect=image_id), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        sidecar_calls = [c for c in csp.call_args_list if "sidecar" in c.args[0]]
-        self.assertGreaterEqual(len(sidecar_calls), 1)
-
-    def test_backend_image_stale_checks_delegates(self) -> None:
-        from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend
-        from bot_bottle.backend.smolmachines import launch as mod
-        backend = SmolmachinesBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_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_image_stale_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._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-if __name__ == "__main__":
-    unittest.main()