refactor(smolmachines): drop obsolete docker-sidecar helpers

CI runs as root, so chmod 0o000 doesn't prevent reads. Both
test_skips_unreadable_file (TestFindRepoBottleFile) and
test_returns_false_when_file_unreadable_on_write now mock
Path.read_text to raise OSError directly.

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-83%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-84%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,6 +75,9 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
+    # True when launched via --headless (no TTY, no interactive prompts).
+    # The git-gate host-key preflight uses this to error rather than prompt.
+    headless: bool = False
 
 
 @dataclass(frozen=True)
@@ -300,6 +303,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
 
         self._preflight()
 
+        from ..git_gate_host_key import preflight_host_keys
+        manifest = preflight_host_keys(
+            manifest,
+            headless=spec.headless,
+            home_md=spec.manifest.home_md,
+        )
+
         manifest_bottle = manifest.bottle
         manifest_agent_provider = manifest_bottle.agent_provider
         agent_provider = get_provider(manifest_agent_provider.template)

bot_bottle/backend/smolmachines/bottle_plan.py

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

bot_bottle/backend/smolmachines/egress_apply.py

@@ -1,20 +1,51 @@
 """Egress apply for the smolmachines backend.
 
-The smolmachines sidecar bundle runs as a host-side Docker container,
+The smolmachines sidecar bundle runs as a sidecar smolVM. Route-file
-so egress signalling is identical to the docker backend.
+inspection and reload signalling therefore go through ``smolvm machine
+exec`` instead of Docker.
 """
 
 from __future__ import annotations
 
-from ..docker.egress_apply import (  # noqa: F401
+from ...egress import EGRESS_ROUTES_IN_CONTAINER
-    DockerEgressApplicator,
+from ...log import warn
-    EgressApplyError,
+from ..egress_apply import EgressApplicator, EgressApplyError
-    applicator,
+from . import sidecar_bundle as _bundle
-    fetch_current_routes,
+from . import smolvm as _smolvm
-)
+
+
+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__ = [
-    "DockerEgressApplicator",
+    "SmolmachinesEgressApplicator",
     "EgressApplyError",
     "applicator",
     "fetch_current_routes",

bot_bottle/backend/smolmachines/enumerate.py

@@ -31,6 +31,7 @@ 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]:
@@ -54,7 +55,11 @@ def enumerate_active() -> list[ActiveAgent]:
     for m in machines:
         name = m.get("name") or ""
         state = m.get("state") or ""
-        if state != "running" or not name.startswith(_VM_NAME_PREFIX):
+        if (
+            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

@@ -1,12 +1,9 @@
-"""End-to-end launch flow for the smolmachines backend
+"""End-to-end launch flow for the smolmachines backend.
-(PRD 0023 chunks 2d + 4b).
 
-Brings up the per-bottle docker bridge + sidecar bundle (with
+Builds the sidecar bundle smolmachine, starts it as a sidecar VM
-real daemons + their config files), creates + starts the smolvm
+with real daemons + their config files, creates + starts the agent
-guest pointed at the bundle's pinned IP via TSI's
+smolVM, yields a `SmolmachinesBottle` handle, and tears everything
-`--allow-cidr <bundle-ip>/32` allowlist, yields a
+down on context exit.
-`SmolmachinesBottle` handle, tears everything down on context
-exit.
 
 The bundle's daemons consume the inner Plans the docker backend
 already produces: egress reads routes + CAs from the EgressPlan.
@@ -17,7 +14,6 @@ from __future__ import annotations
 
 import dataclasses
 import os
-import platform
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
@@ -53,6 +49,7 @@ from ...bottle_state import (
     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
@@ -72,9 +69,8 @@ _SMOLMACHINE_CACHE_DIR = Path.home() / ".cache" / "bot-bottle" / "smolmachines"
 
 
 # Container-internal listening ports for each bundle daemon. The
-# bundle publishes each one on a random host loopback port (see
+# sidecar VM publishes each one on a random host loopback port, and
-# `_bundle.start_bundle`), and `_bundle.bundle_host_port` looks
+# the launch flow wraps those raw ports with per-bottle forwarders.
-# them up post-start.
 _GIT_HTTP_PORT = 9420
 _SUPERVISE_PORT = SUPERVISE_PORT
 
@@ -92,9 +88,8 @@ def launch(
     try:
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
-        proxy_host = _proxy_host(plan, loopback_ip)
+        proxy_host = loopback_ip
         plan = _start_bundle(plan, network, proxy_host, stack)
-        plan = _discover_urls(plan, proxy_host)
 
         agent_from_path = _agent_from_path(plan)
 
@@ -144,19 +139,17 @@ def _allocate_resources(
     plan: SmolmachinesBottlePlan,
     stack: ExitStack,
 ) -> tuple[str, str]:
-    """Reserve a loopback alias and create the per-bottle docker bridge.
+    """Reserve a per-bottle host address.
 
-    The per-bottle alias scopes TSI's allowlist to this bottle's
+    The per-bottle address scopes TSI's allowlist to this bottle's
-    published ports so the agent can't reach other bottles' or host
+    forwarder-published ports so the agent can't reach other bottles'
-    services' ports on loopback. On macOS `ensure_pool` first
+    or host services. The returned network name remains in the bundle
-    sudo-aliases the pool on `lo0`; on Linux that's a no-op since
+    spec for compatibility with older helper tests; no Docker sidecar
-    all of 127.0.0.0/8 is already loopback, but the per-bottle
+    container is launched."""
-    allocation runs on both."""
+    del stack
     _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
 
 
@@ -179,15 +172,40 @@ def _start_bundle(
     proxy_host: str,
     stack: ExitStack,
 ) -> SmolmachinesBottlePlan:
-    """Build the BundleLaunchSpec, resolve token env, start the
+    """Build the BundleLaunchSpec, start the sidecar VM, wrap its raw
-    sidecar bundle container, and register teardown."""
+    smolVM-published loopback ports with per-bottle forwarders, stamp
+    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))
-    _bundle.ensure_bundle_image(bundle_spec.image)
+    artifact = _ensure_smolmachine(
-    _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
+        bundle_spec.image,
-    stack.callback(_bundle.stop_bundle, plan.slug)
+        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-    return plan
+    )
+    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(
@@ -206,35 +224,27 @@ def _provision_git_gate_keys(
 def _discover_urls(
     plan: SmolmachinesBottlePlan,
     proxy_host: str,
+    published_ports: dict[int, int],
 ) -> SmolmachinesBottlePlan:
-    """Discover host-side ports for published container ports and
+    """Stamp URLs + guest_env from per-bottle forwarder ports.
-    return the plan with URLs + guest_env stamped in.
 
-    `proxy_host` is the host IP that both TSI's allowlist and
+    `proxy_host` is the host IP that both TSI's allowlist and the
-    docker's port-forward bindings are keyed to. On macOS it is the
+    forwarder listeners are keyed to. The raw smolVM-published ports
-    per-bottle loopback alias; on Linux it is the per-bottle bridge
+    are intentionally not advertised to the agent.
-    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 = _bundle.bundle_host_port(
+    agent_facing_host_port = published_ports[_EGRESS_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 = _bundle.bundle_host_port(
+        git_gate_host_port = published_ports[_GIT_HTTP_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 = _bundle.bundle_host_port(
+        supervise_host_port = published_ports[_SUPERVISE_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")
@@ -273,14 +283,13 @@ 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.
-    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
+    This ensures the guest can only reach sidecar forwarders published
-    ensures the guest can only reach bundle ports published on that IP,
+    on that IP, not host localhost or another bottle's alias.
-    not the container IP directly. force_allowlist confirms the
+    force_allowlist confirms the allowlist persisted (patching smolvm
-    allowlist persisted (patching smolvm 0.8.0's silent-drop of
+    0.8.0's silent-drop of --allow-cidr when combined with --from) and
-    --allow-cidr when combined with --from) and fails closed if it
+    fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
-    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
+    makes --from and --smolfile mutually exclusive."""
-    and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
     _smolvm.machine_create(
         plan.machine_name,
@@ -327,25 +336,26 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
-def _proxy_host(plan: SmolmachinesBottlePlan, loopback_ip: str) -> str:
+def _label_for_port(port: int) -> str:
-    """Return the host IP for TSI's allowlist and docker port-forward bindings.
+    if port == _EGRESS_PORT:
+        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.
 
-    On Linux, the guest kernel's LOCAL routing table routes all
+def _port_for_label(label: str) -> int:
-    ``127.0.0.0/8`` to the guest's own loopback — those packets never
+    if label == "egress":
-    reach eth0 and TSI never sees them. Using the per-bottle bridge
+        return _EGRESS_PORT
-    gateway (e.g. ``192.168.N.1``) instead sidesteps the problem: it
+    if label == "git-http":
-    is not a loopback address, so the guest routes it via eth0 and TSI
+        return _GIT_HTTP_PORT
-    intercepts it normally. The TSI allowlist is ``gateway/32``, which
+    if label == "supervise":
-    is distinct from the container IP (``192.168.N.2``), so the agent
+        return _SUPERVISE_PORT
-    still can't reach the egress daemon directly — TSI blocks any
+    if label.startswith("port-"):
-    connection to the container IP that isn't via the published port."""
+        return int(label.removeprefix("port-"))
-    if platform.system() == "Linux":
+    raise ValueError(f"unknown sidecar forward label: {label}")
-        return plan.bundle_gateway
-    return loopback_ip
 
 
 def _bundle_launch_spec(

bot_bottle/backend/smolmachines/port_forward.py

@@ -0,0 +1,245 @@
+"""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/resolve_plan.py

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

bot_bottle/backend/smolmachines/sidecar_bundle.py

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

bot_bottle/backend/smolmachines/smolvm.py

@@ -113,8 +113,11 @@ 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
@@ -131,19 +134,24 @@ def machine_create(
     no-pull-at-start property. The flag form gives the same
     result without the Smolfile complication.
 
-    `--net` is sent explicitly when `allow_cidrs` is non-empty.
+    `--net` is sent explicitly when `net=True` or `allow_cidrs` is
-    `--allow-cidr` implies `--net` per the CLI help, but sending
+    non-empty. `--allow-cidr` implies `--net` per the CLI help, but
-    `--net` explicitly is harmless and ensures the guest has
+    sending `--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 allow_cidrs:
+    if net or 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

@@ -209,6 +209,7 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
+        headless=True,
     )
     return _launch_bottle(
         spec,

bot_bottle/contrib/claude/Dockerfile

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

bot_bottle/git_gate_host_key.py

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

bot_bottle/yaml_subset.py

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

docs/prds/0068-smolmachines-linux.md

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

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

@@ -0,0 +1,115 @@
+# 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.300
+pyright>=1.1.411
 coverage>=7.0.0

tests/integration/test_sandbox_escape.py

@@ -81,10 +81,10 @@ class TestSandboxEscape(unittest.TestCase):
         # those are missing rather than die-ing inside backend.prepare.
         backend_name = os.environ.get("BOT_BOTTLE_BACKEND", "docker")
         if backend_name == "smolmachines":
-            if sys.platform != "darwin":
+            if sys.platform not in ("darwin", "linux"):
                 raise unittest.SkipTest(
-                    "BOT_BOTTLE_BACKEND=smolmachines is macOS-only in "
+                    f"BOT_BOTTLE_BACKEND=smolmachines is not supported "
-                    "v1 (libkrun TSI)"
+                    f"on {sys.platform} (macOS and Linux only)"
                 )
             if shutil.which("smolvm") is None:
                 raise unittest.SkipTest(

tests/integration/test_smolmachines_bundle_bringup.py

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

tests/integration/test_smolmachines_launch.py

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

tests/unit/test_git_gate_host_key.py

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

tests/unit/test_smolmachines_egress_apply.py

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

@@ -0,0 +1,66 @@
+"""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_port_forward.py

@@ -0,0 +1,325 @@
+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,6 +9,7 @@ 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
@@ -27,6 +28,7 @@ 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
@@ -432,12 +434,7 @@ class TestBundleLaunchSpec(unittest.TestCase):
 
     def test_canary_env_visible_to_smolvm_guest(self):
         plan = _plan(canary=True)
-        with patch.object(
+        stamped = _launch._discover_urls(plan, "127.0.0.16", {9099: 65000})
-            _launch._bundle,
-            "bundle_host_port",
-            return_value="65000",
-        ):
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
 
         self.assertEqual(
             "fake-canary-value",
@@ -506,23 +503,111 @@ class TestProvisionGitUser(unittest.TestCase):
         self.assertIn("bot@example.com", calls[0][0])
 
 
-class TestProxyHost(unittest.TestCase):
+class TestLaunchResourceWiring(unittest.TestCase):
-    """_proxy_host returns the bridge gateway on Linux and the loopback
+    def test_allocate_resources_uses_loopback_alias_and_bundle_name(self):
-    alias on other platforms."""
-
-    def test_linux_returns_bundle_gateway(self):
         plan = _plan()
-        with patch("bot_bottle.backend.smolmachines.launch.platform.system",
+        with patch(
-                   return_value="Linux"):
+            "bot_bottle.backend.smolmachines.launch._loopback.ensure_pool",
-            result = _launch._proxy_host(plan, "127.0.0.16")
+        ) as ensure_pool, patch(
-        self.assertEqual(plan.bundle_gateway, result)
+            "bot_bottle.backend.smolmachines.launch._loopback.allocate",
+            return_value="127.0.0.16",
+        ) as allocate:
+            loopback_ip, network = _launch._allocate_resources(plan, ExitStack())
 
-    def test_non_linux_returns_loopback(self):
+        ensure_pool.assert_called_once_with()
+        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("bot_bottle.backend.smolmachines.launch.platform.system",
+        with patch(
-                   return_value="Darwin"):
+            "bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
-            result = _launch._proxy_host(plan, "127.0.0.16")
+        ) as machine_exec, patch(
-        self.assertEqual("127.0.0.16", result)
+            "bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
+        ) 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):
@@ -544,14 +629,20 @@ class TestDiscoverUrls(unittest.TestCase):
                 ),),
             ),
         )
-        with patch.object(_launch._bundle, "bundle_host_port", return_value="9420"):
+        stamped = _launch._discover_urls(
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
+            plan,
-        self.assertEqual("127.0.0.16:9420", stamped.agent_git_gate_host)
+            "127.0.0.16",
+            {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)
-        with patch.object(_launch._bundle, "bundle_host_port", return_value="55556"):
+        stamped = _launch._discover_urls(
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
+            plan,
+            "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

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

tests/unit/test_smolmachines_smolvm.py

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