test(smolmachines): verify Linux sandbox escape gate + fix platform check

Empirical verification on NixOS/Linux with /dev/kvm + smolvm 1.4.7:

  1. DB path confirmed: ~/.local/share/smolvm/server/smolvm.db exists,
     schema matches (vms table, data BLOB with allowed_cidrs JSON field).

  2. smolvm 1.4.7 still silently drops --allow-cidr with --from
     (allowed_cidrs=None in DB after create). force_allowlist patch
     is still necessary and correct.

  3. All 5 sandbox-escape attacks blocked with BOT_BOTTLE_BACKEND=smolmachines
     on Linux: hostname/IP allowlist, HTTP DLP, DNS exfil, git-gate gitleaks.

Changes:
- test_sandbox_escape: remove darwin-only guard; allow linux too
- contrib/claude/Dockerfile: add dnsutils (dig) for attack 4 DNS test

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

README.md

@@ -5,7 +5,7 @@
 # bot-bottle
 
 [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
-[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
 [![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
 
 **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.

bot_bottle/backend/__init__.py

@@ -75,12 +75,6 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
-    # True when launched via --headless (no TTY, no interactive prompts).
-    # The git-gate host-key preflight uses this to error rather than prompt.
-    headless: bool = False
-    # Image startup policy. "fresh" preserves the normal build path;
-    # "cached" reuses the current local image/artifact without rebuilding.
-    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -306,13 +300,6 @@ 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 info, warn
-from ...log import die, 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/egress_apply.py

@@ -1,51 +1,20 @@
 """Egress apply for the smolmachines backend.
 
-The smolmachines sidecar bundle runs as a sidecar smolVM. Route-file
+The smolmachines sidecar bundle runs as a host-side Docker container,
-inspection and reload signalling therefore go through ``smolvm machine
+so egress signalling is identical to the docker backend.
-exec`` instead of Docker.
 """
 
 from __future__ import annotations
 
-from ...egress import EGRESS_ROUTES_IN_CONTAINER
+from ..docker.egress_apply import (  # noqa: F401
-from ...log import warn
+    DockerEgressApplicator,
-from ..egress_apply import EgressApplicator, EgressApplyError
+    EgressApplyError,
-from . import sidecar_bundle as _bundle
+    applicator,
-from . import smolvm as _smolvm
+    fetch_current_routes,
-
+)
-
-def fetch_current_routes(slug: str) -> str:
-    machine = _bundle.bundle_machine_name(slug)
-    result = _smolvm.machine_exec(machine, ["cat", EGRESS_ROUTES_IN_CONTAINER])
-    if result.returncode != 0:
-        raise EgressApplyError(
-            f"could not read routes.yaml from {machine}: "
-            f"{(result.stderr or '').strip() or 'sidecar VM not running?'}"
-        )
-    return result.stdout
-
-
-class SmolmachinesEgressApplicator(EgressApplicator):
-    def _signal_bundle_reload(self, slug: str) -> None:
-        machine = _bundle.bundle_machine_name(slug)
-        result = _smolvm.machine_exec(machine, ["sh", "-c", "kill -HUP 1"])
-        if result.returncode != 0:
-            last_error = (result.stderr or "").strip() or (result.stdout or "").strip()
-            warn(
-                f"egress: routes updated on disk for {slug}, but bundle reload failed: "
-                f"{last_error or 'smolvm exec failed'}"
-            )
-            raise EgressApplyError(
-                f"could not reload egress bundle {machine}: "
-                f"{last_error or 'smolvm exec failed'}"
-            )
-
-
-applicator = SmolmachinesEgressApplicator()
-
 
 __all__ = [
-    "SmolmachinesEgressApplicator",
+    "DockerEgressApplicator",
     "EgressApplyError",
     "applicator",
     "fetch_current_routes",

bot_bottle/backend/smolmachines/enumerate.py

@@ -31,7 +31,6 @@ from . import sidecar_bundle as _bundle
 # matching the bundle container name pattern. We use the prefix
 # both as a filter and to strip back to the slug.
 _VM_NAME_PREFIX = "bot-bottle-"
-_SIDECAR_VM_PREFIX = _bundle.bundle_machine_name("")
 
 
 def enumerate_active() -> list[ActiveAgent]:
@@ -55,11 +54,7 @@ def enumerate_active() -> list[ActiveAgent]:
     for m in machines:
         name = m.get("name") or ""
         state = m.get("state") or ""
-        if (
+        if state != "running" or not name.startswith(_VM_NAME_PREFIX):
-            state != "running"
-            or not name.startswith(_VM_NAME_PREFIX)
-            or name.startswith(_SIDECAR_VM_PREFIX)
-        ):
             continue
         slug = name[len(_VM_NAME_PREFIX):]
         metadata = read_metadata(slug)

bot_bottle/backend/smolmachines/launch.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import dataclasses
 import os
+import platform
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
@@ -45,15 +46,13 @@ 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 info, warn
-from ...log import die, info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
     read_committed_image,
 )
 from . import loopback_alias as _loopback
-from . import port_forward as _forward
 from . import sidecar_bundle as _bundle
 from . import smolvm as _smolvm
 from .bottle import SmolmachinesBottle
@@ -93,8 +92,9 @@ def launch(
     try:
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
-        proxy_host = loopback_ip
+        proxy_host = _proxy_host(plan, loopback_ip)
         plan = _start_bundle(plan, network, proxy_host, stack)
+        plan = _discover_urls(plan, proxy_host)
 
         agent_from_path = _agent_from_path(plan)
 
@@ -144,17 +144,19 @@ def _allocate_resources(
     plan: SmolmachinesBottlePlan,
     stack: ExitStack,
 ) -> tuple[str, str]:
-    """Reserve a per-bottle host address.
+    """Reserve a loopback alias and create the per-bottle docker bridge.
 
-    The per-bottle address scopes TSI's allowlist to this bottle's
+    The per-bottle alias scopes TSI's allowlist to this bottle's
-    forwarder-published ports so the agent can't reach other bottles'
+    published ports so the agent can't reach other bottles' or host
-    or host services. The returned network name remains in the bundle
+    services' ports on loopback. On macOS `ensure_pool` first
-    spec for compatibility with older helper tests; no Docker sidecar
+    sudo-aliases the pool on `lo0`; on Linux that's a no-op since
-    container is launched."""
+    all of 127.0.0.0/8 is already loopback, but the per-bottle
-    del stack
+    allocation runs on both."""
     _loopback.ensure_pool()
     loopback_ip = _loopback.allocate(plan.slug)
     network = _bundle.bundle_network_name(plan.slug)
+    _bundle.create_bundle_network(network, plan.bundle_subnet, plan.bundle_gateway)
+    stack.callback(_bundle.remove_bundle_network, network)
     return loopback_ip, network
 
 
@@ -177,43 +179,15 @@ def _start_bundle(
     proxy_host: str,
     stack: ExitStack,
 ) -> SmolmachinesBottlePlan:
-    """Build the BundleLaunchSpec, start the sidecar VM, wrap its raw
+    """Build the BundleLaunchSpec, resolve token env, start the
-    smolVM-published loopback ports with per-bottle forwarders, stamp
+    sidecar bundle container, and register teardown."""
-    agent URLs from those forwarder ports, and register teardown."""
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
-    if _image_policy(plan) == "cached":
+    _bundle.ensure_bundle_image(bundle_spec.image)
-        artifact = _cached_smolmachine(bundle_spec.image, label="sidecar")
+    _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
-    else:
+    stack.callback(_bundle.stop_bundle, plan.slug)
-        artifact = _ensure_smolmachine(
+    return plan
-            bundle_spec.image,
-            dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-        )
-    launch = _bundle.start_bundle_vm(
-        bundle_spec,
-        from_path=artifact,
-        host_env={**os.environ, **token_env},
-    )
-    stack.callback(_bundle.stop_bundle_vm, plan.slug)
-
-    forward_specs = tuple(
-        _forward.ForwardSpec(
-            label=_label_for_port(container_port),
-            listen_host=proxy_host,
-            listen_port=0,
-            target_host="127.0.0.1",
-            target_port=host_port,
-        )
-        for container_port, host_port in launch.raw_ports.items()
-    )
-    handle = _forward.start_forwarder(forward_specs)
-    stack.callback(_forward.stop_forwarder, handle)
-    published_ports = {
-        _port_for_label(spec.label): spec.listen_port
-        for spec in handle.forwards
-    }
-    return _discover_urls(plan, proxy_host, published_ports)
 
 
 def _provision_git_gate_keys(
@@ -232,27 +206,35 @@ def _provision_git_gate_keys(
 def _discover_urls(
     plan: SmolmachinesBottlePlan,
     proxy_host: str,
-    published_ports: dict[int, int],
 ) -> SmolmachinesBottlePlan:
-    """Stamp URLs + guest_env from per-bottle forwarder ports.
+    """Discover host-side ports for published container ports and
+    return the plan with URLs + guest_env stamped in.
 
-    `proxy_host` is the host IP that both TSI's allowlist and the
+    `proxy_host` is the host IP that both TSI's allowlist and
-    forwarder listeners are keyed to. The raw smolVM-published ports
+    docker's port-forward bindings are keyed to. On macOS it is the
-    are intentionally not advertised to the agent.
+    per-bottle loopback alias; on Linux it is the per-bottle bridge
+    gateway (see `_proxy_host`). The agent dials the published port
+    on this IP for all bundle-hosted services.
 
     NO_PROXY includes `proxy_host` so supervise + git-gate URLs
     bypass HTTPS_PROXY."""
-    agent_facing_host_port = published_ports[_EGRESS_PORT]
+    agent_facing_host_port = _bundle.bundle_host_port(
+        plan.slug, _EGRESS_PORT, host_ip=proxy_host,
+    )
     agent_proxy_url = f"http://{proxy_host}:{agent_facing_host_port}"
 
     agent_git_gate_host = ""
     if plan.git_gate_plan.upstreams:
-        git_gate_host_port = published_ports[_GIT_HTTP_PORT]
+        git_gate_host_port = _bundle.bundle_host_port(
+            plan.slug, _GIT_HTTP_PORT, host_ip=proxy_host,
+        )
         agent_git_gate_host = f"{proxy_host}:{git_gate_host_port}"
 
     agent_supervise_url = ""
     if plan.supervise_plan is not None:
-        supervise_host_port = published_ports[_SUPERVISE_PORT]
+        supervise_host_port = _bundle.bundle_host_port(
+            plan.slug, _SUPERVISE_PORT, host_ip=proxy_host,
+        )
         agent_supervise_url = f"http://{proxy_host}:{supervise_host_port}/"
 
     existing_no_proxy = plan.guest_env.get("NO_PROXY", "localhost,127.0.0.1")
@@ -291,13 +273,14 @@ def _launch_vm(
 ) -> None:
     """Create, patch, and start the smolvm VM; register teardown.
 
-    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias.
+    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias
-    This ensures the guest can only reach sidecar forwarders published
+    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
-    on that IP, not host localhost or another bottle's alias.
+    ensures the guest can only reach bundle ports published on that IP,
-    force_allowlist confirms the allowlist persisted (patching smolvm
+    not the container IP directly. force_allowlist confirms the
-    0.8.0's silent-drop of --allow-cidr when combined with --from) and
+    allowlist persisted (patching smolvm 0.8.0's silent-drop of
-    fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
+    --allow-cidr when combined with --from) and fails closed if it
-    makes --from and --smolfile mutually exclusive."""
+    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
+    and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
     _smolvm.machine_create(
         plan.machine_name,
@@ -344,26 +327,25 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
-def _label_for_port(port: int) -> str:
+def _proxy_host(plan: SmolmachinesBottlePlan, loopback_ip: str) -> str:
-    if port == _EGRESS_PORT:
+    """Return the host IP for TSI's allowlist and docker port-forward bindings.
-        return "egress"
-    if port == _GIT_HTTP_PORT:
-        return "git-http"
-    if port == _SUPERVISE_PORT:
-        return "supervise"
-    return f"port-{port}"
 
+    On macOS, the per-bottle loopback alias (e.g. ``127.0.0.16``) works
+    because macOS's network stack lets TSI intercept 127.x.x.x connects
+    from the guest before they reach the host's own loopback.
 
-def _port_for_label(label: str) -> int:
+    On Linux, the guest kernel's LOCAL routing table routes all
-    if label == "egress":
+    ``127.0.0.0/8`` to the guest's own loopback — those packets never
-        return _EGRESS_PORT
+    reach eth0 and TSI never sees them. Using the per-bottle bridge
-    if label == "git-http":
+    gateway (e.g. ``192.168.N.1``) instead sidesteps the problem: it
-        return _GIT_HTTP_PORT
+    is not a loopback address, so the guest routes it via eth0 and TSI
-    if label == "supervise":
+    intercepts it normally. The TSI allowlist is ``gateway/32``, which
-        return _SUPERVISE_PORT
+    is distinct from the container IP (``192.168.N.2``), so the agent
-    if label.startswith("port-"):
+    still can't reach the egress daemon directly — TSI blocks any
-        return int(label.removeprefix("port-"))
+    connection to the container IP that isn't via the published port."""
-    raise ValueError(f"unknown sidecar forward label: {label}")
+    if platform.system() == "Linux":
+        return plan.bundle_gateway
+    return loopback_ip
 
 
 def _bundle_launch_spec(
@@ -471,13 +453,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
@@ -488,35 +465,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/port_forward.py

@@ -1,245 +0,0 @@
-"""Per-bottle TCP forwarder for smolmachines sidecar VMs.
-
-smolVM currently publishes guest ports on host loopback as
-``HOST:GUEST`` port pairs, without an address-scoped bind. The agent
-VM's TSI allowlist is IP-only, so bot-bottle exposes a second, stricter
-surface: listeners bound only to the bottle's allocated host address,
-forwarding to the raw smolVM-published loopback ports.
-"""
-
-from __future__ import annotations
-
-import argparse
-import json
-import os
-import selectors
-import signal
-import socket
-import subprocess
-import sys
-import threading
-from dataclasses import asdict, dataclass
-from typing import Any, Iterable, Sequence
-
-from ...log import die, warn
-
-
-_BUFFER_SIZE = 64 * 1024
-_LOOPBACK_TARGETS = {"127.0.0.1", "::1"}
-_FORBIDDEN_LISTEN_HOSTS = {"", "0.0.0.0", "::", "127.0.0.1", "::1"}
-
-
-@dataclass(frozen=True)
-class ForwardSpec:
-    label: str
-    listen_host: str
-    listen_port: int
-    target_host: str
-    target_port: int
-
-
-@dataclass(frozen=True)
-class ForwarderHandle:
-    process: Any
-    forwards: tuple[ForwardSpec, ...]
-
-
-def start_forwarder(specs: Sequence[ForwardSpec]) -> ForwarderHandle:
-    """Start one helper process for a bottle and wait until all
-    listeners are bound. ``listen_port`` may be 0; the returned specs
-    contain the kernel-assigned ports printed by the helper."""
-    if not specs:
-        return ForwarderHandle(process=_CompletedProcess(), forwards=())
-    _validate_specs(specs)
-    argv = [
-        sys.executable,
-        "-c",
-        "from bot_bottle.backend.smolmachines.port_forward import main; "
-        "raise SystemExit(main())",
-        "--spec-json",
-        json.dumps([asdict(s) for s in specs], separators=(",", ":")),
-    ]
-    proc = subprocess.Popen(
-        argv,
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.PIPE,
-        stderr=None,
-        text=True,
-        env=os.environ,
-    )
-    assert proc.stdout is not None
-    line = proc.stdout.readline()
-    proc.stdout.close()
-    if not line:
-        proc.terminate()
-        proc.wait(timeout=5)
-        die("smolmachines forwarder failed before reporting readiness")
-    try:
-        payload = json.loads(line)
-        bound = tuple(ForwardSpec(**item) for item in payload["forwards"])
-    except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
-        proc.terminate()
-        proc.wait(timeout=5)
-        die(f"smolmachines forwarder returned invalid readiness payload: {exc}")
-    return ForwarderHandle(process=proc, forwards=bound)
-
-
-def stop_forwarder(handle: ForwarderHandle) -> None:
-    proc = handle.process
-    if proc.poll() is not None:
-        return
-    proc.terminate()
-    try:
-        proc.wait(timeout=5)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait(timeout=5)
-
-
-def _validate_specs(specs: Iterable[ForwardSpec]) -> None:
-    for spec in specs:
-        if spec.listen_host in _FORBIDDEN_LISTEN_HOSTS:
-            die(f"refusing unsafe smolmachines forwarder bind: {spec.listen_host!r}")
-        if spec.target_host not in _LOOPBACK_TARGETS:
-            die(f"refusing non-loopback smolmachines forwarder target: {spec.target_host!r}")
-        if not 0 <= spec.listen_port <= 65535:
-            die(f"invalid smolmachines forwarder listen port: {spec.listen_port}")
-        if not 1 <= spec.target_port <= 65535:
-            die(f"invalid smolmachines forwarder target port: {spec.target_port}")
-
-
-class _CompletedProcess:
-    """Popen-like no-op used when a bottle has no forwards."""
-
-    def poll(self) -> int:
-        return 0
-
-    def terminate(self) -> None:
-        return None
-
-    def kill(self) -> None:
-        return None
-
-    def wait(self, timeout: float | None = None) -> int:
-        del timeout
-        return 0
-
-
-class _Forwarder:
-    def __init__(self, specs: Sequence[ForwardSpec]):
-        _validate_specs(specs)
-        self.specs = tuple(specs)
-        self.selector = selectors.DefaultSelector()
-        self.stop_event = threading.Event()
-        self.listeners: list[tuple[socket.socket, ForwardSpec]] = []
-        self.bound: list[ForwardSpec] = []
-
-    def start(self) -> None:
-        for spec in self.specs:
-            listener = socket.create_server(
-                (spec.listen_host, spec.listen_port),
-                reuse_port=False,
-                backlog=64,
-            )
-            listener.setblocking(False)
-            host, port = listener.getsockname()[:2]
-            bound = ForwardSpec(
-                label=spec.label,
-                listen_host=str(host),
-                listen_port=int(port),
-                target_host=spec.target_host,
-                target_port=spec.target_port,
-            )
-            self.listeners.append((listener, bound))
-            self.bound.append(bound)
-            self.selector.register(listener, selectors.EVENT_READ, bound)
-
-    def serve(self) -> None:
-        while not self.stop_event.is_set():
-            try:
-                events = self.selector.select(timeout=0.25)
-            except OSError:
-                return
-            for key, _ in events:
-                listener = key.fileobj
-                if not isinstance(listener, socket.socket):
-                    continue
-                spec = key.data
-                try:
-                    client, _ = listener.accept()
-                except OSError:
-                    continue
-                threading.Thread(
-                    target=self._handle_client,
-                    args=(client, spec),
-                    daemon=True,
-                ).start()
-
-    def stop(self) -> None:
-        self.stop_event.set()
-        for listener, _ in self.listeners:
-            try:
-                self.selector.unregister(listener)
-            except Exception:  # noqa: BLE001 - best effort shutdown
-                pass
-            listener.close()
-        self.selector.close()
-
-    def _handle_client(self, client: socket.socket, spec: ForwardSpec) -> None:
-        with client:
-            try:
-                target = socket.create_connection((spec.target_host, spec.target_port))
-            except OSError as exc:
-                warn(f"smolmachines forwarder {spec.label}: target connect failed: {exc}")
-                return
-            with target:
-                left = threading.Thread(target=_pipe, args=(client, target), daemon=True)
-                right = threading.Thread(target=_pipe, args=(target, client), daemon=True)
-                left.start()
-                right.start()
-                left.join()
-                right.join()
-
-
-def _pipe(src: socket.socket, dst: socket.socket) -> None:
-    while True:
-        try:
-            data = src.recv(_BUFFER_SIZE)
-        except OSError:
-            break
-        if not data:
-            break
-        try:
-            dst.sendall(data)
-        except OSError:
-            break
-    try:
-        dst.shutdown(socket.SHUT_WR)
-    except OSError:
-        pass
-
-
-def main(argv: Sequence[str] | None = None) -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--spec-json", required=True)
-    ns = parser.parse_args(argv)
-    specs = tuple(ForwardSpec(**item) for item in json.loads(ns.spec_json))
-    forwarder = _Forwarder(specs)
-
-    def request_stop(signum: int, _frame: object) -> None:
-        del signum
-        forwarder.stop()
-
-    signal.signal(signal.SIGTERM, request_stop)
-    signal.signal(signal.SIGINT, request_stop)
-    forwarder.start()
-    print(json.dumps({"forwards": [asdict(s) for s in forwarder.bound]}), flush=True)
-    try:
-        forwarder.serve()
-    finally:
-        forwarder.stop()
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -23,8 +23,6 @@ Plans (EgressPlan, …) lands in chunk 2d."""
 
 from __future__ import annotations
 
-import os
-import socket
 import subprocess
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -36,7 +34,6 @@ from ..docker.sidecar_bundle import (
     SIDECAR_BUNDLE_DOCKERFILE,
     SIDECAR_BUNDLE_IMAGE,
 )
-from . import smolvm as _smolvm
 
 
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
@@ -57,11 +54,6 @@ def bundle_container_name(slug: str) -> str:
     return f"bot-bottle-sidecars-{slug}"
 
 
-def bundle_machine_name(slug: str) -> str:
-    """Sidecar smolVM machine name."""
-    return bundle_container_name(slug)
-
-
 @dataclass(frozen=True)
 class BundleLaunchSpec:
     """Everything `start_bundle` needs to bring up one bundle
@@ -100,11 +92,6 @@ class BundleLaunchSpec:
     publish_host_ip: str = "127.0.0.1"
 
 
-@dataclass(frozen=True)
-class BundleVmLaunch:
-    raw_ports: dict[int, int]
-
-
 def ensure_bundle_image(image: str = SIDECAR_BUNDLE_IMAGE) -> None:
     """Build the sidecar bundle image before `docker run`.
 
@@ -120,70 +107,6 @@ def ensure_bundle_image(image: str = SIDECAR_BUNDLE_IMAGE) -> None:
     )
 
 
-def allocate_raw_host_ports(container_ports: Sequence[int]) -> dict[int, int]:
-    """Reserve candidate host loopback ports for smolVM `-p HOST:GUEST`.
-
-    The sockets are closed before smolVM binds them, so this remains
-    best-effort. smolVM failing to bind a selected port is fatal in
-    the caller's launch path."""
-    out: dict[int, int] = {}
-    for container_port in container_ports:
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.bind(("127.0.0.1", 0))
-            out[container_port] = int(sock.getsockname()[1])
-    return out
-
-
-def start_bundle_vm(
-    spec: BundleLaunchSpec,
-    *,
-    from_path: Path,
-    host_env: dict[str, str] | None = None,
-) -> BundleVmLaunch:
-    """Create and start the sidecar bundle as a smolVM.
-
-    smolVM's own published ports are raw host-loopback ports. The
-    launch flow wraps them with per-bottle address-bound forwarders
-    before exposing anything to the agent VM."""
-    raw_ports = allocate_raw_host_ports(spec.ports_to_publish)
-    effective_host_env = host_env if host_env is not None else os.environ
-    env: dict[str, str] = {"BOT_BOTTLE_SIDECAR_DAEMONS": spec.daemons_csv}
-    for entry in spec.environment:
-        name, sep, value = entry.partition("=")
-        if sep:
-            env[name] = value
-        elif entry in effective_host_env:
-            env[entry] = effective_host_env[entry]
-    name = bundle_machine_name(spec.slug)
-    _smolvm.machine_create(
-        name,
-        from_path=from_path,
-        net=True,
-        env=env,
-        volumes=spec.volumes,
-        ports=tuple(
-            (host_port, guest_port)
-            for guest_port, host_port in raw_ports.items()
-        ),
-    )
-    _smolvm.machine_start(name)
-    _smolvm.wait_exec_ready(name)
-    return BundleVmLaunch(raw_ports=raw_ports)
-
-
-def stop_bundle_vm(slug: str) -> None:
-    """Best-effort sidecar VM teardown."""
-    name = bundle_machine_name(slug)
-    try:
-        _smolvm.machine_stop(name)
-    except _smolvm.SmolvmError as exc:
-        warn(f"smolvm machine stop {name} failed: {exc}")
-    try:
-        _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

bot_bottle/backend/smolmachines/smolvm.py

@@ -113,11 +113,8 @@ def machine_create(
     *,
     image: str | None = None,
     from_path: Path | None = None,
-    net: bool = False,
     allow_cidrs: Sequence[str] = (),
     env: Mapping[str, str] | None = None,
-    ports: Sequence[tuple[int, int]] = (),
-    volumes: Sequence[tuple[str, str, bool]] = (),
 ) -> None:
     """`smolvm machine create --name NAME [--image IMG | --from PATH]
     [--allow-cidr CIDR ...] [-e K=V ...]`. NAME is passed as
@@ -134,24 +131,19 @@ def machine_create(
     no-pull-at-start property. The flag form gives the same
     result without the Smolfile complication.
 
-    `--net` is sent explicitly when `net=True` or `allow_cidrs` is
+    `--net` is sent explicitly when `allow_cidrs` is non-empty.
-    non-empty. `--allow-cidr` implies `--net` per the CLI help, but
+    `--allow-cidr` implies `--net` per the CLI help, but sending
-    sending `--net` explicitly is harmless and ensures the guest has
+    `--net` explicitly is harmless and ensures the guest has
     network access even if that implication changes across versions."""
     args: list[str] = ["machine", "create", "--name", name]
     if image is not None:
         args += ["--image", image]
     if from_path is not None:
         args += ["--from", str(from_path)]
-    if net or allow_cidrs:
+    if allow_cidrs:
         args.append("--net")
     for cidr in allow_cidrs:
         args += ["--allow-cidr", cidr]
-    for host_port, guest_port in ports:
-        args += ["-p", f"{host_port}:{guest_port}"]
-    for host_path, guest_path, read_only in volumes:
-        suffix = ":ro" if read_only else ""
-        args += ["-v", f"{host_path}:{guest_path}{suffix}"]
     if env:
         for k, v in env.items():
             args += ["-e", f"{k}={v}"]

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,8 +209,6 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
-        headless=True,
-        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -406,13 +389,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/git_gate_host_key.py

@@ -1,268 +0,0 @@
-"""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,11 +21,6 @@ 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
@@ -581,105 +576,3 @@ 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/prds/prd-new-smolmachines-linux.md

@@ -1,6 +1,6 @@
-# PRD 0068: smolmachines backend on Linux
+# PRD prd-new: smolmachines backend on Linux
 
-- **Status:** Active
+- **Status:** Draft
 - **Author:** Claude
 - **Created:** 2026-06-25
 - **Issue:** #283

docs/prds/prd-new-smolmachines-sidecar-vm.md

@@ -1,115 +0,0 @@
-# PRD prd-new: smolmachines sidecar VM
-
-- **Status:** Active
-- **Author:** codex
-- **Created:** 2026-07-09
-- **Issue:** #332
-
-## Summary
-
-Run the smolmachines backend's trusted sidecar bundle as its own smolVM instead
-of a Docker container. A bottle then consists of an agent VM plus a sidecar VM,
-with the agent VM's TSI allowlist limited to the per-bottle agent-facing
-sidecar surface.
-
-## Problem
-
-The smolmachines backend currently runs the agent in smolVM but keeps
-egress/git-gate/supervise in a Docker sidecar bundle. That hybrid launch path
-keeps Docker in the trusted runtime path and leaves smolmachines with a
-different isolation boundary than its agent VM design suggests.
-
-The existing Docker sidecar path also uses Docker port publishing to bind only
-agent-facing services to the per-bottle host address. TSI is IP-only, so the
-smolVM replacement must preserve that scoping: publishing sidecar services on
-generic host localhost would let the agent reach unrelated host services or
-other bottle sidecars if those services share the allowed address.
-
-## Goals / Success Criteria
-
-1. `BOT_BOTTLE_BACKEND=smolmachines ./cli.py start <agent>` runs with both the
-   agent and sidecar as smolVMs.
-2. The smolmachines launch path no longer uses `docker run` for the sidecar
-   bundle.
-3. Agent-facing egress, git-gate, and supervise endpoints are exposed only
-   through the bottle's own published sidecar surface.
-4. Internal sidecar-only ports are not reachable from the agent VM.
-5. The agent VM TSI allowlist remains fail-closed and limited to the per-bottle
-   address.
-6. Sidecar config, secrets, and state are delivered to the sidecar VM without
-   exposing provider or forge credentials to the agent VM.
-7. Teardown removes both VMs and any host-side published port state.
-
-## Non-goals
-
-- Rewriting the egress, git-gate, or supervise daemons.
-- Changing the Docker backend's sidecar bundle behavior.
-- Replacing the existing agent image build and pack pipeline except where shared
-  helper extraction is needed for sidecar image packing.
-- Weakening the current smolmachines TSI allowlist checks.
-
-## Design
-
-The sidecar VM continues to use the existing sidecar bundle image and
-`/app/sidecar_init.py` supervisor. The launch flow builds the sidecar image,
-packs it into a `.smolmachine` artifact, creates a per-bottle sidecar VM from
-that artifact, passes the same daemon-selection environment the Docker bundle
-uses today, mounts or copies the same daemon-private config/state inputs, starts
-the sidecar VM, wraps the sidecar VM's raw smolVM-published ports with a local
-per-bottle forwarder, then starts the agent VM with
-`--allow-cidr <per-bottle-address>/32`.
-
-smolVM currently exposes guest ports as `HOST:GUEST` port pairs bound on host
-loopback. It does not expose an address-scoped bind like
-`127.0.0.16:<port>:9099`. Because TSI is IP-only, bot-bottle must not advertise
-those raw loopback ports directly to the agent. Instead, bot-bottle runs one
-small stdlib Python TCP forwarder process per bottle:
-
-```text
-agent VM
-  -> TSI allows only <bottle-address>/32
-  -> bot-bottle forwarder bound to <bottle-address>:<random-port>
-  -> raw smolVM-published sidecar port on 127.0.0.1:<raw-port>
-  -> sidecar VM service
-```
-
-Agent-facing URLs are stamped from the forwarder ports, never the raw smolVM
-ports:
-
-- `HTTP_PROXY` / `HTTPS_PROXY` point at the published egress proxy port.
-- `GIT_GATE_URL` points at the published git HTTP port when git-gate is enabled.
-- `MCP_SUPERVISE_URL` points at the published supervise port when supervise is
-  enabled.
-
-Internal sidecar-only services stay bound inside the sidecar VM and are not
-published.
-
-### Forwarder constraints
-
-- Reject wildcard listener binds (`0.0.0.0`, `::`) and generic localhost
-  listener binds (`127.0.0.1`, `::1`) for agent-facing forwards.
-- Accept only fixed loopback targets selected by launch; the client cannot
-  choose or influence the forwarding target.
-- Forward bytes transparently without parsing HTTP, CONNECT, TLS, Git, or MCP.
-- Log lifecycle and endpoint metadata only; never log payload bytes.
-- Create forwards only for egress, git HTTP when enabled, and supervise when
-  enabled.
-- Fail closed if any listener cannot bind exactly the requested per-bottle
-  address.
-
-## Implementation chunks
-
-1. Extract the existing image-to-smolmachine pack helper so agent and sidecar
-   artifacts share the same cache and registry path.
-2. Add a strict stdlib host TCP forwarder for per-bottle address-bound
-   forwarding to raw smolVM-published sidecar ports.
-3. Add a sidecar VM launch spec and lifecycle helpers: pack, create, start,
-   publish/discover ports, stop, delete.
-4. Switch smolmachines launch from Docker sidecar bundle lifecycle to sidecar VM
-   plus forwarder lifecycle.
-5. Update egress apply / reload paths for the sidecar VM supervisor.
-6. Add unit tests for argv shape, URL stamping, teardown, and fail-closed
-   behavior.
-7. Add or update integration coverage for agent-to-sidecar reachability, host
-   localhost denial, other-bottle alias denial, internal port denial, and
-   teardown cleanup.

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.411
+pyright>=1.1.300
 coverage>=7.0.0

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,
@@ -267,8 +265,7 @@ class TestAgentAlwaysPresent(unittest.TestCase):
     def test_agent_depends_only_on_sidecars(self):
         # Bundle shape: the init supervisor owns intra-bundle daemon
         # ordering, so the agent waits on the bundle container alone.
-        cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
+        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
-        for kwargs in cases:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
@@ -303,11 +300,6 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
         self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
 
-    def test_cached_policy_omits_bundle_build(self):
-        sc = self._render(image_policy="cached")["services"]["sidecars"]
-        self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
-        self.assertNotIn("build", sc)
-
     def test_bundle_container_name_uses_sidecars_prefix(self):
         sc = self._render()["services"]["sidecars"]
         self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])

tests/unit/test_config_store.py

@@ -1,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

@@ -1,698 +0,0 @@
-"""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

@@ -1,58 +0,0 @@
-from __future__ import annotations
-
-import unittest
-from unittest.mock import patch
-
-from bot_bottle.backend.egress_apply import EgressApplyError
-from bot_bottle.backend.smolmachines import egress_apply
-from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
-
-
-class TestFetchCurrentRoutes(unittest.TestCase):
-    def test_reads_routes_from_sidecar_vm(self):
-        with patch.object(
-            egress_apply._smolvm,
-            "machine_exec",
-            return_value=SmolvmRunResult(0, "routes", ""),
-        ) as exec_:
-            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"],
-        )
-
-    def test_read_failure_raises_apply_error(self):
-        with patch.object(
-            egress_apply._smolvm,
-            "machine_exec",
-            return_value=SmolvmRunResult(1, "", "nope"),
-        ):
-            with self.assertRaises(EgressApplyError):
-                egress_apply.fetch_current_routes("dev-abc")
-
-
-class TestReload(unittest.TestCase):
-    def test_sends_hup_to_sidecar_init(self):
-        with patch.object(
-            egress_apply._smolvm,
-            "machine_exec",
-            return_value=SmolvmRunResult(0, "", ""),
-        ) as exec_:
-            egress_apply.SmolmachinesEgressApplicator()._signal_bundle_reload("dev-abc")
-        exec_.assert_called_once_with(
-            "bot-bottle-sidecars-dev-abc",
-            ["sh", "-c", "kill -HUP 1"],
-        )
-
-    def test_reload_failure_raises_apply_error(self):
-        with patch.object(
-            egress_apply._smolvm,
-            "machine_exec",
-            return_value=SmolvmRunResult(1, "", "nope"),
-        ):
-            with self.assertRaises(EgressApplyError):
-                egress_apply.SmolmachinesEgressApplicator()._signal_bundle_reload("dev-abc")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_smolmachines_enumerate.py

@@ -1,66 +0,0 @@
-"""Unit: smolmachines active-agent enumeration."""
-
-from __future__ import annotations
-
-import json
-import subprocess
-import unittest
-from unittest.mock import patch
-
-from bot_bottle.backend.smolmachines import enumerate as _enumerate
-
-
-def _completed(
-    stdout: str,
-    returncode: int = 0,
-) -> subprocess.CompletedProcess:  # type: ignore[type-arg]
-    return subprocess.CompletedProcess(
-        args=[],
-        returncode=returncode,
-        stdout=stdout,
-        stderr="",
-    )
-
-
-class TestEnumerateActive(unittest.TestCase):
-    def test_skips_sidecar_vms(self):
-        machines = [
-            {"name": "bot-bottle-demo-abc12", "state": "running"},
-            {"name": "bot-bottle-sidecars-demo-abc12", "state": "running"},
-            {"name": "bot-bottle-stopped-xyz", "state": "stopped"},
-            {"name": "other", "state": "running"},
-        ]
-
-        class _Metadata:
-            agent_name = "demo"
-            started_at = "2026-07-09T00:00:00Z"
-            label = "Demo"
-            color = "green"
-
-        with patch(
-            "bot_bottle.backend.smolmachines.enumerate.subprocess.run",
-            return_value=_completed(json.dumps(machines)),
-        ), patch(
-            "bot_bottle.backend.smolmachines.enumerate._query_bundle_services",
-            return_value={"demo-abc12": ("egress",)},
-        ), patch(
-            "bot_bottle.backend.smolmachines.enumerate.read_metadata",
-            return_value=_Metadata(),
-        ) as read_metadata:
-            active = _enumerate.enumerate_active()
-
-        self.assertEqual(1, len(active))
-        self.assertEqual("demo-abc12", active[0].slug)
-        self.assertEqual(("egress",), active[0].services)
-        read_metadata.assert_called_once_with("demo-abc12")
-
-    def test_invalid_smolvm_output_returns_empty(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.enumerate.subprocess.run",
-            return_value=_completed("not json"),
-        ):
-            self.assertEqual([], _enumerate.enumerate_active())
-
-
-if __name__ == "__main__":
-    unittest.main()

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

@@ -1,325 +0,0 @@
-from __future__ import annotations
-
-import json
-import socket
-import subprocess
-import threading
-import unittest
-from unittest.mock import MagicMock, patch
-
-from bot_bottle.backend.smolmachines.port_forward import (
-    ForwardSpec,
-    ForwarderHandle,
-    _CompletedProcess,
-    _Forwarder,
-    _pipe,
-    main,
-    start_forwarder,
-    stop_forwarder,
-)
-
-
-def _echo_server() -> tuple[socket.socket, int]:
-    listener = socket.create_server(("127.0.0.1", 0))
-    port = int(listener.getsockname()[1])
-
-    def run() -> None:
-        with listener:
-            conn, _ = listener.accept()
-            with conn:
-                while True:
-                    data = conn.recv(65536)
-                    if not data:
-                        return
-                    conn.sendall(data)
-
-    threading.Thread(target=run, daemon=True).start()
-    return listener, port
-
-
-class TestForwarderValidation(unittest.TestCase):
-    def test_empty_specs_returns_noop_handle(self):
-        handle = start_forwarder(())
-
-        self.assertEqual((), handle.forwards)
-        self.assertEqual(0, handle.process.poll())
-        stop_forwarder(handle)
-
-    def test_rejects_generic_localhost_listener(self):
-        with self.assertRaises(SystemExit):
-            start_forwarder((
-                ForwardSpec(
-                    label="egress",
-                    listen_host="127.0.0.1",
-                    listen_port=0,
-                    target_host="127.0.0.1",
-                    target_port=9099,
-                ),
-            ))
-
-    def test_rejects_wildcard_listener(self):
-        with self.assertRaises(SystemExit):
-            start_forwarder((
-                ForwardSpec(
-                    label="egress",
-                    listen_host="0.0.0.0",
-                    listen_port=0,
-                    target_host="127.0.0.1",
-                    target_port=9099,
-                ),
-            ))
-
-    def test_rejects_non_loopback_target(self):
-        with self.assertRaises(SystemExit):
-            start_forwarder((
-                ForwardSpec(
-                    label="egress",
-                    listen_host="127.0.0.16",
-                    listen_port=0,
-                    target_host="192.0.2.10",
-                    target_port=9099,
-                ),
-            ))
-
-    def test_rejects_invalid_ports(self):
-        with self.assertRaises(SystemExit):
-            start_forwarder((
-                ForwardSpec("bad-listen", "127.0.0.2", 70000, "127.0.0.1", 9099),
-            ))
-        with self.assertRaises(SystemExit):
-            start_forwarder((
-                ForwardSpec("bad-target", "127.0.0.2", 0, "127.0.0.1", 0),
-            ))
-
-    def test_completed_process_methods_are_noops(self):
-        proc = _CompletedProcess()
-
-        proc.terminate()
-        proc.kill()
-
-        self.assertEqual(0, proc.poll())
-        self.assertEqual(0, proc.wait(timeout=1))
-
-
-class _FakeStdout:
-    def __init__(self, line: str):
-        self.line = line
-        self.closed = False
-
-    def readline(self) -> str:
-        return self.line
-
-    def close(self) -> None:
-        self.closed = True
-
-
-class TestForwarderProcess(unittest.TestCase):
-    def test_start_parses_ready_payload(self):
-        payload = {
-            "forwards": [{
-                "label": "egress",
-                "listen_host": "127.0.0.2",
-                "listen_port": 61000,
-                "target_host": "127.0.0.1",
-                "target_port": 55000,
-            }],
-        }
-        proc = MagicMock()
-        proc.stdout = _FakeStdout(json.dumps(payload) + "\n")
-
-        with patch(
-            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
-            return_value=proc,
-        ) as popen:
-            handle = start_forwarder((
-                ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
-            ))
-
-        self.assertIs(proc, handle.process)
-        self.assertEqual(61000, handle.forwards[0].listen_port)
-        self.assertTrue(proc.stdout.closed)
-        argv = popen.call_args.args[0]
-        self.assertIn("--spec-json", argv)
-
-    def test_start_dies_when_helper_exits_before_ready(self):
-        proc = MagicMock()
-        proc.stdout = _FakeStdout("")
-
-        with patch(
-            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
-            return_value=proc,
-        ):
-            with self.assertRaises(SystemExit):
-                start_forwarder((
-                    ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
-                ))
-
-        proc.terminate.assert_called_once()
-        proc.wait.assert_called_once_with(timeout=5)
-
-    def test_start_dies_on_invalid_ready_payload(self):
-        proc = MagicMock()
-        proc.stdout = _FakeStdout('{"wrong": []}\n')
-
-        with patch(
-            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
-            return_value=proc,
-        ):
-            with self.assertRaises(SystemExit):
-                start_forwarder((
-                    ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
-                ))
-
-        proc.terminate.assert_called_once()
-        proc.wait.assert_called_once_with(timeout=5)
-
-    def test_stop_kills_process_after_timeout(self):
-        proc = MagicMock()
-        proc.poll.return_value = None
-        proc.wait.side_effect = [subprocess.TimeoutExpired("cmd", 5), 0]
-
-        stop_forwarder(ForwarderHandle(process=proc, forwards=()))
-
-        proc.terminate.assert_called_once()
-        proc.kill.assert_called_once()
-        self.assertEqual(2, proc.wait.call_count)
-
-
-class TestForwarding(unittest.TestCase):
-    def test_forwards_bytes_both_directions(self):
-        try:
-            probe = socket.create_server(("127.0.0.2", 0))
-        except OSError:
-            self.skipTest("127.0.0.2 is not bindable on this host")
-        else:
-            probe.close()
-        _listener, target_port = _echo_server()
-        handle = start_forwarder((
-            ForwardSpec(
-                label="egress",
-                listen_host="127.0.0.2",
-                listen_port=0,
-                target_host="127.0.0.1",
-                target_port=target_port,
-            ),
-        ))
-        try:
-            bound = handle.forwards[0]
-            with socket.create_connection((bound.listen_host, bound.listen_port)) as sock:
-                sock.sendall(b"hello")
-                self.assertEqual(b"hello", sock.recv(5))
-        finally:
-            stop_forwarder(handle)
-
-
-class TestForwarderInProcess(unittest.TestCase):
-    def test_forwarder_serves_and_stops_in_process(self):
-        _listener, target_port = _echo_server()
-        forwarder = _Forwarder((
-            ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", target_port),
-        ))
-        forwarder.start()
-        thread = threading.Thread(target=forwarder.serve, daemon=True)
-        thread.start()
-        try:
-            bound = forwarder.bound[0]
-            with socket.create_connection((bound.listen_host, bound.listen_port)) as sock:
-                sock.sendall(b"ping")
-                self.assertEqual(b"ping", sock.recv(4))
-        finally:
-            forwarder.stop()
-            thread.join(timeout=2)
-
-        self.assertFalse(thread.is_alive())
-
-    def test_forwarder_logs_target_connect_failure(self):
-        forwarder = _Forwarder((
-            ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 1),
-        ))
-        forwarder.start()
-        thread = threading.Thread(target=forwarder.serve, daemon=True)
-        thread.start()
-        try:
-            bound = forwarder.bound[0]
-            with patch("bot_bottle.backend.smolmachines.port_forward.warn") as warn:
-                with socket.create_connection((bound.listen_host, bound.listen_port)):
-                    pass
-                for _ in range(100):
-                    if warn.called:
-                        break
-                    threading.Event().wait(0.01)
-            self.assertTrue(warn.called)
-        finally:
-            forwarder.stop()
-            thread.join(timeout=2)
-
-    def test_pipe_stops_when_recv_fails(self):
-        src, dst = socket.socketpair()
-        src.close()
-        try:
-            _pipe(src, dst)
-        finally:
-            dst.close()
-
-    def test_pipe_stops_when_send_fails(self):
-        src, src_peer = socket.socketpair()
-        _dst_peer, dst = socket.socketpair()
-        dst.close()
-        try:
-            src_peer.sendall(b"hello")
-            src_peer.shutdown(socket.SHUT_WR)
-            _pipe(src, dst)
-        finally:
-            src.close()
-            src_peer.close()
-            _dst_peer.close()
-
-
-class TestMain(unittest.TestCase):
-    def test_main_starts_serves_and_stops_forwarder(self):
-        calls: list[str] = []
-        handlers = []
-
-        class FakeForwarder:
-            def __init__(self, specs):  # type: ignore[no-untyped-def]
-                self.specs = specs
-                self.bound = specs
-
-            def start(self) -> None:
-                calls.append("start")
-
-            def serve(self) -> None:
-                calls.append("serve")
-
-            def stop(self) -> None:
-                calls.append("stop")
-
-        def record_signal(_signum, handler):  # type: ignore[no-untyped-def]
-            handlers.append(handler)
-
-        spec = ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 9099)
-        with patch(
-            "bot_bottle.backend.smolmachines.port_forward._Forwarder",
-            FakeForwarder,
-        ), patch(
-            "bot_bottle.backend.smolmachines.port_forward.signal.signal",
-            side_effect=record_signal,
-        ), patch("builtins.print") as print_mock:
-            self.assertEqual(
-                0,
-                main(("--spec-json", json.dumps([{
-                    "label": spec.label,
-                    "listen_host": spec.listen_host,
-                    "listen_port": spec.listen_port,
-                    "target_host": spec.target_host,
-                    "target_port": spec.target_port,
-                }]))),
-            )
-            handlers[0](15, None)
-
-        self.assertEqual(["start", "serve", "stop", "stop"], calls)
-        self.assertIn("forwards", print_mock.call_args.args[0])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_smolmachines_provision.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 import subprocess
 import tempfile
 import unittest
-from contextlib import ExitStack
 from dataclasses import replace
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -28,7 +27,6 @@ from bot_bottle.backend.smolmachines.bottle_plan import (
     SmolmachinesBottlePlan,
 )
 from bot_bottle.backend.smolmachines import launch as _launch
-from bot_bottle.backend.smolmachines import port_forward as _forward
 from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
 from bot_bottle.backend.util import AGENT_CA_PATH
 from bot_bottle.egress import EgressPlan, EgressRoute
@@ -434,7 +432,12 @@ class TestBundleLaunchSpec(unittest.TestCase):
 
     def test_canary_env_visible_to_smolvm_guest(self):
         plan = _plan(canary=True)
-        stamped = _launch._discover_urls(plan, "127.0.0.16", {9099: 65000})
+        with patch.object(
+            _launch._bundle,
+            "bundle_host_port",
+            return_value="65000",
+        ):
+            stamped = _launch._discover_urls(plan, "127.0.0.16")
 
         self.assertEqual(
             "fake-canary-value",
@@ -503,111 +506,23 @@ class TestProvisionGitUser(unittest.TestCase):
         self.assertIn("bot@example.com", calls[0][0])
 
 
-class TestLaunchResourceWiring(unittest.TestCase):
+class TestProxyHost(unittest.TestCase):
-    def test_allocate_resources_uses_loopback_alias_and_bundle_name(self):
+    """_proxy_host returns the bridge gateway on Linux and the loopback
+    alias on other platforms."""
+
+    def test_linux_returns_bundle_gateway(self):
         plan = _plan()
-        with patch(
+        with patch("bot_bottle.backend.smolmachines.launch.platform.system",
-            "bot_bottle.backend.smolmachines.launch._loopback.ensure_pool",
+                   return_value="Linux"):
-        ) as ensure_pool, patch(
+            result = _launch._proxy_host(plan, "127.0.0.16")
-            "bot_bottle.backend.smolmachines.launch._loopback.allocate",
+        self.assertEqual(plan.bundle_gateway, result)
-            return_value="127.0.0.16",
-        ) as allocate:
-            loopback_ip, network = _launch._allocate_resources(plan, ExitStack())
 
-        ensure_pool.assert_called_once_with()
+    def test_non_linux_returns_loopback(self):
-        allocate.assert_called_once_with("demo-abc12")
-        self.assertEqual("127.0.0.16", loopback_ip)
-        self.assertEqual("bot-bottle-bundle-demo-abc12", network)
-
-    def test_start_bundle_wraps_raw_vm_ports_with_forwarder_ports(self):
-        plan = _plan(supervise=True)
-        plan = replace(
-            plan,
-            git_gate_plan=replace(
-                plan.git_gate_plan,
-                upstreams=(GitGateUpstream(
-                    name="bot-bottle",
-                    upstream_url="ssh://git@host/repo.git",
-                    upstream_host="host",
-                    upstream_port="22",
-                    identity_file="/tmp/key",
-                    known_host_key="",
-                ),),
-            ),
-        )
-        stack = MagicMock(spec=ExitStack)
-        raw_launch = MagicMock(raw_ports={9099: 55001, 9420: 55002, 9100: 55003})
-        handle = _forward.ForwarderHandle(
-            process=MagicMock(),
-            forwards=(
-                _forward.ForwardSpec("egress", "127.0.0.16", 65001, "127.0.0.1", 55001),
-                _forward.ForwardSpec("git-http", "127.0.0.16", 65002, "127.0.0.1", 55002),
-                _forward.ForwardSpec("supervise", "127.0.0.16", 65003, "127.0.0.1", 55003),
-            ),
-        )
-
-        with patch(
-            "bot_bottle.backend.smolmachines.launch._provision_git_gate_keys",
-            return_value=plan,
-        ), patch(
-            "bot_bottle.backend.smolmachines.launch._ensure_smolmachine",
-            return_value=Path("/cache/sidecar.smolmachine"),
-        ) as ensure, patch(
-            "bot_bottle.backend.smolmachines.launch._bundle.start_bundle_vm",
-            return_value=raw_launch,
-        ) as start_vm, patch(
-            "bot_bottle.backend.smolmachines.launch._forward.start_forwarder",
-            return_value=handle,
-        ) as start_forwarder:
-            stamped = _launch._start_bundle(plan, "net", "127.0.0.16", stack)
-
-        ensure.assert_called_once()
-        start_vm.assert_called_once()
-        self.assertEqual(Path("/cache/sidecar.smolmachine"), start_vm.call_args.kwargs["from_path"])
-        specs = start_forwarder.call_args.args[0]
-        self.assertEqual(
-            (
-                _forward.ForwardSpec("egress", "127.0.0.16", 0, "127.0.0.1", 55001),
-                _forward.ForwardSpec("git-http", "127.0.0.16", 0, "127.0.0.1", 55002),
-                _forward.ForwardSpec("supervise", "127.0.0.16", 0, "127.0.0.1", 55003),
-            ),
-            specs,
-        )
-        self.assertEqual("http://127.0.0.16:65001", stamped.agent_proxy_url)
-        self.assertEqual("127.0.0.16:65002", stamped.agent_git_gate_host)
-        self.assertEqual("http://127.0.0.16:65003/", stamped.agent_supervise_url)
-        self.assertEqual(2, stack.callback.call_count)
-
-    def test_init_vm_repairs_ownership_then_waits_for_exec(self):
         plan = _plan()
-        with patch(
+        with patch("bot_bottle.backend.smolmachines.launch.platform.system",
-            "bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
+                   return_value="Darwin"):
-        ) as machine_exec, patch(
+            result = _launch._proxy_host(plan, "127.0.0.16")
-            "bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
+        self.assertEqual("127.0.0.16", result)
-        ) as wait_exec_ready:
-            _launch._init_vm(plan)
-
-        machine_exec.assert_called_once()
-        argv = machine_exec.call_args.args[1]
-        self.assertEqual(["sh", "-c"], argv[:2])
-        self.assertIn("chown -R node:node /home/node", argv[2])
-        wait_exec_ready.assert_called_once_with(plan.machine_name)
-
-
-class TestPortLabels(unittest.TestCase):
-    def test_known_and_dynamic_port_labels_round_trip(self):
-        for port, label in (
-            (9099, "egress"),
-            (9420, "git-http"),
-            (9100, "supervise"),
-            (12345, "port-12345"),
-        ):
-            self.assertEqual(label, _launch._label_for_port(port))
-            self.assertEqual(port, _launch._port_for_label(label))
-
-    def test_unknown_label_is_rejected(self):
-        with self.assertRaises(ValueError):
-            _launch._port_for_label("sidecar")
 
 
 class TestDiscoverUrls(unittest.TestCase):
@@ -629,20 +544,14 @@ class TestDiscoverUrls(unittest.TestCase):
                 ),),
             ),
         )
-        stamped = _launch._discover_urls(
+        with patch.object(_launch._bundle, "bundle_host_port", return_value="9420"):
-            plan,
+            stamped = _launch._discover_urls(plan, "127.0.0.16")
-            "127.0.0.16",
+        self.assertEqual("127.0.0.16:9420", stamped.agent_git_gate_host)
-            {9099: 65000, 9420: 65001},
-        )
-        self.assertEqual("127.0.0.16:65001", stamped.agent_git_gate_host)
 
     def test_supervise_url_set_when_supervise_present(self):
         plan = _plan(supervise=True)
-        stamped = _launch._discover_urls(
+        with patch.object(_launch._bundle, "bundle_host_port", return_value="55556"):
-            plan,
+            stamped = _launch._discover_urls(plan, "127.0.0.16")
-            "127.0.0.16",
-            {9099: 65000, 9100: 55556},
-        )
         self.assertEqual("http://127.0.0.16:55556/", stamped.agent_supervise_url)
 
 

tests/unit/test_smolmachines_sidecar_bundle.py

@@ -14,16 +14,13 @@ from unittest.mock import patch
 
 from bot_bottle.backend.smolmachines.sidecar_bundle import (
     BundleLaunchSpec,
-    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,
 )
 
 
@@ -203,66 +200,6 @@ class TestEnsureBundleImage(unittest.TestCase):
         self.assertEqual("Dockerfile.sidecars", kwargs["dockerfile"])
 
 
-class TestStartBundleVm(unittest.TestCase):
-    def test_allocate_raw_host_ports_returns_bindable_ports(self):
-        raw_ports = allocate_raw_host_ports((9099, 9420))
-
-        self.assertEqual({9099, 9420}, set(raw_ports))
-        for host_port in raw_ports.values():
-            self.assertIsInstance(host_port, int)
-            self.assertGreater(host_port, 0)
-        self.assertNotEqual(raw_ports[9099], raw_ports[9420])
-
-    def test_creates_sidecar_vm_with_ports_volumes_and_env(self):
-        calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
-
-        def record(name):  # type: ignore
-            def inner(*args, **kwargs):  # type: ignore
-                calls.append((name, args, kwargs))
-            return inner
-
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.allocate_raw_host_ports",
-            return_value={9099: 55001, 9420: 55002},
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_create",
-            side_effect=record("create"),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_start",
-            side_effect=record("start"),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.wait_exec_ready",
-            side_effect=record("wait"),
-        ):
-            launch = start_bundle_vm(
-                _spec(
-                    daemons_csv="egress,git-http",
-                    environment=("FOO=bar", "TOKEN"),
-                    volumes=(("/host/routes", "/etc/egress", True),),
-                    ports_to_publish=(9099, 9420),
-                ),
-                from_path=Path("/cache/sidecar.smolmachine"),
-                host_env={"TOKEN": "secret"},
-            )
-
-        self.assertEqual({9099: 55001, 9420: 55002}, launch.raw_ports)
-        create = calls[0]
-        self.assertEqual("create", create[0])
-        self.assertEqual(("bot-bottle-sidecars-demo-abc12",), create[1])
-        self.assertEqual(Path("/cache/sidecar.smolmachine"), create[2]["from_path"])
-        self.assertTrue(create[2]["net"])
-        self.assertEqual(((55001, 9099), (55002, 9420)), create[2]["ports"])
-        self.assertEqual((("/host/routes", "/etc/egress", True),), create[2]["volumes"])
-        self.assertEqual(
-            {
-                "BOT_BOTTLE_SIDECAR_DAEMONS": "egress,git-http",
-                "FOO": "bar",
-                "TOKEN": "secret",
-            },
-            create[2]["env"],
-        )
-
-
 class TestStopBundle(unittest.TestCase):
     def _patch_run(self, **kwargs):  # type: ignore
         return patch(
@@ -285,38 +222,5 @@ class TestStopBundle(unittest.TestCase):
             stop_bundle("demo-abc12")  # no raise
 
 
-class TestStopBundleVm(unittest.TestCase):
-    def test_stops_then_deletes_sidecar_vm(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
-        ) as stop, patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
-        ) as delete:
-            stop_bundle_vm("demo-abc12")
-        stop.assert_called_once_with("bot-bottle-sidecars-demo-abc12")
-        delete.assert_called_once_with("bot-bottle-sidecars-demo-abc12")
-
-    def test_stop_and_delete_errors_are_warnings(self):
-        from bot_bottle.backend.smolmachines.smolvm import SmolvmError
-        result = subprocess.CompletedProcess(
-            args=[],
-            returncode=1,
-            stdout="",
-            stderr="failed",
-        )
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
-            side_effect=SmolvmError(["smolvm", "machine", "stop"], result),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
-            side_effect=SmolvmError(["smolvm", "machine", "delete"], result),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.warn",
-        ) as warn:
-            stop_bundle_vm("demo-abc12")
-
-        self.assertEqual(2, warn.call_count)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_smolvm.py

@@ -106,22 +106,6 @@ class TestArgvShapes(unittest.TestCase):
         self.assertIn("--name", argv)
         self.assertIn("agent-xyz", argv)
 
-    def test_machine_create_with_net_ports_and_volumes(self):
-        with self._patch_run() as m:
-            machine_create(
-                "sidecar-xyz",
-                from_path=Path("/stage/sidecar.smolmachine"),
-                net=True,
-                ports=((55000, 9099),),
-                volumes=(("/host/routes", "/etc/egress", True),),
-            )
-        argv = m.call_args.args[0]
-        self.assertIn("--net", argv)
-        self.assertIn("-p", argv)
-        self.assertIn("55000:9099", argv)
-        self.assertIn("-v", argv)
-        self.assertIn("/host/routes:/etc/egress:ro", argv)
-
     def test_machine_create_omits_net_when_no_allow_cidrs(self):
         with self._patch_run() as m:
             machine_create("agent-xyz", from_path=Path("/x.smolmachine"))

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, serialize_yaml_subset
+from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
 
 
 def _y(s: str):
@@ -457,141 +457,5 @@ 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()