docs(research): update landscape doc — SuperHQ, multi-backend, multi-provider, in-flight directions

- Broaden scope from "Claude Code in Docker" to "AI coding agents in isolated sandboxes"
- Add SuperHQ (superhq.ai, v0.4.4) as a new adjacent-competitor entry: macOS-only
  microVM desktop app, overlaps on isolation/credential-proxy/multi-provider but has
  no manifest layer and no audit logging
- Note SuperHQ's user-voiced audit gap (Brian Cheong, Dunialabs.io) and that
  bot-bottle already covers it
- Update differentiation list to reflect three backends (Docker, Apple container,
  smolmachines) and three built-in providers (Claude Code, Codex, Pi) plus plugin system
- Add in-flight directions for forge-native dispatch (#317) and paid web control plane
  (#327) with honest framing: lifecycle concept is not novel vs. cloud services (Devin,
  Copilot Workspace); differentiation is self-hosted + manifest-driven + stronger isolation

bot_bottle/backend/smolmachines/bottle_plan.py

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

bot_bottle/backend/smolmachines/egress_apply.py

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

bot_bottle/backend/smolmachines/enumerate.py

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

bot_bottle/backend/smolmachines/launch.py

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

bot_bottle/backend/smolmachines/port_forward.py

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

bot_bottle/backend/smolmachines/resolve_plan.py

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

bot_bottle/backend/smolmachines/sidecar_bundle.py

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

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

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

docs/research/landscape-containerized-claude.md

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

tests/integration/test_smolmachines_bundle_bringup.py

@@ -0,0 +1,111 @@
+"""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,9 +1,10 @@
-"""Integration: end-to-end smolmachines launch + exec round trip.
+"""Integration: PRD 0023 chunk 2d — end-to-end launch + exec
+round trip + the acceptance probes.
 
-The smoke confirms the launch flow (sidecar bundle smolVM →
+The smoke confirms the launch flow (per-bottle docker bridge →
-host-loopback forwarders → agent smolVM with TSI allowlist → exec)
+sidecar bundle with host-loopback published ports → smolvm guest
-plumbs together end to end. The probes confirm the security
+with TSI allowlist → exec) plumbs together end to end. The probes confirm the
-properties the design pivot was about:
+security properties the design pivot was about:
 
   - **localhost-reach probe** — guest tries to dial a service
     bound on the host's `127.0.0.1`. TSI's per-bottle loopback
@@ -13,6 +14,13 @@ 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."""
 
@@ -106,8 +114,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 the per-bottle loopback alias, so
+        # allowlist contains only <bundle-ip>/32, so this must
-        # this must refuse. We use a port unlikely to be bound on the host
+        # 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(
@@ -188,6 +196,28 @@ 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_smolmachines_egress_apply.py

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

tests/unit/test_smolmachines_enumerate.py

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

tests/unit/test_smolmachines_port_forward.py

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

tests/unit/test_smolmachines_provision.py

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

tests/unit/test_smolmachines_sidecar_bundle.py

@@ -1,4 +1,9 @@
-"""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
 
@@ -9,12 +14,25 @@ from unittest.mock import patch
 
 from bot_bottle.backend.smolmachines.sidecar_bundle import (
     BundleLaunchSpec,
-    allocate_raw_host_ports,
     bundle_container_name,
     bundle_network_name,
+    create_bundle_network,
     ensure_bundle_image,
-    start_bundle_vm,
+    remove_bundle_network,
-    stop_bundle_vm,
+    start_bundle,
+    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,
     )
 
 
@@ -50,6 +68,123 @@ 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(
@@ -65,97 +200,26 @@ class TestEnsureBundleImage(unittest.TestCase):
         self.assertEqual("Dockerfile.sidecars", kwargs["dockerfile"])
 
 
-class TestStartBundleVm(unittest.TestCase):
+class TestStopBundle(unittest.TestCase):
-    def test_allocate_raw_host_ports_returns_bindable_ports(self):
+    def _patch_run(self, **kwargs):  # type: ignore
-        raw_ports = allocate_raw_host_ports((9099, 9420))
+        return patch(
-
+            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-        self.assertEqual({9099, 9420}, set(raw_ports))
+            **kwargs,
-        for host_port in raw_ports.values():
-            self.assertIsInstance(host_port, int)
-            self.assertGreater(host_port, 0)
-        self.assertNotEqual(raw_ports[9099], raw_ports[9420])
-
-    def test_creates_sidecar_vm_with_ports_volumes_and_env(self):
-        calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
-
-        def record(name):  # type: ignore
-            def inner(*args, **kwargs):  # type: ignore
-                calls.append((name, args, kwargs))
-            return inner
-
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.allocate_raw_host_ports",
-            return_value={9099: 55001, 9420: 55002},
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_create",
-            side_effect=record("create"),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_start",
-            side_effect=record("start"),
-        ), patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.wait_exec_ready",
-            side_effect=record("wait"),
-        ):
-            launch = start_bundle_vm(
-                _spec(
-                    daemons_csv="egress,git-http",
-                    environment=("FOO=bar", "TOKEN"),
-                    volumes=(("/host/routes", "/etc/egress", True),),
-                    ports_to_publish=(9099, 9420),
-                ),
-                from_path=Path("/cache/sidecar.smolmachine"),
-                host_env={"TOKEN": "secret"},
         )
 
-        self.assertEqual({9099: 55001, 9420: 55002}, launch.raw_ports)
+    def test_argv_force_removes(self):
-        create = calls[0]
+        with self._patch_run(return_value=_ok()) as m:
-        self.assertEqual("create", create[0])
+            stop_bundle("demo-abc12")
-        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"],
-                "BOT_BOTTLE_SIDECAR_DAEMONS": "egress,git-http",
+            m.call_args.args[0],
-                "FOO": "bar",
-                "TOKEN": "secret",
-            },
-            create[2]["env"],
         )
 
-
+    def test_missing_container_is_idempotent(self):
-class TestStopBundleVm(unittest.TestCase):
+        with self._patch_run(return_value=_fail(
-    def test_stops_then_deletes_sidecar_vm(self):
+            "Error: No such container: bot-bottle-sidecars-demo-abc12"
-        with patch(
+        )):
-            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
+            stop_bundle("demo-abc12")  # no raise
-        ) 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,22 +106,6 @@ class TestArgvShapes(unittest.TestCase):
         self.assertIn("--name", argv)
         self.assertIn("agent-xyz", argv)
 
-    def test_machine_create_with_net_ports_and_volumes(self):
-        with self._patch_run() as m:
-            machine_create(
-                "sidecar-xyz",
-                from_path=Path("/stage/sidecar.smolmachine"),
-                net=True,
-                ports=((55000, 9099),),
-                volumes=(("/host/routes", "/etc/egress", True),),
-            )
-        argv = m.call_args.args[0]
-        self.assertIn("--net", argv)
-        self.assertIn("-p", argv)
-        self.assertIn("55000:9099", argv)
-        self.assertIn("-v", argv)
-        self.assertIn("/host/routes:/etc/egress:ro", argv)
-
     def test_machine_create_omits_net_when_no_allow_cidrs(self):
         with self._patch_run() as m:
             machine_create("agent-xyz", from_path=Path("/x.smolmachine"))