fix(smolmachines): consolidate virtiofs mounts to stay within libkrun limit

libkrun limits total mounts + port-mappings to 5. With all three
sidecar daemons active (egress, git-gate, supervise), the prior
layout used 3 mounts + 3 ports = 6, causing the VM to crash at boot
with krun_start_enter returned -22.

Merge the egress confdir and git-gate scripts into a single staging
directory mounted at /bot-bottle-data/ with egress/ and git-gate/
subdirectories, reducing to 2 mounts + 3 ports = 5. Also clean up
leftover VMs before creating new ones to handle interrupted teardowns.

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
+#   /etc/git-gate/entrypoint.sh          per-bottle (docker-cp or virtiofs mount)
-#   /git-gate-entrypoint.sh              docker-cp'd at start time
+#   /etc/git-gate/pre-receive            per-bottle (docker-cp or virtiofs mount)
-#   /git-gate/creds/*                    docker-cp'd at start time
+#   /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)
@@ -276,18 +273,6 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
 CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan)
 
 
-@dataclass(frozen=True)
-class BottleImages:
-    """Resolved image references (or artifact paths) for a bottle launch.
-
-    For Docker/macOS-container backends, `agent` and `sidecar` are string
-    image refs. For the smolmachines backend they are Path objects pointing
-    to pre-built `.smolmachine` artifacts."""
-
-    agent: str | Path
-    sidecar: str | Path
-
-
 class BottleBackend(ABC, Generic[PlanT, CleanupT]):
     """Abstract base for selectable bottle backends. Concrete subclasses
     (e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -447,27 +432,9 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
-    def prelaunch_checks(self, plan: PlanT) -> None:
-        """Raise StaleImageError if any cached image used by this plan is stale.
-        No-op default; backends override to call the shared check_stale*
-        helpers on their image/artifact timestamps. Called by the CLI before
-        launch so the operator can be prompted outside the launch context."""
-
-    @contextmanager
-    def launch(self, plan: PlanT) -> Generator[Bottle, None, None]:
-        """Template: build or load images, then delegate to _launch_impl."""
-        images = self._build_or_load_images(plan)
-        with self._launch_impl(plan, images) as bottle:
-            yield bottle
-
     @abstractmethod
-    def _build_or_load_images(self, plan: PlanT) -> BottleImages:
+    def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
-        """Return the agent and sidecar image references (or artifact paths)
+        """Build/run the bottle and yield a handle; tear down on exit."""
-        for this plan, building fresh images when the policy requires it."""
-
-    @abstractmethod
-    def _launch_impl(self, plan: PlanT, images: BottleImages) -> AbstractContextManager[Bottle]:
-        """Bring up the bottle using pre-resolved images; yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy host-side files (CA cert, prompt, skills, .git) into

bot_bottle/backend/docker/backend.py

@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -85,15 +85,9 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
-    def prelaunch_checks(self, plan: DockerBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
-    def _build_or_load_images(self, plan: DockerBottlePlan) -> BottleImages:
-        return _launch.build_or_load_images(plan)
-
     @contextmanager
-    def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
+    def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
-        with _launch.launch(plan, images, provision=self.provision) as bottle:
+        with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 
     def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:

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

@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -67,17 +67,11 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
-    def prelaunch_checks(self, plan: MacosContainerBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
-    def _build_or_load_images(self, plan: MacosContainerBottlePlan) -> BottleImages:
-        return _launch.build_or_load_images(plan)
-
     @contextmanager
-    def _launch_impl(
+    def launch(
-        self, plan: MacosContainerBottlePlan, images: BottleImages
+        self, plan: MacosContainerBottlePlan
     ) -> Generator[MacosContainerBottle, None, None]:
-        with _launch.launch(plan, images, provision=self.provision) as bottle:
+        with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 
     def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:

bot_bottle/backend/macos_container/launch.py

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

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

@@ -19,7 +19,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -77,17 +77,11 @@ class SmolmachinesBottleBackend(
             stage_dir=stage_dir,
         )
 
-    def prelaunch_checks(self, plan: SmolmachinesBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
-    def _build_or_load_images(self, plan: SmolmachinesBottlePlan) -> BottleImages:
-        return _launch.build_or_load_images(plan)
-
     @contextmanager
-    def _launch_impl(
+    def launch(
-        self, plan: SmolmachinesBottlePlan, images: BottleImages
+        self, plan: SmolmachinesBottlePlan
     ) -> Generator[SmolmachinesBottle, None, None]:
-        with _launch.launch(plan, images, provision=self.provision) as bottle:
+        with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 
     def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str:

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,28 +28,19 @@ 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 info, warn
-from ...log import die, info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
     read_committed_image,
 )
-from .. import BottleImages
 from . import loopback_alias as _loopback
 from . import port_forward as _forward
 from . import sidecar_bundle as _bundle
@@ -63,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
@@ -77,32 +80,13 @@ _GIT_HTTP_PORT = 9420
 _SUPERVISE_PORT = SUPERVISE_PORT
 
 
-def build_or_load_images(plan: SmolmachinesBottlePlan) -> BottleImages:
-    """Return pre-built or freshly built smolmachine artifact paths."""
-    return BottleImages(
-        agent=_agent_from_path(plan),
-        sidecar=_sidecar_from_path(plan),
-    )
-
-
-def _sidecar_from_path(plan: SmolmachinesBottlePlan) -> Path:
-    """Return the sidecar bundle artifact path, building it if needed."""
-    if _image_policy(plan) == "cached":
-        return _cached_smolmachine(_bundle.SIDECAR_BUNDLE_IMAGE, label="sidecar")
-    return _ensure_smolmachine(
-        _bundle.SIDECAR_BUNDLE_IMAGE,
-        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-    )
-
-
 @contextmanager
 def launch(
     plan: SmolmachinesBottlePlan,
-    images: BottleImages,
     *,
     provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
 ) -> Generator[SmolmachinesBottle, None, None]:
-    """Run the bottle from pre-built images and yield a handle; tear everything
+    """Build + run the bottle and yield a handle; tear everything
     down on exit. Errors during bringup unwind any partial state
     via the ExitStack."""
     stack = ExitStack()
@@ -110,9 +94,11 @@ def launch(
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
         proxy_host = loopback_ip
-        plan = _start_bundle(plan, network, proxy_host, Path(images.sidecar), stack)
+        plan = _start_bundle(plan, network, proxy_host, stack)
 
-        _launch_vm(plan, Path(images.agent), proxy_host, stack)
+        agent_from_path = _agent_from_path(plan)
+
+        _launch_vm(plan, agent_from_path, proxy_host, stack)
         _init_vm(plan)
 
         bottle = SmolmachinesBottle(
@@ -189,18 +175,21 @@ def _start_bundle(
     plan: SmolmachinesBottlePlan,
     network: str,
     proxy_host: str,
-    sidecar_artifact: Path,
     stack: ExitStack,
 ) -> SmolmachinesBottlePlan:
-    """Build the BundleLaunchSpec, start the sidecar VM from the pre-resolved
+    """Build the BundleLaunchSpec, start the sidecar VM, wrap its raw
-    artifact, wrap its raw smolVM-published loopback ports with per-bottle
+    smolVM-published loopback ports with per-bottle forwarders, stamp
-    forwarders, stamp agent URLs from those forwarder ports, and register teardown."""
+    agent URLs from those forwarder ports, and register teardown."""
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
+    artifact = _ensure_smolmachine(
+        bundle_spec.image,
+        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
+    )
     launch = _bundle.start_bundle_vm(
         bundle_spec,
-        from_path=sidecar_artifact,
+        from_path=artifact,
         host_env={**os.environ, **token_env},
     )
     stack.callback(_bundle.stop_bundle_vm, plan.slug)
@@ -307,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,
@@ -374,6 +373,64 @@ 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 credential paths: the rendered entrypoint hardcodes
+        # /git-gate/creds/; rewrite to the in-VM git-gate subdir.
+        ep_path = scripts_dir / "entrypoint.sh"
+        ep_path.write_text(
+            ep_path.read_text().replace(
+                "/git-gate/creds/",
+                f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/creds/",
+            )
+        )
+
+        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:
@@ -392,35 +449,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))
+    gp = plan.git_gate_plan
-    if ep.routes:
+    staging = _stage_sidecar_data(plan)
-        volumes.append((str(ep.routes_path.parent), str(Path(EGRESS_ROUTES_IN_CONTAINER).parent), True))
+    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
@@ -431,7 +479,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
@@ -481,9 +535,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
@@ -494,60 +545,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/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,15 +532,6 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        try:
-            backend.prelaunch_checks(plan)
-        except StaleImageError as exc:
-            if assume_yes:
-                die(str(exc))
-            sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
-            sys.stderr.flush()
-            if read_tty_line() not in ("y", "Y", "yes", "YES"):
-                return 0
         with backend.launch(plan) as bottle:
             agent_provider_template = getattr(plan, "agent_provider_template", "claude")
             extra_args: tuple[str, ...] = ()
@@ -583,6 +550,10 @@ def _launch_bottle(
                 f"session ended (exit {exit_code}); "
                 f"container {bottle.name} will be removed"
             )
+            # While the container is still alive: always snapshot the
+            # transcript and — if the agent exited non-zero — mark
+            # the state for preservation. This picks up crashes /
+            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
             if agent_provider_template == "claude":
                 capture_claude_session_state(identity, exit_code)
         return 0

bot_bottle/config_store.py

@@ -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/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
+The "AI coding agents in isolated sandboxes" space is active but not saturated.
-occupies a distinct position: no surveyed project combines all five of its
+bot-bottle occupies a distinct position: no surveyed project combines all five
-defining features. Publishing is likely worthwhile, with the main risk being
+of its defining features. Publishing is likely worthwhile, with the main risk
-claudebox expanding to absorb the same niche.
+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)
+1. Named-agent manifest with per-agent env resolution (prompt / host-forward / literal), supporting multiple providers (Claude Code, Codex, Pi, arbitrary plugins)
-2. Claude Code skills directory injection
+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
+- The Python-stdlib-first, low-dependency design — competitors are npm-based,
-  Kubernetes-native.
+  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
+Main risk: claudebox adds manifest/agent config; SuperHQ is moving fast on the
-enough that publishing sooner is better if establishing prior art matters.
+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,146 +0,0 @@
-"""Unit: _launch_bottle StaleImageError handling.
-
-Exercises prelaunch_checks / backend.launch flow:
-  - headless mode → die on stale
-  - interactive mode, user declines → stop without launching
-  - interactive mode, user confirms → skip stale check and launch once
-"""
-
-from __future__ import annotations
-
-import io
-import tempfile
-import unittest
-from types import SimpleNamespace
-from typing import Any, cast
-from unittest.mock import MagicMock, patch
-
-from bot_bottle.image_cache import StaleImageError
-from bot_bottle.log import Die
-
-
-def _fake_plan() -> Any:
-    provision = SimpleNamespace(startup_args=())
-    return cast(Any, SimpleNamespace(
-        agent_provision=provision,
-        agent_provider_template="claude",
-        slug="dev-abc",
-    ))
-
-
-def _ok_cm(bottle: Any) -> MagicMock:
-    """Return a context-manager mock that yields `bottle`."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(return_value=bottle)
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
-class TestLaunchBottleStaleHandling(unittest.TestCase):
-    def setUp(self) -> None:
-        self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.")
-
-    def _spec(self) -> Any:
-        from bot_bottle.backend import BottleSpec
-        from bot_bottle.manifest import ManifestIndex
-        idx = ManifestIndex.from_json_obj({
-            "bottles": {"dev": {}},
-            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
-        })
-        return BottleSpec(
-            manifest=idx,
-            agent_name="demo",
-            copy_cwd=False,
-            user_cwd=self._tmp,
-            identity="dev-abc",
-        )
-
-    def _run_launch(self, **patch_kwargs: Any) -> int:
-        import bot_bottle.cli.start as start_mod
-        spec = self._spec()
-        with patch.object(start_mod, "prepare_with_preflight",
-                          return_value=(_fake_plan(), "dev-abc")), \
-             patch.object(start_mod, "settle_state"), \
-             patch.object(start_mod, "info"):
-            return start_mod._launch_bottle(
-                spec,
-                dry_run=False,
-                backend_name="docker",
-                **patch_kwargs,
-            )
-
-    def test_headless_stale_calls_die(self) -> None:
-        """In headless mode (assume_yes=True), a StaleImageError from prelaunch_checks must call die()."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "die", side_effect=Die()):
-            with self.assertRaises(Die):
-                self._run_launch(assume_yes=True)
-
-        backend_mock.launch.assert_not_called()
-
-    def test_interactive_user_declines_stops_before_launch(self) -> None:
-        """Interactive user answering 'n' → launch is never called."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="n"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        backend_mock.launch.assert_not_called()
-
-    def test_interactive_user_confirms_launches_once(self) -> None:
-        """Interactive user answering 'y' → prelaunch stale error is bypassed; launch called once."""
-        import bot_bottle.cli.start as start_mod
-
-        bottle_mock = MagicMock()
-        bottle_mock.name = "dev-abc"
-
-        backend_mock = MagicMock()
-        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
-        backend_mock.launch.return_value = _ok_cm(bottle_mock)
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="y"), \
-             patch.object(start_mod, "attach_agent", return_value=0), \
-             patch.object(start_mod, "capture_claude_session_state"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        backend_mock.prelaunch_checks.assert_called_once()
-        backend_mock.launch.assert_called_once()
-
-    def test_interactive_yes_uppercase_also_accepted(self) -> None:
-        """'Y' or 'YES' should also be accepted as confirmation."""
-        import bot_bottle.cli.start as start_mod
-
-        bottle_mock = MagicMock()
-        bottle_mock.name = "dev-abc"
-
-        backend_mock = MagicMock()
-        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
-        backend_mock.launch.return_value = _ok_cm(bottle_mock)
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="YES"), \
-             patch.object(start_mod, "attach_agent", return_value=0), \
-             patch.object(start_mod, "capture_claude_session_state"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        backend_mock.launch.assert_called_once()
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_compose.py

@@ -118,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 [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
-        for kwargs in cases:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
@@ -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
 
 
@@ -90,7 +88,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
         committed_tag: str | None = None,
         image_present: bool = True,
     ) -> list[str]:
-        """Drive build_or_load_images() + launch() with the committed-image
+        """Drive launch() through its full sequence with the committed-image
         behaviour controlled by the arguments. Returns the images that were
         passed to `build_image` (empty list if it was never called)."""
         built: list[str] = []
@@ -124,9 +122,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
-            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, images, provision=provision):
+            with launch_mod.launch(plan, provision=provision):
                 pass
 
         return built
@@ -171,9 +168,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
            mock.patch.object(launch_mod, "compose_dump_logs"), \
            mock.patch.object(launch_mod, "compose_down"), \
            contextlib.redirect_stderr(io.StringIO()):
-            images = launch_mod.build_or_load_images(plan)
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, images, provision=provision):
+            with launch_mod.launch(plan, provision=provision):
                 pass
 
         self.assertEqual(1, len(captured_plans))
@@ -184,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_launch_teardown.py

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

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=())),
@@ -339,9 +337,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ), patch.object(launch, "info"):
-            images = launch.build_or_load_images(plan)
+            updated = launch._build_images(plan)
 
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
+        self.assertEqual("bot-bottle-committed-dev-abc:latest", updated.image)
         self.assertEqual(1, len(calls))
         self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0])
 
@@ -360,89 +358,12 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
         ):
-            images = launch.build_or_load_images(plan)
+            updated = launch._build_images(plan)
 
-        self.assertEqual("bot-bottle-agent:latest", images.agent)
+        self.assertEqual("bot-bottle-agent:latest", updated.image)
         self.assertEqual(2, len(calls))
         self.assertEqual("bot-bottle-agent:latest", calls[1][0])
 
 
-class TestMacosContainerLaunchCachedImages(unittest.TestCase):
-    def setUp(self):
-        self._tmp = tempfile.TemporaryDirectory()
-        self.stage_dir = Path(self._tmp.name)
-
-    def tearDown(self):
-        self._tmp.cleanup()
-
-    def _cached_plan(self) -> MacosContainerBottlePlan:
-        return dataclasses.replace(
-            _build_plan(self.stage_dir),
-            spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
-        )
-
-    def test_cached_mode_dies_when_agent_image_missing(self) -> None:
-        plan = self._cached_plan()
-
-        def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
-            return False  # Nothing cached.
-
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", side_effect=image_exists,
-        ), patch.object(
-            launch, "die", side_effect=Die(),
-        ):
-            with self.assertRaises(Die):
-                launch.build_or_load_images(plan)
-
-    def test_cached_mode_dies_when_sidecar_image_missing(self) -> None:
-        plan = self._cached_plan()
-
-        def image_exists(ref: str) -> bool:
-            return ref == plan.image  # Agent present; sidecar missing.
-
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", side_effect=image_exists,
-        ), patch.object(
-            launch, "die", side_effect=Die(),
-        ):
-            with self.assertRaises(Die):
-                launch.build_or_load_images(plan)
-
-    def test_cached_mode_both_present_returns_unchanged_plan(self) -> None:
-        plan = self._cached_plan()
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", return_value=True,
-        ), patch.object(
-            launch.container_mod, "build_image",
-        ) as build, patch.object(launch, "info"):
-            images = launch.build_or_load_images(plan)
-
-        build.assert_not_called()
-        self.assertEqual(plan.image, images.agent)
-
-    def test_committed_image_plus_cached_skips_sidecar_build(self) -> None:
-        plan = self._cached_plan()
-        with patch.object(
-            launch, "read_committed_image",
-            return_value="bot-bottle-committed-dev-abc:latest",
-        ), patch.object(
-            launch.container_mod, "image_exists", return_value=True,
-        ), patch.object(
-            launch.container_mod, "build_image",
-        ) as build, patch.object(launch, "info"):
-            images = launch.build_or_load_images(plan)
-
-        # In cached mode with a committed image, no builds should fire.
-        build.assert_not_called()
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_container_util.py

@@ -273,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_provision.py

@@ -404,6 +404,10 @@ class TestBundleLaunchSpec(unittest.TestCase):
             ),
         )
 
+        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(
@@ -416,6 +420,10 @@ class TestBundleLaunchSpec(unittest.TestCase):
     def test_canary_env_registered_as_sensitive_in_bundle(self):
         plan = _plan(canary=True)
 
+        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)
@@ -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)
+        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)
@@ -546,22 +560,27 @@ class TestLaunchResourceWiring(unittest.TestCase):
             ),
         )
 
-        sidecar_artifact = Path("/cache/sidecar.smolmachine")
-
         with patch(
             "bot_bottle.backend.smolmachines.launch._provision_git_gate_keys",
             return_value=plan,
         ), patch(
+            "bot_bottle.backend.smolmachines.launch._ensure_smolmachine",
+            return_value=Path("/cache/sidecar.smolmachine"),
+        ) as ensure, patch(
             "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:
-            stamped = _launch._start_bundle(plan, "net", "127.0.0.16", sidecar_artifact, stack)
+            stamped = _launch._start_bundle(plan, "net", "127.0.0.16", stack)
 
+        ensure.assert_called_once()
         start_vm.assert_called_once()
-        self.assertEqual(sidecar_artifact, start_vm.call_args.kwargs["from_path"])
+        self.assertEqual(Path("/cache/sidecar.smolmachine"), start_vm.call_args.kwargs["from_path"])
         specs = start_forwarder.call_args.args[0]
         self.assertEqual(
             (

tests/unit/test_stale_checks.py

@@ -1,292 +0,0 @@
-"""Unit: stale-image check functions across backends, and the
-BottleBackend.launch template method (prelaunch_checks + build_or_load_images).
-
-No real images or containers are used — all Docker/container/smolmachine
-calls are mocked at the module boundary."""
-
-from __future__ import annotations
-
-import 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 — prelaunch_checks + _build_or_load_images
-# ---------------------------------------------------------------------------
-
-class TestBottleBackendLaunchTemplate(unittest.TestCase):
-    """Verify the concrete launch() method on BottleBackend calls
-    _build_or_load_images and _launch_impl, and that prelaunch_checks is a no-op
-    on the base class."""
-
-    def _make_backend(self) -> Any:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        return DockerBottleBackend()
-
-    def test_launch_delegates_to_build_or_load_and_launch_impl(self) -> None:
-        from bot_bottle.backend import BottleImages
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        images = BottleImages(agent="agent:latest", sidecar="sidecar:latest")
-        with patch.object(
-            backend, "_build_or_load_images", return_value=images,
-        ) as build_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ) as impl_mock:
-            with backend.launch(plan):
-                pass
-        build_mock.assert_called_once_with(plan)
-        impl_mock.assert_called_once_with(plan, images)
-
-    def test_noop_default_prelaunch_checks(self) -> None:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        from bot_bottle.backend import BottleBackend
-        backend = DockerBottleBackend()
-        # Base-class prelaunch_checks is a no-op — must not raise.
-        BottleBackend.prelaunch_checks(backend, cast(Any, SimpleNamespace()))  # type: ignore[arg-type]
-
-
-# ---------------------------------------------------------------------------
-# Docker backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestDockerStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
-        spec = SimpleNamespace(image_policy=policy)
-        provision = SimpleNamespace(image="bot-bottle-agent:latest")
-        return cast(Any, SimpleNamespace(
-            spec=spec,
-            slug=slug,
-            image="bot-bottle-agent:latest",
-            agent_provision=provision,
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        cs.assert_not_called()
-
-    def test_committed_image_present_checks_only_committed(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
-             patch.object(mod.docker_mod, "image_exists", return_value=True), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_called_once()
-        self.assertIn("committed:latest", cs.call_args.args[0])
-
-    def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", return_value=True), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        # Should have been called for both agent and sidecar images.
-        self.assertEqual(2, cs.call_count)
-
-    def test_image_not_present_skips_check(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", return_value=False), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_not_called()
-
-    def test_only_sidecar_missing_checks_only_agent(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        plan = self._plan()
-
-        def image_exists(ref: str) -> bool:
-            return ref == plan.image  # Only agent present; sidecar missing.
-
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(plan)
-        self.assertEqual(1, cs.call_count)
-        self.assertIn(plan.image, cs.call_args.args[0])
-
-    def test_backend_prelaunch_checks_delegates(self) -> None:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        from bot_bottle.backend.docker import launch as mod
-        backend = DockerBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend.prelaunch_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# 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_prelaunch_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.prelaunch_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# macOS container backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestMacosContainerStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
-        return cast(Any, SimpleNamespace(
-            spec=SimpleNamespace(image_policy=policy),
-            slug=slug,
-            image="bot-bottle-agent:latest",
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        cs.assert_not_called()
-
-    def test_committed_image_present_checks_only_committed(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
-             patch.object(mod.container_mod, "image_exists", return_value=True), \
-             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_called_once()
-        self.assertIn("committed:latest", cs.call_args.args[0])
-
-    def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.container_mod, "image_exists", return_value=True), \
-             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        self.assertEqual(2, cs.call_count)
-
-    def test_image_not_present_skips_check(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.container_mod, "image_exists", return_value=False), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_not_called()
-
-    def test_backend_prelaunch_checks_delegates(self) -> None:
-        from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
-        from bot_bottle.backend.macos_container import launch as mod
-        backend = MacosContainerBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend.prelaunch_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-if __name__ == "__main__":
-    unittest.main()