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

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

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

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

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

Dockerfile.sidecars

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

bot_bottle/backend/__init__.py

@@ -75,9 +75,9 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
-    # Image startup policy. "fresh" preserves the normal build path;
-    # "cached" reuses the current local image/artifact without rebuilding.
-    image_policy: str = "fresh"
+    # 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
 
 
 @dataclass(frozen=True)
@@ -303,6 +303,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
 
         self._preflight()
 
+        from ..git_gate_host_key import preflight_host_keys
+        manifest = preflight_host_keys(
+            manifest,
+            headless=spec.headless,
+            home_md=spec.manifest.home_md,
+        )
+
         manifest_bottle = manifest.bottle
         manifest_agent_provider = manifest_bottle.agent_provider
         agent_provider = get_provider(manifest_agent_provider.template)

bot_bottle/backend/docker/compose.py

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

bot_bottle/backend/docker/launch.py

@@ -41,8 +41,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import warn_if_stale
-from ...log import die, info, warn
+from ...log import info, warn
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -64,7 +63,6 @@ from .compose import (
     write_compose_file,
 )
 from .egress import egress_tls_init
-from .sidecar_bundle import SIDECAR_BUNDLE_IMAGE
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
@@ -102,39 +100,12 @@ def launch(
         # Dockerfile. Sidecar images get built lazily by `docker compose
         # up` via the renderer's `build:` directives.
         committed = read_committed_image(plan.slug)
-        cached_policy = plan.spec.image_policy == "cached"
         if committed and docker_mod.image_exists(committed):
             info(f"using committed image {committed!r}")
-            if cached_policy:
-                warn_if_stale(
-                    f"agent image {committed!r}",
-                    docker_mod.image_created_at(committed),
-                )
             plan = dataclasses.replace(
                 plan,
                 agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
             )
-        elif cached_policy:
-            if not docker_mod.image_exists(plan.image):
-                die(
-                    f"cached agent image {plan.image!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            if not docker_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
-                die(
-                    f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            info(f"using cached agent image {plan.image!r}")
-            info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
-            warn_if_stale(
-                f"agent image {plan.image!r}",
-                docker_mod.image_created_at(plan.image),
-            )
-            warn_if_stale(
-                f"sidecar image {SIDECAR_BUNDLE_IMAGE!r}",
-                docker_mod.image_created_at(SIDECAR_BUNDLE_IMAGE),
-            )
         else:
             docker_mod.build_image(
                 plan.image, _REPO_DIR,

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/smolmachines/bottle_plan.py

@@ -1,10 +1,8 @@
 """SmolmachinesBottlePlan — concrete BottlePlan for the smolmachines
 backend (PRD 0023).
 
-Slug + bundle docker subnet / gateway / pinned IP + smolvm
-machine name + agent `.smolmachine` artifact + per-bottle guest
-env. Provisioning fields (CA cert path, prompt path, etc.) land
-in chunk 4."""
+Slug + legacy bundle network coordinates + smolvm machine name +
+agent `.smolmachine` artifact + per-bottle guest env."""
 
 from __future__ import annotations
 
@@ -23,9 +21,10 @@ class SmolmachinesBottlePlan(BottlePlan):
     `supervise_plan`, and `agent_provision` from BottlePlan."""
 
     slug: str
-    # Per-bottle docker subnet for the sidecar bundle container.
-    # The bundle runs at `bundle_ip` (always `.2`); the gateway is
-    # at `.1`. smolvm's TSI allowlist is set to `bundle_ip/32`.
+    # Legacy per-bottle bundle network coordinates. These remain on
+    # the plan while BundleLaunchSpec still carries the original shape,
+    # but the smolmachines launch path exposes the sidecar VM through
+    # host-loopback forwarders instead of a Docker bridge IP.
     bundle_subnet: str
     bundle_gateway: str
     bundle_ip: str
@@ -36,22 +35,10 @@ class SmolmachinesBottlePlan(BottlePlan):
     # `--smolfile` is mutually exclusive with `--from`, and
     # `--from` is the path that avoids the registry-pull race).
     guest_env: dict[str, str]
-    # Inner Plans for the sidecar bundle daemons. The same shape the
-    # docker backend uses — same `.prepare()` calls produced
-    # them — but our launch step doesn't populate the
-    # docker-specific network fields (internal_network,
-    # egress_network) because the smolmachines bundle isn't on
-    # docker's `--internal` + egress bridge topology; it's on a
-    # per-bottle bridge with a pinned IP. The unused fields stay
-    # at their dataclass defaults.
-    # Agent-side endpoints. On Docker Desktop the docker bridge
-    # IPs aren't reachable from the smolvm guest (TSI uses macOS
-    # networking; docker container IPs live in the daemon's VM),
-    # so the agent dials the bundle via host loopback +
-    # docker-published random ports. Empty at prepare time;
-    # launch populates these after bundle bringup via
-    # `dataclasses.replace`. Format: a `host:port` for git-gate
-    # (insteadOf URL prefix) + full URLs for proxy / supervise.
+    # Agent-side endpoints. Empty at prepare time; launch populates
+    # these after sidecar VM bringup via `dataclasses.replace`.
+    # Format: a `host:port` for git-gate (insteadOf URL prefix) +
+    # full URLs for proxy / supervise.
     agent_proxy_url: str = ""
     agent_git_gate_host: str = ""
     agent_supervise_url: 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

@@ -1,12 +1,9 @@
-"""End-to-end launch flow for the smolmachines backend
-(PRD 0023 chunks 2d + 4b).
+"""End-to-end launch flow for the smolmachines backend.
 
-Brings up the per-bottle docker bridge + sidecar bundle (with
-real daemons + their config files), creates + starts the smolvm
-guest pointed at the bundle's pinned IP via TSI's
-`--allow-cidr <bundle-ip>/32` allowlist, yields a
-`SmolmachinesBottle` handle, tears everything down on context
-exit.
+Builds the sidecar bundle smolmachine, starts it as a sidecar VM
+with real daemons + their config files, creates + starts the agent
+smolVM, yields a `SmolmachinesBottle` handle, and tears everything
+down on context exit.
 
 The bundle's daemons consume the inner Plans the docker backend
 already produces: egress reads routes + CAs from the EgressPlan.
@@ -17,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,
@@ -31,22 +28,14 @@ from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker import util as docker_mod
 from ..docker.egress import (
-    EGRESS_CA_IN_CONTAINER,
     EGRESS_PORT as _EGRESS_PORT,
     egress_tls_init,
 )
-from ..docker.git_gate import (
-    GIT_GATE_ACCESS_HOOK_IN_CONTAINER,
-    GIT_GATE_CREDS_DIR_IN_CONTAINER,
-    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
-    GIT_GATE_HOOK_IN_CONTAINER,
-)
 from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import warn_if_stale_path
-from ...log import die, info, warn
+from ...log import info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
@@ -65,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
@@ -73,9 +74,8 @@ _SMOLMACHINE_CACHE_DIR = Path.home() / ".cache" / "bot-bottle" / "smolmachines"
 
 
 # Container-internal listening ports for each bundle daemon. The
-# bundle publishes each one on a random host loopback port (see
-# `_bundle.start_bundle`), and `_bundle.bundle_host_port` looks
-# them up post-start.
+# sidecar VM publishes each one on a random host loopback port, and
+# the launch flow wraps those raw ports with per-bottle forwarders.
 _GIT_HTTP_PORT = 9420
 _SUPERVISE_PORT = SUPERVISE_PORT
 
@@ -93,13 +93,13 @@ def launch(
     try:
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
-        proxy_host = _proxy_host(plan, loopback_ip)
+        proxy_host = loopback_ip
         plan = _start_bundle(plan, network, proxy_host, stack)
 
         agent_from_path = _agent_from_path(plan)
 
         _launch_vm(plan, agent_from_path, proxy_host, stack)
-        _init_vm(plan)
+        _init_vm(plan, proxy_host)
 
         bottle = SmolmachinesBottle(
             plan.machine_name,
@@ -183,13 +183,10 @@ def _start_bundle(
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
-    if _image_policy(plan) == "cached":
-        artifact = _cached_smolmachine(bundle_spec.image, label="sidecar")
-    else:
-        artifact = _ensure_smolmachine(
-            bundle_spec.image,
-            dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-        )
+    artifact = _ensure_smolmachine(
+        bundle_spec.image,
+        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
+    )
     launch = _bundle.start_bundle_vm(
         bundle_spec,
         from_path=artifact,
@@ -291,15 +288,24 @@ def _launch_vm(
 ) -> None:
     """Create, patch, and start the smolvm VM; register teardown.
 
-    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias
-    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
-    ensures the guest can only reach bundle ports published on that IP,
-    not the container IP directly. force_allowlist confirms the
-    allowlist persisted (patching smolvm 0.8.0's silent-drop of
-    --allow-cidr when combined with --from) and fails closed if it
-    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
-    and --smolfile mutually exclusive."""
+    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias.
+    This ensures the guest can only reach sidecar forwarders published
+    on that IP, not host localhost or another bottle's alias.
+    force_allowlist confirms the allowlist persisted (patching smolvm
+    0.8.0's silent-drop of --allow-cidr when combined with --from) and
+    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,
@@ -316,17 +322,24 @@ def _launch_vm(
     stack.callback(_smolvm.machine_stop, plan.machine_name)
 
 
-def _init_vm(plan: SmolmachinesBottlePlan) -> None:
-    """Repair filesystem ownership and wait for exec channel readiness.
+def _init_vm(plan: SmolmachinesBottlePlan, proxy_host: str) -> None:
+    """Repair filesystem ownership, enforce loopback isolation, and
+    wait for exec channel readiness.
 
     Ownership repair: smolvm's pack process remaps files to the host
     invoker's uid (e.g. 501 on macOS, 1000 on Linux). The chowns use
     names not numbers so they're correct on either. /home/node must
-    be node:node so
-    Claude Code can write ~/.claude.json; /tmp + /var/tmp need root
-    mode 1777 so non-root processes can create per-uid scratch dirs.
-    All folded into one sh -c to avoid back-to-back exec calls
-    immediately after machine_start (libkrun exec-channel race).
+    be node:node so Claude Code can write ~/.claude.json; /tmp +
+    /var/tmp need root mode 1777 so non-root processes can create
+    per-uid scratch dirs.
+
+    Loopback isolation (Linux only): on macOS, smolvm's TSI allowlist
+    restricts which host loopback IPs the guest can reach. On Linux,
+    the allowlist DB patch crashes TSI boot, so we enforce the same
+    /32 scope with guest-side iptables instead. The rules allow
+    outbound to proxy_host/32, then reject all other 127.0.0.0/8.
+    Since the agent runs as non-root (node), it cannot flush these
+    rules.
 
     mkdir -p guards: when booting from a committed snapshot, /tmp and
     /var/tmp are excluded from the archive (they're ephemeral and their
@@ -342,13 +355,39 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
         "chown root:root /tmp /var/tmp && "
         "chmod 1777 /tmp /var/tmp",
     ])
+    if not _loopback._is_macos():
+        _enforce_loopback_isolation(plan.machine_name, proxy_host)
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
-def _proxy_host(plan: SmolmachinesBottlePlan, loopback_ip: str) -> str:
-    """Return the per-bottle host address used for TSI and forwarders."""
-    del plan
-    return loopback_ip
+def _enforce_loopback_isolation(machine_name: str, proxy_host: str) -> None:
+    """Install iptables rules restricting the guest to proxy_host/32.
+
+    On Linux, smolvm's TSI bridges the full host loopback into the
+    guest, and the allow-cidr DB patch is incompatible with TSI.
+    Guest-side iptables achieves the same per-bottle isolation:
+    only the allocated loopback alias is reachable, so the agent
+    can't probe other bottles' ports or other host services.
+
+    Runs as root (exec default); the agent process runs as `node`
+    (non-root) and cannot modify iptables rules."""
+    result = _smolvm.machine_exec(machine_name, [
+        "sh", "-c",
+        # Allow the bottle's allocated loopback alias.
+        f"iptables -A OUTPUT -d {proxy_host}/32 -j ACCEPT && "
+        # Drop all other loopback destinations. DROP (not REJECT)
+        # because the libkrun kernel lacks the REJECT target module.
+        # Connections to blocked addresses time out rather than fail
+        # fast, which is acceptable — the agent shouldn't be probing
+        # non-allowed loopback addresses.
+        "iptables -A OUTPUT -d 127.0.0.0/8 -j DROP",
+    ])
+    if result.returncode != 0:
+        warn(
+            f"guest iptables setup failed (exit {result.returncode}): "
+            f"{(result.stderr or '').strip() or '<no stderr>'}. "
+            f"Per-bottle loopback isolation is not enforced."
+        )
 
 
 def _label_for_port(port: int) -> str:
@@ -373,6 +412,62 @@ def _port_for_label(label: str) -> int:
     raise ValueError(f"unknown sidecar forward label: {label}")
 
 
+def _stage_sidecar_data(plan: SmolmachinesBottlePlan) -> Path:
+    """Stage egress + git-gate files into one virtiofs-mountable dir.
+
+    libkrun limits total mounts + port-mappings to 5. With 3 daemon
+    ports the sidecar VM can carry at most 2 mounts (the second is
+    the supervise DB). Egress and git-gate share a single mount:
+
+        <staging>/egress/     → _EGRESS_CONFDIR_IN_VM
+        <staging>/git-gate/   → _GIT_GATE_SCRIPTS_DIR_IN_VM
+
+    The mount is writable so mitmproxy can write combined-trust.pem
+    and cache per-host certs under the egress subdir."""
+    staging = egress_state_dir(plan.slug) / "smolvm-sidecar-data"
+
+    # --- egress subdir ---
+    confdir = staging / "egress"
+    confdir.mkdir(parents=True, exist_ok=True)
+    ep = plan.egress_plan
+    shutil.copy2(str(ep.mitmproxy_ca_host_path), str(confdir / "mitmproxy-ca.pem"))
+    if ep.routes:
+        shutil.copy2(str(ep.routes_path), str(confdir / "routes.yaml"))
+
+    # --- git-gate subdir (only when upstreams are configured) ---
+    gp = plan.git_gate_plan
+    if gp.upstreams:
+        scripts_dir = staging / "git-gate"
+        scripts_dir.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(str(gp.entrypoint_script), str(scripts_dir / "entrypoint.sh"))
+        shutil.copy2(str(gp.hook_script), str(scripts_dir / "pre-receive"))
+        shutil.copy2(str(gp.access_hook_script), str(scripts_dir / "access-hook"))
+        for name in ("entrypoint.sh", "pre-receive", "access-hook"):
+            (scripts_dir / name).chmod(0o755)
+
+        # Patch paths: the rendered entrypoint hardcodes /git-gate/creds/
+        # and /etc/git-gate/ for hooks; rewrite both to the in-VM subdir.
+        ep_path = scripts_dir / "entrypoint.sh"
+        text = ep_path.read_text()
+        text = text.replace("/git-gate/creds/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/creds/")
+        text = text.replace("/etc/git-gate/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/")
+        ep_path.write_text(text)
+
+        creds_dir = scripts_dir / "creds"
+        creds_dir.mkdir(exist_ok=True)
+        for u in gp.upstreams:
+            keypath = Path(expand_tilde(u.identity_file))
+            dest_key = creds_dir / f"{u.name}-key"
+            shutil.copy2(str(keypath), str(dest_key))
+            dest_key.chmod(0o600)
+            if u.known_hosts_file:
+                dest_kh = creds_dir / f"{u.name}-known_hosts"
+                shutil.copy2(str(u.known_hosts_file), str(dest_kh))
+                dest_kh.chmod(0o600)
+
+    return staging
+
+
 def _bundle_launch_spec(
     plan: SmolmachinesBottlePlan, network: str, proxy_host: str,
 ) -> _bundle.BundleLaunchSpec:
@@ -391,35 +486,26 @@ def _bundle_launch_spec(
     env: list[str] = []
     volumes: list[tuple[str, str, bool]] = []
 
-    # --- egress -----------------------------------------------
+    # --- egress + git-gate (single mount) ---------------------
+    # Stage both into one dir and mount it at _SIDECAR_DATA_DIR_IN_VM.
+    # libkrun limits mounts + port-mappings to 5; with 3 daemon ports
+    # we can carry at most 2 mounts (this one + supervise DB).
+    # Writable so egress_entrypoint.sh can write combined-trust.pem
+    # and mitmproxy can create its per-host cert cache.
     ep = plan.egress_plan
-    volumes.append((str(ep.mitmproxy_ca_host_path), EGRESS_CA_IN_CONTAINER, True))
-    if ep.routes:
-        volumes.append((str(ep.routes_path.parent), str(Path(EGRESS_ROUTES_IN_CONTAINER).parent), True))
+    gp = plan.git_gate_plan
+    staging = _stage_sidecar_data(plan)
+    volumes.append((str(staging), _SIDECAR_DATA_DIR_IN_VM, False))
+    # Tell the egress entrypoint where to find its CA + routes.
+    env.append(f"EGRESS_CONFDIR={_EGRESS_CONFDIR_IN_VM}")
+    # Always set EGRESS_ROUTES so the addon reads from the confdir path
+    # even when no routes were configured at launch (apply_routes_change
+    # writes here and a SIGHUP causes the addon to pick them up).
+    env.append(f"EGRESS_ROUTES={_EGRESS_CONFDIR_IN_VM}/routes.yaml")
     env.extend(egress_sidecar_env_entries(ep))
 
-    # --- git-gate ---------------------------------------------
-    gp = plan.git_gate_plan
     if gp.upstreams:
         daemons += ["git-gate", "git-http"]
-        volumes += [
-            (str(gp.entrypoint_script), GIT_GATE_ENTRYPOINT_IN_CONTAINER, True),
-            (str(gp.hook_script), GIT_GATE_HOOK_IN_CONTAINER, True),
-            (str(gp.access_hook_script), GIT_GATE_ACCESS_HOOK_IN_CONTAINER, True),
-        ]
-        for u in gp.upstreams:
-            keypath = expand_tilde(u.identity_file)
-            volumes.append((
-                keypath,
-                f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-key",
-                True,
-            ))
-            if u.known_hosts_file:
-                volumes.append((
-                    str(u.known_hosts_file),
-                    f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-known_hosts",
-                    True,
-                ))
 
     # --- supervise --------------------------------------------
     sp = plan.supervise_plan
@@ -430,7 +516,13 @@ def _bundle_launch_spec(
             f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
             f"SUPERVISE_PORT={SUPERVISE_PORT}",
         ]
-        volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
+        # virtiofs requires directory mount — mount the DB's parent
+        # dir so bot-bottle.db lands at the right in-VM path.
+        volumes.append((
+            str(sp.db_path.parent),
+            str(Path(DB_PATH_IN_CONTAINER).parent),
+            False,
+        ))
 
     # Container ports the agent reaches from the smolvm guest —
     # published on `proxy_host` so the TSI allowlist and the docker
@@ -478,13 +570,8 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
         committed_path = Path(committed)
         if committed_path.is_file():
             info(f"using committed smolmachine {str(committed_path)!r}")
-            if _image_policy(plan) == "cached":
-                warn_if_stale_path("agent smolmachine artifact", committed_path)
             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
@@ -495,35 +582,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
     )
 
 
-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}")
-    warn_if_stale_path(f"{label} smolmachine artifact", sidecar)
-    return sidecar
-
-
 def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:
     """Build the agent docker image and convert it into a
     `.smolmachine` artifact, caching the result under

bot_bottle/backend/smolmachines/loopback_alias.py

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

bot_bottle/backend/smolmachines/resolve_plan.py

@@ -56,13 +56,11 @@ def resolve_plan(
     git_gate_plan: GitGatePlan,
     stage_dir: Path,
 ) -> SmolmachinesBottlePlan:
-    """Materialize the smolmachines plan. The bundle's docker
-    subnet + pinned IP are derived from the slug; the agent's
-    `.smolmachine` artifact is built (or cache-hit) here so
-    launch's `machine create --from` boots without a registry
-    pull. Per-bottle guest env + the TSI allow_cidrs land on the
-    plan for launch to pass straight through to
-    `machine create` flags."""
+    """Materialize the smolmachines plan. The agent `.smolmachine`
+    artifact is built (or cache-hit) here so launch's
+    `machine create --from` boots without a registry pull. Per-bottle
+    guest env lands on the plan for launch to pass straight through
+    to `machine create` flags."""
 
     # ==== smolmachines specific setup ====
     subnet, gateway, bundle_ip = smolmachines_bundle_subnet(slug)

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -1,36 +1,18 @@
-"""Per-bottle sidecar bundle bringup for the smolmachines backend
-(PRD 0023).
+"""Per-bottle sidecar bundle bringup for the smolmachines backend.
 
-Two docker resources per bottle live here:
-
-  - **A dedicated bridge network**, subnet derived from the slug.
-    The bundle container gets a pinned IP at `<subnet>.2` so the
-    smolvm guest's TSI allowlist (`<bundle-ip>/32`) has a stable
-    target. Without pinning, we'd have to inspect the container's
-    assigned IP after start and feed it back into the Smolfile
-    — a race we can sidestep with `--ip`.
-
-  - **The bundle container itself**, running the PRD 0024 bundle
-    image (`bot-bottle-sidecars:latest` by default). Same
-    image, same daemons, same daemon-private env / bind-mounts
-    as the docker backend.
-
-This module ships the lifecycle primitives only — create
-network, start bundle, stop bundle, remove network — wrapped
-around `subprocess.run(["docker", ...])`. Wiring them into the
-launch flow + populating the `BundleLaunchSpec` from the inner
-Plans (EgressPlan, …) lands in chunk 2d."""
+The sidecar bundle runs as its own smolVM. The agent VM reaches
+bundle daemons through host-loopback ports published by that sidecar
+VM and wrapped by per-bottle address-bound forwarders."""
 
 from __future__ import annotations
 
 import os
 import socket
-import subprocess
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Sequence
 
-from ...log import die, warn
+from ...log import warn
 from ..docker import util as docker_mod
 from ..docker.sidecar_bundle import (
     SIDECAR_BUNDLE_DOCKERFILE,
@@ -78,10 +60,8 @@ class BundleLaunchSpec:
     # supervisor inside the bundle reads it to skip
     # bottle-irrelevant daemons (e.g. supervise=False bottles).
     daemons_csv: str = "egress"
-    # Plain "KEY=VALUE" strings + "KEY" bare names (the bare-name
-    # form inherits the value from the docker-run subprocess env,
-    # matching the docker backend's compose-up secret-forwarding
-    # pattern).
+    # Plain "KEY=VALUE" strings + "KEY" bare names. Bare names inherit
+    # from the host env passed to the sidecar VM launch.
     environment: Sequence[str] = field(default_factory=tuple)
     # (host_path, container_path, read_only) bind mounts.
     volumes: Sequence[tuple[str, str, bool]] = field(default_factory=tuple)
@@ -155,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,
@@ -182,138 +172,3 @@ def stop_bundle_vm(slug: str) -> None:
         _smolvm.machine_delete(name)
     except _smolvm.SmolvmError as exc:
         warn(f"smolvm machine delete {name} failed: {exc}")
-
-
-def create_bundle_network(network_name: str, subnet: str, gateway: str) -> None:
-    """`docker network create` with an explicit subnet + gateway
-    so the bundle's `--ip` lands on the address the Smolfile's
-    TSI allowlist points at. Idempotent on the caller's side —
-    `start_bundle` catches the "network exists" error and treats
-    it as success (chunk-2d teardown is paired with each create).
-    """
-    result = subprocess.run(
-        ["docker", "network", "create",
-         "--subnet", subnet, "--gateway", gateway,
-         network_name],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        # Already-exists is fine on a resume path; everything else
-        # is fatal — the bundle won't have an addressable network.
-        if "already exists" in (result.stderr or "").lower():
-            return
-        die(
-            f"docker network create {network_name} failed: "
-            f"{(result.stderr or '').strip()}"
-        )
-
-
-def remove_bundle_network(network_name: str) -> None:
-    """Idempotent: a missing network returns success."""
-    result = subprocess.run(
-        ["docker", "network", "rm", network_name],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode == 0:
-        return
-    if "no such network" in (result.stderr or "").lower():
-        return
-    # Network with attached containers is the common non-fatal
-    # case during a partial teardown — warn but don't die.
-    warn(
-        f"docker network rm {network_name} failed: "
-        f"{(result.stderr or '').strip()}"
-    )
-
-
-def start_bundle(spec: BundleLaunchSpec, *,
-                 env: dict[str, str] | None = None) -> None:
-    """Bring the bundle container up on the per-bottle bridge with
-    the pinned IP. Argv is built deterministically from `spec`;
-    `env` is the host subprocess env (forwarded values for any
-    bare-name entries in `spec.environment`)."""
-    container = bundle_container_name(spec.slug)
-    argv = [
-        "docker", "run",
-        "--name", container,
-        "--detach",
-        "--rm",
-        "--network", spec.network_name,
-        "--ip", spec.bundle_ip,
-        "-e", f"BOT_BOTTLE_SIDECAR_DAEMONS={spec.daemons_csv}",
-    ]
-    for entry in spec.environment:
-        argv += ["-e", entry]
-    for host_path, container_path, read_only in spec.volumes:
-        suffix = ":ro" if read_only else ""
-        argv += ["-v", f"{host_path}:{container_path}{suffix}"]
-    # Loopback-only host port-forwards — the smolvm guest's TSI
-    # uses macOS networking, and macOS loopback is the only host
-    # surface that round-trips into Docker Desktop's daemon VM.
-    # Binds to the per-bottle alias so TSI's IP-only allowlist
-    # narrows reachability to this bottle's bundle only.
-    for port in spec.ports_to_publish:
-        argv += ["-p", f"{spec.publish_host_ip}::{port}"]
-    argv.append(spec.image)
-    result = subprocess.run(
-        argv, capture_output=True, text=True,
-        env=dict(env) if env is not None else None, check=False,
-    )
-    if result.returncode != 0:
-        die(
-            f"docker run for bundle {container} failed: "
-            f"{(result.stderr or '').strip()}"
-        )
-
-
-def bundle_host_port(
-    slug: str, container_port: int, *, host_ip: str = "127.0.0.1",
-) -> int:
-    """`docker port <bundle> <container_port>/tcp` → the random
-    host-side port docker assigned for the binding on `host_ip`.
-    Called after `start_bundle` on each container port listed in
-    `BundleLaunchSpec.ports_to_publish` so the launch step can
-    build the agent's HTTPS_PROXY / GIT_GATE / SUPERVISE URLs in
-    `<host_ip>:<host port>` form."""
-    container = bundle_container_name(slug)
-    result = subprocess.run(
-        ["docker", "port", container, f"{container_port}/tcp"],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        die(
-            f"docker port {container} {container_port}/tcp failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}"
-        )
-    # Each line looks like `127.0.0.16:54321` — one per address
-    # family / host IP. Match on the expected host_ip prefix so
-    # bottles bound to per-bottle aliases pick the right line.
-    for raw in (result.stdout or "").splitlines():
-        line = raw.strip()
-        if line.startswith(f"{host_ip}:"):
-            _, _, port_str = line.rpartition(":")
-            try:
-                return int(port_str)
-            except ValueError:
-                die(f"unexpected `docker port` output: {line!r}")
-    die(
-        f"no port mapping on {host_ip} for {container} "
-        f"{container_port}/tcp; got: {(result.stdout or '').strip()!r}"
-    )
-
-
-def stop_bundle(slug: str) -> None:
-    """Idempotent: a missing container returns success."""
-    container = bundle_container_name(slug)
-    result = subprocess.run(
-        ["docker", "rm", "-f", container],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode == 0:
-        return
-    if "no such container" in (result.stderr or "").lower():
-        return
-    warn(
-        f"docker rm -f {container} failed: "
-        f"{(result.stderr or '').strip()}"
-    )

bot_bottle/cli/start.py

@@ -63,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",
@@ -103,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"
 
@@ -152,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,
@@ -164,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,
@@ -224,7 +209,7 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
-        image_policy="cached" if args.cached_images else "fresh",
+        headless=True,
     )
     return _launch_bottle(
         spec,
@@ -405,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)

bot_bottle/config_store.py

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

bot_bottle/contrib/claude/Dockerfile

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

bot_bottle/contrib/codex/Dockerfile

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

bot_bottle/egress_entrypoint.sh

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

bot_bottle/git_gate_host_key.py

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

bot_bottle/image_cache.py

@@ -1,38 +0,0 @@
-"""Shared helpers for cached-image quickstart warnings."""
-
-from __future__ import annotations
-
-from datetime import datetime, timezone
-from pathlib import Path
-
-try:
-    from .config_store import ConfigStore
-    from .log import warn
-except ImportError:
-    from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
-    from log import warn  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
-
-
-def warn_if_stale(label: str, created_at: datetime) -> None:
-    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
-    warn(
-        f"cached {label} is {age.days} day(s) old; "
-        "quickstart does not verify it matches the current Dockerfile/context"
-    )
-
-
-def warn_if_stale_path(label: str, path: Path) -> None:
-    warn_if_stale(
-        label,
-        datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc),
-    )
-
-
-__all__ = ["warn_if_stale", "warn_if_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"]

bot_bottle/yaml_subset.py

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

docs/research/landscape-containerized-claude.md

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

requirements-dev.txt

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

tests/integration/test_smolmachines_bundle_bringup.py

@@ -1,111 +0,0 @@
-"""Integration: PRD 0023 chunk 2c — bundle bringup on a per-bottle
-docker bridge with the pinned IP.
-
-End-to-end against the real docker daemon. Brings up just the
-sidecar bundle on its own bridge, confirms the container lands at
-the pinned IP, then tears down. Skipped under act_runner (docker
-socket mount topology breaks bridge visibility) and when the
-bundle image isn't available.
-
-Full launch flow (smolvm + bundle + provisioning + the
-localhost-reach / egress-port-bypass probes) lives in chunk 2d."""
-
-from __future__ import annotations
-
-import os
-import subprocess
-import time
-import unittest
-
-from bot_bottle.backend.smolmachines.sidecar_bundle import (
-    BundleLaunchSpec,
-    bundle_container_name,
-    bundle_network_name,
-    create_bundle_network,
-    remove_bundle_network,
-    start_bundle,
-    stop_bundle,
-)
-from tests._docker import skip_unless_docker
-
-
-@skip_unless_docker()
-@unittest.skipIf(
-    os.environ.get("GITEA_ACTIONS") == "true",
-    "skipped under act_runner: docker socket mount topology breaks "
-    "in-process visibility of networks created on the host daemon",
-)
-class TestBundleBringup(unittest.TestCase):
-    def setUp(self):
-        self.slug = f"cb-test-bundle-{os.getpid()}-{int(time.time())}"
-        self.network = bundle_network_name(self.slug)
-        self.container = bundle_container_name(self.slug)
-
-    def tearDown(self):
-        stop_bundle(self.slug)
-        remove_bundle_network(self.network)
-
-    def _bundle_image_built(self) -> bool:
-        """The bundle image (`bot-bottle-sidecars:latest`) is
-        built lazily by the docker backend's compose. If a
-        smolmachines-only operator hasn't run the docker backend
-        first, the image won't exist locally. Skip rather than
-        fail."""
-        r = subprocess.run(
-            ["docker", "image", "inspect", "bot-bottle-sidecars:latest"],
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            check=False,
-        )
-        return r.returncode == 0
-
-    def test_create_network_then_start_bundle_pins_ip(self):
-        if not self._bundle_image_built():
-            self.skipTest(
-                "bot-bottle-sidecars:latest not built; run a docker "
-                "bottle first or `docker build -f Dockerfile.sidecars .`"
-            )
-
-        # Pick a subnet unlikely to collide on the host. Last
-        # octet of the slug hash isn't deterministic across runs;
-        # we hardcode a high octet (.211) that the docker default
-        # bridges almost never use.
-        subnet = "192.168.211.0/24"
-        gateway = "192.168.211.1"
-        bundle_ip = "192.168.211.2"
-
-        create_bundle_network(self.network, subnet, gateway)
-
-        spec = BundleLaunchSpec(
-            slug=self.slug,
-            network_name=self.network,
-            subnet=subnet,
-            gateway=gateway,
-            bundle_ip=bundle_ip,
-            # Empty daemons_csv → init exits "no daemons selected"
-            # immediately. We just need the container to land on
-            # the network at the right IP before it exits.
-            daemons_csv="",  # empty → init exits "no daemons selected"
-        )
-        start_bundle(spec)
-
-        # Inspect the container's IP on the per-bottle network.
-        r = subprocess.run(
-            ["docker", "inspect",
-             "--format",
-             "{{(index .NetworkSettings.Networks \"" + self.network + "\").IPAddress}}",
-             self.container],
-            capture_output=True, text=True, check=False,
-        )
-        # Container may have exited (no daemons selected → exit 0).
-        # The inspect still works on exited containers as long as
-        # `--rm` hasn't fired yet, which is a race. Even if it has,
-        # the launch succeeded — the container existed, on the
-        # right network, at the right IP. We don't fail here on
-        # missing inspect.
-        if r.returncode == 0 and r.stdout.strip():
-            self.assertEqual(bundle_ip, r.stdout.strip(),
-                             f"bundle landed at wrong IP: {r.stdout!r}")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/integration/test_smolmachines_launch.py

@@ -1,10 +1,9 @@
-"""Integration: PRD 0023 chunk 2d — end-to-end launch + exec
-round trip + the acceptance probes.
+"""Integration: end-to-end smolmachines launch + exec round trip.
 
-The smoke confirms the launch flow (per-bottle docker bridge →
-sidecar bundle with host-loopback published ports → smolvm guest
-with TSI allowlist → exec) plumbs together end to end. The probes confirm the
-security properties the design pivot was about:
+The smoke confirms the launch flow (sidecar bundle smolVM →
+host-loopback forwarders → agent smolVM with TSI allowlist → exec)
+plumbs together end to end. The probes confirm the security
+properties the design pivot was about:
 
   - **localhost-reach probe** — guest tries to dial a service
     bound on the host's `127.0.0.1`. TSI's per-bottle loopback
@@ -14,13 +13,6 @@ security properties the design pivot was about:
     the injected `HTTPS_PROXY`/`HTTP_PROXY` URL on the per-bottle
     loopback alias, while direct egress with proxy vars unset fails.
 
-  - **egress-port-bypass probe** — guest tries to dial
-    `<bundle-ip>:9099` (egress's port). TSI permits the IP but
-    the bundle's egress daemon binds `127.0.0.1` inside its
-    container, so the connect refuses at the socket level. The
-    bind-address mitigation is what closes TSI's port-granularity
-    gap.
-
 Gated on macOS/Linux + smolvm + docker + not GITEA_ACTIONS — the
 runner can't host libkrun-backed VMs."""
 
@@ -114,8 +106,8 @@ class TestSmolmachinesLaunch(unittest.TestCase):
 
     def test_localhost_reach_probe(self):
         # Agent dials a 127.0.0.1 service on the host. TSI's
-        # allowlist contains only <bundle-ip>/32, so this must
-        # refuse. We use a port unlikely to be bound on the host
+        # allowlist contains only the per-bottle loopback alias, so
+        # this must refuse. We use a port unlikely to be bound on the host
         # (high-numbered) so we're confirming TSI refusal, not
         # just "no service listening."
         r = self.bottle.exec(
@@ -196,28 +188,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
         self.assertEqual(0, r.returncode, msg=r.stderr)
         self.assertEqual(_AGENT_PROMPT, r.stdout.rstrip("\n"))
 
-    def test_egress_port_bypass_probe(self):
-        # Agent dials <bundle-ip>:9099 (egress's port). TSI
-        # permits the IP, but egress will bind 127.0.0.1:9099
-        # inside the bundle in chunk 3, so the connect refuses
-        # at the socket level. NOTE: in chunk 2d the bundle's
-        # daemons aren't running (daemons_csv=""), so nothing
-        # is listening on :9099 anyway — this test asserts the
-        # connect fails, which is the property chunk 3 will
-        # preserve once egress is actually running.
-        r = self.bottle.exec(
-            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
-            f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
-            "2>&1 || true"
-        )
-        self.assertTrue(
-            "refused" in r.stdout.lower()
-            or "timed out" in r.stdout.lower()
-            or "unreachable" in r.stdout.lower()
-            or "failed" in r.stdout.lower(),
-            f"expected egress port refusal; got: {r.stdout!r}",
-        )
-
 
 if __name__ == "__main__":
     unittest.main()

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_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,
@@ -302,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,59 +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())
-
-
-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
 
 
@@ -105,10 +103,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
             launch_mod.docker_mod, "image_exists", return_value=image_present,
         ), mock.patch.object(
             launch_mod.docker_mod, "build_image", side_effect=fake_build,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "image_created_at",
-        ), mock.patch.object(
-            launch_mod, "warn_if_stale",
         ), mock.patch.object(
             launch_mod, "egress_tls_init",
             return_value=(Path("/egress_ca"), Path("/egress_cert")),
@@ -186,24 +180,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
         built = self._run_launch(plan, committed_tag=None)
         self.assertEqual([_DEFAULT_IMAGE], built)
 
-    def test_cached_images_skip_build_when_present(self) -> None:
-        base = _plan(self._tmp)
-        plan = dataclasses.replace(
-            base,
-            spec=dataclasses.replace(base.spec, image_policy="cached"),
-        )
-        built = self._run_launch(plan, committed_tag=None, image_present=True)
-        self.assertEqual([], built)
-
-    def test_cached_images_die_when_agent_missing(self) -> None:
-        base = _plan(self._tmp)
-        plan = dataclasses.replace(
-            base,
-            spec=dataclasses.replace(base.spec, image_policy="cached"),
-        )
-        with self.assertRaises(Die):
-            self._run_launch(plan, committed_tag=None, image_present=False)
-
     def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
         plan = _plan(self._tmp)
         built = self._run_launch(

tests/unit/test_docker_util_image.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 
 import subprocess
 import unittest
-from datetime import timezone
 from unittest.mock import patch
 
 from bot_bottle.backend.docker import util as docker_mod
@@ -55,22 +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],
-        )
-
-
 class TestSave(unittest.TestCase):
     def test_save_runs_docker_save(self):
         with patch.object(

tests/unit/test_git_gate_host_key.py

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

tests/unit/test_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,56 +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, "warn_if_stale_path",
-            ), patch.object(
-                _launch_mod, "_ensure_smolmachine",
-            ) as ensure:
-                result = _launch_mod._agent_from_path(cast(Any, plan))
-
-        self.assertEqual(artifact, result)
-        ensure.assert_not_called()
-
-    def test_cached_policy_dies_when_artifact_missing(self):
-        with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
-            plan = SimpleNamespace(
-                slug="dev-abc12",
-                agent_image="bot-bottle-claude:latest",
-                spec=SimpleNamespace(image_policy="cached"),
-            )
-            with patch.object(
-                _launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
-            ), patch.object(
-                _launch_mod, "read_committed_image", return_value="",
-            ), patch.object(
-                _launch_mod.docker_mod, "image_exists", return_value=True,
-            ), patch.object(
-                _launch_mod.docker_mod, "image_id",
-                return_value="sha256:abcdef0123456789fffffffffffffffff",
-            ):
-                from bot_bottle.log import Die
-                with self.assertRaises(Die):
-                    _launch_mod._agent_from_path(cast(Any, plan))
-
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_loopback_alias.py

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

tests/unit/test_smolmachines_provision.py

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

tests/unit/test_smolmachines_sidecar_bundle.py

@@ -1,9 +1,4 @@
-"""Unit: bundle bringup primitives for the smolmachines backend
-(PRD 0023 chunk 2c).
-
-Tests mock `subprocess.run` and assert on the docker argv shape.
-The end-to-end integration smoke (real docker daemon, real
-bundle image) lands in chunk 2d."""
+"""Unit: bundle bringup primitives for the smolmachines backend."""
 
 from __future__ import annotations
 
@@ -17,28 +12,12 @@ from bot_bottle.backend.smolmachines.sidecar_bundle import (
     allocate_raw_host_ports,
     bundle_container_name,
     bundle_network_name,
-    create_bundle_network,
     ensure_bundle_image,
     start_bundle_vm,
-    remove_bundle_network,
-    start_bundle,
-    stop_bundle,
     stop_bundle_vm,
 )
 
 
-def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:  # type: ignore
-    return subprocess.CompletedProcess(
-        args=[], returncode=0, stdout=stdout, stderr=stderr,
-    )
-
-
-def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
-    return subprocess.CompletedProcess(
-        args=[], returncode=1, stdout="", stderr=stderr,
-    )
-
-
 def _spec(**kwargs) -> BundleLaunchSpec:  # type: ignore
     defaults = dict(
         slug="demo-abc12",
@@ -71,123 +50,6 @@ class TestNamingHelpers(unittest.TestCase):
         )
 
 
-class TestNetworkLifecycle(unittest.TestCase):
-    def _patch_run(self, **kwargs):  # type: ignore
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            **kwargs,
-        )
-
-    def test_create_argv_explicit_subnet_and_gateway(self):
-        with self._patch_run(return_value=_ok()) as m:
-            create_bundle_network("nn", "192.168.50.0/24", "192.168.50.1")
-        self.assertEqual(
-            ["docker", "network", "create",
-             "--subnet", "192.168.50.0/24",
-             "--gateway", "192.168.50.1",
-             "nn"],
-            m.call_args.args[0],
-        )
-
-    def test_create_treats_existing_network_as_success(self):
-        with self._patch_run(return_value=_fail("network nn already exists")):
-            # No SystemExit.
-            create_bundle_network("nn", "192.168.50.0/24", "192.168.50.1")
-
-    def test_create_other_failure_is_fatal(self):
-        with self._patch_run(return_value=_fail("invalid subnet")):
-            with self.assertRaises(SystemExit):
-                create_bundle_network("nn", "bogus", "bogus")
-
-    def test_remove_missing_network_is_idempotent(self):
-        # No SystemExit / no warn-and-continue noise; missing
-        # network is the expected case during a partial teardown.
-        with self._patch_run(return_value=_fail("Error: No such network: nn")):
-            remove_bundle_network("nn")
-
-    def test_remove_clean_returns_success(self):
-        with self._patch_run(return_value=_ok()):
-            remove_bundle_network("nn")
-
-
-class TestStartBundle(unittest.TestCase):
-    def _patch_run(self):
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_ok(),
-        )
-
-    def test_argv_pins_ip_on_network(self):
-        with self._patch_run() as m:
-            start_bundle(_spec())
-        argv = m.call_args.args[0]
-        # --network NETNAME --ip <bundle-ip> on the docker run.
-        self.assertIn("--network", argv)
-        self.assertIn("bot-bottle-bundle-demo-abc12", argv)
-        self.assertIn("--ip", argv)
-        self.assertIn("192.168.50.2", argv)
-        # Detached and auto-removed.
-        self.assertIn("--detach", argv)
-        self.assertIn("--rm", argv)
-        # Container name uses the per-slug bundle prefix.
-        i = argv.index("--name")
-        self.assertEqual("bot-bottle-sidecars-demo-abc12", argv[i + 1])
-        # Image at the end.
-        self.assertEqual("bot-bottle-sidecars:latest", argv[-1])
-
-    def test_daemons_env_passed_in(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(daemons_csv="egress,supervise"))
-        argv = m.call_args.args[0]
-        self.assertIn("-e", argv)
-        self.assertIn(
-            "BOT_BOTTLE_SIDECAR_DAEMONS=egress,supervise",
-            argv,
-        )
-
-    def test_environment_entries_pass_through(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(environment=(
-                "SUPERVISE_BOTTLE_SLUG=demo-abc12",
-                "EGRESS_TOKEN_0",  # bare-name → host env inherit
-            )))
-        argv = m.call_args.args[0]
-        self.assertIn("SUPERVISE_BOTTLE_SLUG=demo-abc12", argv)
-        self.assertIn("EGRESS_TOKEN_0", argv)
-
-    def test_volumes_render_with_ro_flag(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(volumes=(
-                ("/host/egress-ca.pem", "/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", True),
-                ("/host/queue", "/run/supervise/queue", False),
-            )))
-        argv = m.call_args.args[0]
-        self.assertIn(
-            "/host/egress-ca.pem:/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem:ro",
-            argv,
-        )
-        self.assertIn("/host/queue:/run/supervise/queue", argv)
-
-    def test_failure_dies(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_fail("invalid mount"),
-        ):
-            with self.assertRaises(SystemExit):
-                start_bundle(_spec())
-
-    def test_host_env_inherited_to_subprocess(self):
-        # Bare-name entries in spec.environment rely on the docker
-        # subprocess being run with the host env. Confirm `env=`
-        # threads through.
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_ok(),
-        ) as m:
-            start_bundle(_spec(), env={"FOO": "bar"})
-        self.assertEqual({"FOO": "bar"}, m.call_args.kwargs["env"])
-
-
 class TestEnsureBundleImage(unittest.TestCase):
     def test_builds_sidecar_dockerfile_before_plain_docker_run(self):
         with patch(
@@ -263,28 +125,6 @@ class TestStartBundleVm(unittest.TestCase):
         )
 
 
-class TestStopBundle(unittest.TestCase):
-    def _patch_run(self, **kwargs):  # type: ignore
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            **kwargs,
-        )
-
-    def test_argv_force_removes(self):
-        with self._patch_run(return_value=_ok()) as m:
-            stop_bundle("demo-abc12")
-        self.assertEqual(
-            ["docker", "rm", "-f", "bot-bottle-sidecars-demo-abc12"],
-            m.call_args.args[0],
-        )
-
-    def test_missing_container_is_idempotent(self):
-        with self._patch_run(return_value=_fail(
-            "Error: No such container: bot-bottle-sidecars-demo-abc12"
-        )):
-            stop_bundle("demo-abc12")  # no raise
-
-
 class TestStopBundleVm(unittest.TestCase):
     def test_stops_then_deletes_sidecar_vm(self):
         with patch(

tests/unit/test_yaml_subset.py

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