Merge remote-tracking branch 'origin/main' into post-build-image-smoke-test

# Conflicts:
#	bot_bottle/orchestrator/lifecycle.py

bot_bottle/agent_provider.py

@@ -61,6 +61,13 @@ class AgentProviderRuntime:
     prompt_mode: PromptMode
     bypass_args: tuple[str, ...]
     resume_args: tuple[str, ...]
+    # argv run inside a throwaway container of a freshly built agent
+    # image, right after `build_image()`, to catch a build that
+    # exited 0 but produced a broken CLI (e.g. an npm
+    # optionalDependencies fetch for a platform-native binary that
+    # silently no-ops on a transient failure). Empty tuple skips the
+    # check — not every provider has opted in yet.
+    smoke_test: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)

bot_bottle/backend/__init__.py

@@ -40,7 +40,7 @@ from abc import ABC, abstractmethod
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Generic, Sequence, TypeVar
+from typing import Any, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -54,9 +54,6 @@ from ..workspace import WorkspacePlan, workspace_plan
 from .print_util import print_multi, visible_agent_env_names
 from .util import host_skill_dir
 
-if TYPE_CHECKING:
-    from .freeze import CommitCancelled, Freezer, get_freezer
-
 
 @dataclass(frozen=True)
 class BottleSpec:
@@ -575,63 +572,28 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         Not called by the launch path or the test suite."""
 
 
-# _backends is None until the first call to _get_backends(), at which
-# point all three concrete backend classes are imported and instantiated.
-# Keeping the imports out of module scope means that importing any
-# backend sub-module (e.g. `backend.docker.util`) no longer drags the
-# firecracker and macos-container implementations into memory.
-#
-# Tests may replace _backends with a {name: fake} dict via patch.object;
-# _get_backends() returns the current module-level value as-is when it
-# is not None, so test fakes take effect without triggering real imports.
-_backends: dict[str, BottleBackend[Any, Any]] | None = None
+# Import concrete backend classes AFTER the base types are defined, so
+# each backend module can pull BottleSpec / BottlePlan / BottleBackend
+# via `from . import ...` without hitting a partially-initialized module.
+from .docker import DockerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
+from .firecracker import FirecrackerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
+from .macos_container import MacosContainerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
+
+# Freezer is imported after the backend classes for the same reason:
+# Freezer.commit_slug constructs ActiveAgent, which must be fully
+# defined first.
+from .freeze import CommitCancelled, Freezer, get_freezer  # noqa: E402  # pylint: disable=wrong-import-position
 
 
-def _get_backends() -> dict[str, BottleBackend[Any, Any]]:
-    """Return the registry of all backend instances, loading lazily on first call."""
-    global _backends  # pylint: disable=global-statement
-    if _backends is None:
-        from .docker import DockerBottleBackend
-        from .firecracker import FirecrackerBottleBackend
-        from .macos_container import MacosContainerBottleBackend
-        _backends = {
-            "docker": DockerBottleBackend(),
-            "firecracker": FirecrackerBottleBackend(),
-            "macos-container": MacosContainerBottleBackend(),
-        }
-    return _backends
-
-
-def __getattr__(name: str) -> Any:
-    """Lazily surface concrete backend classes and freeze symbols at the
-    package level so existing `from bot_bottle.backend import X` and
-    `patch.object(backend_mod, X, ...)` call-sites keep working without
-    forcing an import of every backend at module-init time."""
-    if name == "DockerBottleBackend":
-        from .docker import DockerBottleBackend
-        globals()[name] = DockerBottleBackend
-        return DockerBottleBackend
-    if name == "FirecrackerBottleBackend":
-        from .firecracker import FirecrackerBottleBackend
-        globals()[name] = FirecrackerBottleBackend
-        return FirecrackerBottleBackend
-    if name == "MacosContainerBottleBackend":
-        from .macos_container import MacosContainerBottleBackend
-        globals()[name] = MacosContainerBottleBackend
-        return MacosContainerBottleBackend
-    if name == "CommitCancelled":
-        from .freeze import CommitCancelled
-        globals()[name] = CommitCancelled
-        return CommitCancelled
-    if name == "Freezer":
-        from .freeze import Freezer
-        globals()[name] = Freezer
-        return Freezer
-    if name == "get_freezer":
-        from .freeze import get_freezer
-        globals()[name] = get_freezer
-        return get_freezer
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+# The dict is heterogeneous: each value is a BottleBackend specialized
+# over its own plan type. Concrete plan types are erased here because
+# the registry is selected at runtime and the CLI only needs the
+# unparameterized methods (prepare → plan → launch(plan), cleanup, etc.).
+_BACKENDS: dict[str, BottleBackend[Any, Any]] = {
+    "docker": DockerBottleBackend(),
+    "firecracker": FirecrackerBottleBackend(),
+    "macos-container": MacosContainerBottleBackend(),
+}
 
 
 def get_bottle_backend(
@@ -649,11 +611,10 @@ def get_bottle_backend(
     Dies with a pointer at the known backends if the chosen name
     isn't implemented."""
     resolved = name or os.environ.get("BOT_BOTTLE_BACKEND") or _default_backend_name()
-    backends = _get_backends()
-    if resolved not in backends:
-        known = ", ".join(sorted(backends))
+    if resolved not in _BACKENDS:
+        known = ", ".join(sorted(_BACKENDS))
         die(f"unknown backend {resolved!r}; known backends: {known}")
-    return backends[resolved]
+    return _BACKENDS[resolved]
 
 
 def _default_backend_name() -> str:
@@ -663,17 +624,16 @@ def _default_backend_name() -> str:
     # `firecracker` binary isn't installed yet: selecting it here routes
     # start through firecracker's preflight, which prints an install
     # pointer, instead of silently falling back to docker.
-    from .firecracker import FirecrackerBottleBackend
     if FirecrackerBottleBackend.is_host_capable():
         return "firecracker"
     return "docker"
 
 
 def known_backend_names() -> tuple[str, ...]:
-    """Sorted tuple of all backend keys in `_get_backends()`. Used by
+    """Sorted tuple of all backend keys in `_BACKENDS`. Used by
     argparse (`--backend` choices) and the dashboard's backend
     picker."""
-    return tuple(sorted(_get_backends()))
+    return tuple(sorted(_BACKENDS))
 
 
 def has_backend(name: str) -> bool:
@@ -685,10 +645,9 @@ def has_backend(name: str) -> bool:
 
     Returns False for unknown names so callers can pass
     arbitrary input without separate validation."""
-    backends = _get_backends()
-    if name not in backends:
+    if name not in _BACKENDS:
         return False
-    return backends[name].is_available()
+    return _BACKENDS[name].is_available()
 
 
 def enumerate_active_agents() -> list[ActiveAgent]:
@@ -704,11 +663,10 @@ def enumerate_active_agents() -> list[ActiveAgent]:
     deterministic tiebreaker. Agents with missing metadata
     (`started_at == ""`) sort first."""
     out: list[ActiveAgent] = []
-    backends = _get_backends()
-    for name in sorted(backends):
-        if not backends[name].is_available():
+    for name in known_backend_names():
+        if not has_backend(name):
             continue
-        out.extend(backends[name].enumerate_active())
+        out.extend(_BACKENDS[name].enumerate_active())
     out.sort(key=lambda a: (a.started_at, a.slug))
     return out
 

bot_bottle/backend/docker/launch.py

@@ -36,6 +36,7 @@ from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
+from ...agent_provider import runtime_for
 from ...egress import egress_resolve_token_values
 from ...git_gate import (
     provision_git_gate_dynamic_keys,
@@ -110,6 +111,9 @@ def launch(
                 plan.image, _REPO_DIR,
                 dockerfile=plan.dockerfile_path,
             )
+            docker_mod.verify_agent_image(
+                plan.image, runtime_for(plan.agent_provider_template).smoke_test,
+            )
 
         # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
         # provisioning the bottle's repos into the shared gateway.

bot_bottle/backend/docker/util.py

@@ -4,10 +4,11 @@ existence, and building images."""
 
 from __future__ import annotations
 
+import os
 import re
 import shutil
 import subprocess
-from typing import Iterator
+from typing import Iterable, Iterator
 
 from ...docker_cmd import run_docker
 from ...log import die, info
@@ -31,7 +32,12 @@ def container_name_candidates(base: str) -> Iterator[str]:
 def runsc_available() -> bool:
     """Return True if the Docker daemon has the gVisor (`runsc`) runtime
     registered. Called once per prepare; the result lives on the plan."""
-    r = run_docker(["docker", "info", "--format", "{{json .Runtimes}}"])
+    r = subprocess.run(
+        ["docker", "info", "--format", "{{json .Runtimes}}"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
     return r.returncode == 0 and "runsc" in r.stdout
 
 
@@ -45,15 +51,20 @@ def require_docker() -> None:
 
 
 def image_exists(ref: str) -> bool:
-    return run_docker(["docker", "image", "inspect", ref]).returncode == 0
+    return _silent_run(["docker", "image", "inspect", ref]) == 0
 
 
 def container_exists(name: str) -> bool:
     """Returns True if a container (running or stopped) with the given
     name exists. Uses `docker ps -a -q -f name=^<name>$` so substring
     matches don't false-positive."""
-    result = run_docker(["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"])
-    return result.returncode == 0 and bool(result.stdout.strip())
+    result = subprocess.run(
+        ["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return bool(result.stdout.strip())
 
 
 def force_remove_container(name: str) -> None:
@@ -61,7 +72,12 @@ def force_remove_container(name: str) -> None:
     doesn't — and the rm itself is best-effort (errors swallowed) so
     this is safe to register as a teardown callback."""
     if container_exists(name):
-        run_docker(["docker", "rm", "-f", name])
+        subprocess.run(
+            ["docker", "rm", "-f", name],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            check=False,
+        )
 
 
 def docker_exec_root(container: str, argv: list[str]) -> None:
@@ -74,6 +90,29 @@ def docker_exec_root(container: str, argv: list[str]) -> None:
     )
 
 
+def docker_exec(container: str, argv: list[str], *, user: str = "") -> None:
+    """Run `docker exec` in the named container, dying with the
+    command's own stderr on failure. Pass `user=\"0\"` to run as root."""
+    cmd = ["docker", "exec"]
+    if user:
+        cmd += ["-u", user]
+    cmd += [container, *argv]
+    result = run_docker(cmd)
+    if result.returncode != 0:
+        die(
+            f"docker exec in {container} failed: "
+            f"{(result.stderr or '').strip() or '<no stderr>'}"
+        )
+
+
+def docker_cp(src: str, dest: str) -> None:
+    """Run `docker cp`, dying with the command's own stderr on failure."""
+    result = run_docker(["docker", "cp", src, dest])
+    if result.returncode != 0:
+        die(f"docker cp {src} -> {dest} failed: "
+            f"{(result.stderr or '').strip() or '<no stderr>'}")
+
+
 _SLUG_RE = re.compile(r"[^a-z0-9]+")
 
 
@@ -94,15 +133,40 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
 
     `dockerfile` is an optional path (relative to `context`, or
     absolute) for callers that need to build from a non-default
-    Dockerfile in the same context — e.g. `Dockerfile.git-gate`."""
+    Dockerfile in the same context — e.g. `Dockerfile.git-gate`.
+
+    Set `BOT_BOTTLE_NO_CACHE=1` (the `start --no-cache` flag) to force
+    `--no-cache`. The npm/curl installers some provider Dockerfiles
+    shell out to can silently no-op on a transient network failure —
+    e.g. an `optionalDependencies` fetch for a platform-native binary —
+    and Docker will then cache that broken layer indefinitely."""
     info(f"building image {ref} from {context} (layer cache keeps repeat builds fast)")
     args = ["docker", "build", "-t", ref]
+    if os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
+        args.append("--no-cache")
     if dockerfile:
         args.extend(["-f", dockerfile])
     args.append(context)
     subprocess.run(args, check=True)
 
 
+def verify_agent_image(image: str, argv: tuple[str, ...]) -> None:
+    """Run `argv` inside a throwaway container of a freshly built agent
+    image and die loudly if it fails, instead of shipping an image
+    whose CLI only breaks at first real use. No-op when the provider
+    hasn't declared a smoke test (`AgentProviderRuntime.smoke_test`)."""
+    if not argv:
+        return
+    result = run_docker(["docker", "run", "--rm", "--entrypoint", argv[0], image, *argv[1:]])
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        die(
+            f"agent image {image!r} failed its post-build smoke test "
+            f"({' '.join(argv)}): {detail}\n"
+            f"Try rebuilding from scratch: bot-bottle start --no-cache"
+        )
+
+
 # def build_image_with_cwd(
 #     derived: str,
 #     base: str,
@@ -141,10 +205,22 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
 def commit_container(container_name: str, image_tag: str) -> None:
     """Run `docker commit <container_name> <image_tag>` to snapshot the
     running container's filesystem state as a local Docker image."""
-    result = run_docker(["docker", "commit", container_name, image_tag])
+    result = subprocess.run(
+        ["docker", "commit", container_name, image_tag],
+        capture_output=True, text=True, check=False,
+    )
     if result.returncode != 0:
         die(
             f"docker commit {container_name!r} → {image_tag!r} failed: "
             f"{(result.stderr or '').strip() or '<no stderr>'}"
         )
     info(f"committed {container_name!r} → {image_tag!r}")
+
+
+def _silent_run(cmd: Iterable[str]) -> int:
+    return subprocess.run(
+        list(cmd),
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        check=False,
+    ).returncode

bot_bottle/backend/firecracker/consolidated_launch.py

@@ -0,0 +1,162 @@
+"""Consolidated bottle launch sequence for the Firecracker backend (PRD 0070).
+
+Mirrors bot_bottle.backend.docker.consolidated_launch but wired for
+Firecracker's TAP-based network topology instead of a shared Docker bridge.
+
+The per-bottle sidecar bundle (one `docker run` per bottle, published on the
+slot's host-side TAP IP) is replaced by a single persistent gateway that
+every Firecracker VM shares. The gateway runs as a Docker container in the
+dev-harness (a Firecracker VM is stage B per PRD 0070), with its ports
+published on the host (`0.0.0.0:PORT`). VMs reach it at their slot's
+host-side TAP IP because Docker's iptables PREROUTING DNAT redirects
+port 9099/9100/9420 traffic to the gateway container — a path the nft
+isolation table already allows via `ct status dnat accept` in the forward
+chain.
+
+Attribution is by the VM's guest IP, which is unspoofable by construction:
+the /31 point-to-point TAP topology + the `bot_bottle_fc` nft table ensure
+that only the expected VM can source-IP that address.
+
+Sequence:
+  1. ensure the Firecracker-flavoured orchestrator + gateway are up;
+  2. register the bottle by its guest IP (attribution key) → bottle id +
+     identity token;
+  3. provision its git-gate repos/creds into the running gateway;
+  4. fetch the shared gateway CA for the provisioner to install in the rootfs.
+
+The TAP slot allocation, rootfs build, and VM boot are the caller's job.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ...egress import EgressPlan
+from ...git_gate import GitGatePlan
+from ...orchestrator.client import OrchestratorClient
+from ...orchestrator.gateway import (
+    GATEWAY_NETWORK,
+    DockerGateway,
+)
+from ...orchestrator.lifecycle import (
+    OrchestratorService,
+    OrchestratorStartError,  # re-exported so callers can catch it
+)
+from ...orchestrator.registration import registration_inputs
+from ..docker.egress import EGRESS_PORT
+from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
+from ...supervise import SUPERVISE_PORT
+
+_GIT_HTTP_PORT = 9420
+
+# Separate names from the Docker gateway so both backends can coexist on one
+# host (and for clarity in `docker ps` output).
+_FC_GATEWAY_NAME = "bot-bottle-fc-gateway"
+_FC_ORCHESTRATOR_NAME = "bot-bottle-fc-orchestrator"
+_FC_ORCHESTRATOR_LABEL = "bot-bottle-fc-orchestrator=1"
+# Ports the gateway publishes on the host so Firecracker VMs can reach it
+# via their TAP link. Docker's PREROUTING DNAT + nft's `ct status dnat
+# accept` in the forward chain route the traffic.
+_FC_GATEWAY_HOST_PORTS = (EGRESS_PORT, SUPERVISE_PORT, _GIT_HTTP_PORT)
+
+
+class ConsolidatedLaunchError(RuntimeError):
+    """The consolidated register/provision sequence could not complete."""
+
+
+@dataclass(frozen=True)
+class LaunchContext:
+    """What the Firecracker launch needs from the consolidated sequence."""
+
+    bottle_id: str
+    identity_token: str
+    source_ip: str          # the VM's guest IP — the attribution key
+    gateway_ca_pem: str     # the shared gateway CA the provisioner installs
+    orchestrator_url: str
+
+
+class _FirecrackerOrchestratorService(OrchestratorService):
+    """Dev-harness orchestrator for the Firecracker backend.
+
+    Uses a gateway that publishes its ports on the host so Firecracker VMs can
+    reach it via their TAP link. The gateway and orchestrator containers use
+    `*-fc-*` names so both backends can run independently on the same host.
+    """
+
+    def __init__(self, **kwargs: object) -> None:
+        super().__init__(
+            orchestrator_name=_FC_ORCHESTRATOR_NAME,
+            orchestrator_label=_FC_ORCHESTRATOR_LABEL,
+            **kwargs,  # type: ignore[arg-type]
+        )
+
+    def _gateway(self) -> DockerGateway:
+        return DockerGateway(
+            self.image,
+            name=_FC_GATEWAY_NAME,
+            network=self.network,
+            orchestrator_url=self.internal_url,
+            host_port_bindings=_FC_GATEWAY_HOST_PORTS,
+        )
+
+
+def launch_consolidated(
+    egress_plan: EgressPlan,
+    git_gate_plan: GitGatePlan,
+    *,
+    guest_ip: str,
+    image_ref: str = "",
+    tokens: dict[str, str] | None = None,
+    service: OrchestratorService | None = None,
+    gateway_name: str = _FC_GATEWAY_NAME,
+) -> LaunchContext:
+    """Ensure the orchestrator + Firecracker gateway are up, register the
+    bottle by its guest IP, and provision its git-gate state. Returns the
+    context the VM launch needs. Raises `ConsolidatedLaunchError` (or the
+    primitives' own errors) on failure — the caller tears down on failure."""
+    service = service or _FirecrackerOrchestratorService()
+    url = service.ensure_running()
+    client = OrchestratorClient(url)
+
+    inputs = registration_inputs(egress_plan)
+    reg = client.register_bottle(
+        guest_ip, image_ref=image_ref, policy=inputs.policy,
+        metadata=inputs.metadata, tokens=tokens,
+    )
+    try:
+        provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
+    except Exception:
+        client.teardown_bottle(reg.bottle_id)
+        raise
+
+    # Fetch the shared gateway CA here so the caller can install it in the
+    # rootfs (the same CA every agent on this host trusts for TLS interception).
+    gateway_ca_pem = DockerGateway(
+        name=gateway_name, network=GATEWAY_NETWORK,
+    ).ca_cert_pem()
+
+    return LaunchContext(
+        bottle_id=reg.bottle_id,
+        identity_token=reg.identity_token,
+        source_ip=guest_ip,
+        gateway_ca_pem=gateway_ca_pem,
+        orchestrator_url=url,
+    )
+
+
+def teardown_consolidated(
+    bottle_id: str, *, orchestrator_url: str, gateway_name: str = _FC_GATEWAY_NAME,
+) -> None:
+    """Deregister the bottle and remove its git-gate state from the gateway.
+    Both steps are idempotent so this is safe from a cleanup trap."""
+    OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
+    deprovision_git_gate(gateway_name, bottle_id)
+
+
+__all__ = [
+    "LaunchContext",
+    "launch_consolidated",
+    "teardown_consolidated",
+    "ConsolidatedLaunchError",
+    "OrchestratorStartError",
+]

bot_bottle/backend/firecracker/launch.py

@@ -1,75 +1,64 @@
-"""Launch flow for the Firecracker backend.
+"""Launch flow for the Firecracker backend (PRD 0070, consolidated).
 
 Per bottle:
-  1. mint the egress CA, build the agent image (docker), export it to a
-     cached ext4 rootfs;
-  2. claim a free TAP pool slot (rootless flock);
-  3. bring up the Docker sidecar bundle, publishing egress / git-gate /
-     supervise on the slot's host-side TAP IP at fixed ports;
-  4. boot the microVM on that TAP; wait for SSH;
-  5. provision (CA, prompt, skills, workspace, git, supervise) over SSH.
+  1. build the agent image (docker), export it to a cached ext4 rootfs;
+  2. ensure the per-host orchestrator + shared gateway are up;
+  3. claim a free TAP pool slot (rootless flock);
+  4. register the bottle on the orchestrator by the VM's guest IP (the
+     attribution key) and provision its git-gate state into the gateway;
+  5. boot the microVM on that TAP; wait for SSH;
+  6. provision (shared gateway CA, prompt, skills, workspace, git, supervise)
+     over SSH.
+
+The per-bottle Docker sidecar bundle is gone. The shared gateway handles
+egress / git-gate / supervise for every VM; Docker's PREROUTING DNAT routes
+the VMs' traffic to it, and the nft table's `ct status dnat accept` rule
+in the forward chain lets it pass. The VM still sends to `host_tap_ip:PORT`
+— the address its world is, by nft design, limited to.
 
 Isolation is enforced by the operator-provisioned nft table (checked
-fail-closed in preflight): a VM reaches only its sidecar (DNAT'd from
-the host TAP IP) and nothing else. The agent's HTTPS_PROXY therefore
-points at `http://<host_tap_ip>:9099`, its only route to the world.
+fail-closed in preflight): a VM reaches only the sidecar (DNAT'd from
+the host TAP IP) and nothing else.
 """
 
 from __future__ import annotations
 
 import dataclasses
 import os
-import subprocess
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
+from ...agent_provider import runtime_for
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
     read_committed_image,
 )
 from ...egress import (
-    EGRESS_ROUTES_IN_CONTAINER,
     egress_agent_env_entries,
     egress_resolve_token_values,
-    egress_sidecar_env_entries,
 )
 from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import die, info, warn
-from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
-from ...util import expand_tilde
-from ..docker.egress import (
-    EGRESS_CA_IN_CONTAINER,
-    EGRESS_PORT,
-    egress_tls_init,
-)
-from ..docker.git_gate import (
-    GIT_GATE_ACCESS_HOOK_IN_CONTAINER,
-    GIT_GATE_CREDS_DIR_IN_CONTAINER,
-    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
-    GIT_GATE_HOOK_IN_CONTAINER,
-)
-from ..docker.sidecar_bundle import (
-    SIDECAR_BUNDLE_DOCKERFILE,
-    SIDECAR_BUNDLE_IMAGE,
-)
+from ...log import info, warn
+from ...supervise import SUPERVISE_PORT
+from ..docker import util as docker_mod
+from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
 from . import firecracker_vm, isolation_probe, netpool, util
 from .bottle import FirecrackerBottle
 from .bottle_plan import FirecrackerBottlePlan
+from .consolidated_launch import (
+    launch_consolidated,
+    teardown_consolidated,
+)
 
 
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 _GIT_HTTP_PORT = 9420
-_GIT_GATE_READY_FILE = "/run/git-gate/ready"
-
-
-def sidecar_container_name(slug: str) -> str:
-    return f"bot-bottle-sidecars-{slug}"
 
 
 @contextmanager
@@ -78,6 +67,8 @@ def launch(
     *,
     provision: Callable[[FirecrackerBottlePlan, "FirecrackerBottle"], str | None],
 ) -> Generator[FirecrackerBottle, None, None]:
+    """Build, launch, and provision a Firecracker bottle via the consolidated
+    orchestrator. Teardown on exit."""
     stack = ExitStack()
     bottle_for_revoke = plan.manifest.bottle
     git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
@@ -94,26 +85,75 @@ def launch(
             raise teardown_exc
 
     try:
-        plan = _mint_certs(plan)
+        # Step 1: agent image. The sidecar bundle image is built by the
+        # orchestrator service (ensure_running → ensure_built); we only
+        # build the agent image here. Use a committed snapshot when available.
         plan = _build_agent_image(plan)
 
-        # Claim a TAP slot; the flock is held until teardown closes it.
+        # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any.
+        git_gate_plan = plan.git_gate_plan
+        if git_gate_plan.upstreams:
+            git_gate_plan = provision_git_gate_dynamic_keys(
+                plan.manifest.bottle, git_gate_plan, git_gate_state_dir(plan.slug),
+            )
+
+        # Step 3: claim a TAP slot; the flock is held until teardown.
         slot, lock = netpool.allocate(plan.slug)
         stack.callback(lock.close)
         info(f"firecracker slot {slot.iface}: host={slot.host_ip} "
              f"guest={slot.guest_ip}")
 
-        plan = _provision_git_gate_keys(plan)
+        # Step 4: register on the orchestrator + provision this bottle's
+        # git-gate state into the shared gateway. The per-bottle egress tokens
+        # are resolved from the host env now and handed to the orchestrator
+        # (in memory) for the gateway to inject — the agent never sees them.
+        # Attribution is by the VM's guest IP (unspoofable via /31 TAP + nft).
+        effective_env = {**os.environ, **plan.agent_provision.provisioned_env}
+        token_values = egress_resolve_token_values(
+            plan.egress_plan.token_env_map, effective_env,
+        )
+        ctx = launch_consolidated(
+            plan.egress_plan, git_gate_plan,
+            guest_ip=slot.guest_ip,
+            image_ref=plan.image,
+            tokens=token_values,
+        )
+        stack.callback(
+            teardown_consolidated, ctx.bottle_id,
+            orchestrator_url=ctx.orchestrator_url,
+        )
 
-        sidecar_name = sidecar_container_name(plan.slug)
-        _force_remove_container(sidecar_name)
-        _start_sidecar_bundle(plan, sidecar_name, slot.host_ip)
-        stack.callback(_force_remove_container, sidecar_name)
-        _stage_git_gate(plan, sidecar_name)
+        # Step 5: install the SHARED gateway CA (replaces the per-bottle CA).
+        # Write it to a stable host path so the provisioner can copy it over SSH.
+        ca_dir = egress_state_dir(plan.slug) / "gateway-ca"
+        ca_dir.mkdir(parents=True, exist_ok=True)
+        ca_file = ca_dir / "gateway-ca.pem"
+        ca_file.write_text(ctx.gateway_ca_pem)
+        egress_plan = dataclasses.replace(
+            plan.egress_plan,
+            mitmproxy_ca_host_path=ca_file,
+            mitmproxy_ca_cert_only_host_path=ca_file,
+        )
+        # Point the agent's git-gate insteadOf rewrites and supervise MCP URL
+        # at the shared gateway (reached at the slot's host TAP IP — the VM
+        # sends there and Docker DNAT routes to the gateway container).
+        git_gate_url = (
+            f"http://{slot.host_ip}:{_GIT_HTTP_PORT}" if git_gate_plan.upstreams else ""
+        )
+        supervise_url = (
+            f"http://{slot.host_ip}:{SUPERVISE_PORT}/"
+            if plan.supervise_plan is not None else ""
+        )
+        plan = dataclasses.replace(
+            plan,
+            git_gate_plan=git_gate_plan,
+            egress_plan=egress_plan,
+            agent_proxy_url=f"http://{slot.host_ip}:{EGRESS_PORT}",
+            agent_git_gate_url=git_gate_url,
+            agent_supervise_url=supervise_url,
+        )
 
-        plan = _stamp_agent_urls(plan, slot.host_ip)
-
-        # Build the per-bottle rootfs + SSH key, then boot.
+        # Step 6: build the per-bottle rootfs + SSH key, then boot.
         base_dir = util.build_base_rootfs_dir(plan.image)
         run_dir = util.cache_dir() / "run" / plan.slug
         run_dir.mkdir(parents=True, exist_ok=True)
@@ -133,8 +173,8 @@ def launch(
         stack.callback(vm.terminate)
         firecracker_vm.wait_for_ssh(vm, private_key)
 
-        # Authoritative fail-closed egress-boundary check, before the
-        # agent runs: prove the VM cannot reach the host directly.
+        # Authoritative fail-closed egress-boundary check, before the agent
+        # runs: prove the VM cannot reach the host directly.
         isolation_probe.verify_isolation(private_key, slot.guest_ip)
 
         bottle = FirecrackerBottle(
@@ -159,171 +199,21 @@ def launch(
         teardown()
 
 
-def _mint_certs(plan: FirecrackerBottlePlan) -> FirecrackerBottlePlan:
-    egress_ca_host, egress_ca_cert_only = egress_tls_init(egress_state_dir(plan.slug))
-    egress_plan = dataclasses.replace(
-        plan.egress_plan,
-        mitmproxy_ca_host_path=egress_ca_host,
-        mitmproxy_ca_cert_only_host_path=egress_ca_cert_only,
-    )
-    return dataclasses.replace(plan, egress_plan=egress_plan)
-
-
 def _build_agent_image(plan: FirecrackerBottlePlan) -> FirecrackerBottlePlan:
-    _docker_build(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
     committed = read_committed_image(plan.slug)
-    if committed and _image_exists(committed):
+    if committed and docker_mod.image_exists(committed):
         info(f"using committed image {committed!r}")
         return dataclasses.replace(
             plan,
             agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
         )
-    _docker_build(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    docker_mod.verify_agent_image(
+        plan.image, runtime_for(plan.agent_provider_template).smoke_test,
+    )
     return plan
 
 
-def _provision_git_gate_keys(plan: FirecrackerBottlePlan) -> FirecrackerBottlePlan:
-    if not plan.git_gate_plan.upstreams:
-        return plan
-    git_gate_plan = provision_git_gate_dynamic_keys(
-        plan.manifest.bottle, plan.git_gate_plan, git_gate_state_dir(plan.slug),
-    )
-    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
-
-
-def _stamp_agent_urls(
-    plan: FirecrackerBottlePlan, host_ip: str,
-) -> FirecrackerBottlePlan:
-    proxy_url = f"http://{host_ip}:{EGRESS_PORT}"
-    supervise_url = (
-        f"http://{host_ip}:{SUPERVISE_PORT}/" if plan.supervise_plan is not None else ""
-    )
-    git_gate_url = (
-        f"http://{host_ip}:{_GIT_HTTP_PORT}" if plan.git_gate_plan.upstreams else ""
-    )
-    return dataclasses.replace(
-        plan,
-        agent_proxy_url=proxy_url,
-        agent_git_gate_url=git_gate_url,
-        agent_supervise_url=supervise_url,
-    )
-
-
-# --- sidecar bundle (Docker) ----------------------------------------
-
-def _start_sidecar_bundle(
-    plan: FirecrackerBottlePlan, sidecar_name: str, host_ip: str,
-) -> None:
-    argv = ["docker", "run", "--name", sidecar_name, "--detach", "--rm",
-            "-e", f"BOT_BOTTLE_SIDECAR_DAEMONS={','.join(_sidecar_daemons(plan))}"]
-    for entry in _sidecar_env_entries(plan):
-        argv += ["-e", entry]
-    for host_path, container_path, read_only in _sidecar_mounts(plan):
-        argv += ["-v", f"{host_path}:{container_path}{':ro' if read_only else ''}"]
-    # Publish on the slot's host TAP IP at fixed ports — each bottle
-    # has a distinct host_ip, so fixed ports never collide, and the VM
-    # reaches them at a stable, well-known address (its only route out).
-    for port in _sidecar_ports(plan):
-        argv += ["-p", f"{host_ip}:{port}:{port}"]
-    argv.append(SIDECAR_BUNDLE_IMAGE)
-
-    effective_env = {**dict(os.environ), **plan.agent_provision.provisioned_env}
-    token_values = egress_resolve_token_values(
-        plan.egress_plan.token_env_map, effective_env,
-    )
-    env = {**os.environ, **token_values}
-    info(f"docker run sidecar bundle {sidecar_name} (published on {host_ip})")
-    result = subprocess.run(argv, capture_output=True, text=True, env=env, check=False)
-    if result.returncode != 0:
-        die(f"docker run for sidecar bundle {sidecar_name} failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}")
-
-
-def _sidecar_daemons(plan: FirecrackerBottlePlan) -> tuple[str, ...]:
-    daemons = ["egress"]
-    if plan.git_gate_plan.upstreams:
-        daemons += ["git-gate", "git-http"]
-    if plan.supervise_plan is not None:
-        daemons.append("supervise")
-    return tuple(daemons)
-
-
-def _sidecar_ports(plan: FirecrackerBottlePlan) -> tuple[int, ...]:
-    ports = [EGRESS_PORT]
-    if plan.git_gate_plan.upstreams:
-        ports.append(_GIT_HTTP_PORT)
-    if plan.supervise_plan is not None:
-        ports.append(SUPERVISE_PORT)
-    return tuple(ports)
-
-
-def _sidecar_env_entries(plan: FirecrackerBottlePlan) -> tuple[str, ...]:
-    env: list[str] = list(egress_sidecar_env_entries(plan.egress_plan))
-    if plan.git_gate_plan.upstreams:
-        env.append(f"BOT_BOTTLE_GIT_GATE_READY_FILE={_GIT_GATE_READY_FILE}")
-    if plan.supervise_plan is not None:
-        env += [
-            f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
-            f"SUPERVISE_PORT={SUPERVISE_PORT}",
-        ]
-    return tuple(env)
-
-
-def _sidecar_mounts(
-    plan: FirecrackerBottlePlan,
-) -> tuple[tuple[str, str, bool], ...]:
-    mounts: list[tuple[str, str, bool]] = []
-    ep = plan.egress_plan
-    mounts.append((str(ep.mitmproxy_ca_host_path.parent),
-                   str(Path(EGRESS_CA_IN_CONTAINER).parent), False))
-    if ep.routes:
-        mounts.append((str(ep.routes_path.parent),
-                       str(Path(EGRESS_ROUTES_IN_CONTAINER).parent), True))
-    sp = plan.supervise_plan
-    if sp is not None:
-        mounts.append((str(sp.db_path.parent),
-                       str(Path(DB_PATH_IN_CONTAINER).parent), False))
-    return tuple(mounts)
-
-
-def _stage_git_gate(plan: FirecrackerBottlePlan, sidecar_name: str) -> None:
-    gp = plan.git_gate_plan
-    if not gp.upstreams:
-        return
-    _docker_exec(sidecar_name, [
-        "mkdir", "-p",
-        str(Path(GIT_GATE_HOOK_IN_CONTAINER).parent),
-        GIT_GATE_CREDS_DIR_IN_CONTAINER, "/git",
-        str(Path(_GIT_GATE_READY_FILE).parent),
-    ])
-    for host_path, container_path in _git_gate_files(plan):
-        _docker_cp(host_path, f"{sidecar_name}:{container_path}")
-    _docker_exec(sidecar_name, [
-        "sh", "-c",
-        f"chmod 755 {GIT_GATE_ENTRYPOINT_IN_CONTAINER} "
-        f"{GIT_GATE_HOOK_IN_CONTAINER} {GIT_GATE_ACCESS_HOOK_IN_CONTAINER} && "
-        f"chmod 600 {GIT_GATE_CREDS_DIR_IN_CONTAINER}/* && "
-        f"touch {_GIT_GATE_READY_FILE}",
-    ])
-
-
-def _git_gate_files(plan: FirecrackerBottlePlan) -> tuple[tuple[str, str], ...]:
-    gp = plan.git_gate_plan
-    files: list[tuple[str, str]] = [
-        (str(gp.entrypoint_script), GIT_GATE_ENTRYPOINT_IN_CONTAINER),
-        (str(gp.hook_script), GIT_GATE_HOOK_IN_CONTAINER),
-        (str(gp.access_hook_script), GIT_GATE_ACCESS_HOOK_IN_CONTAINER),
-    ]
-    for upstream in gp.upstreams:
-        files.append((expand_tilde(upstream.identity_file),
-                      f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{upstream.name}-key"))
-        if upstream.known_hosts_file:
-            files.append((str(upstream.known_hosts_file),
-                          f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{upstream.name}-known_hosts"))
-    return tuple(files)
-
-
 # --- agent guest env -------------------------------------------------
 
 def _agent_guest_env(plan: FirecrackerBottlePlan, host_ip: str) -> dict[str, str]:
@@ -355,50 +245,3 @@ def _agent_guest_env(plan: FirecrackerBottlePlan, host_ip: str) -> dict[str, str
         if value is not None:
             env[name] = value
     return env
-
-
-# --- docker helpers --------------------------------------------------
-
-def _docker_build(ref: str, context: str, *, dockerfile: str = "") -> None:
-    info(f"docker build {ref}")
-    args = ["docker", "build", "-t", ref]
-    if dockerfile:
-        if not os.path.isabs(dockerfile):
-            dockerfile = os.path.join(context, dockerfile)
-        args += ["-f", dockerfile]
-    args.append(context)
-    result = subprocess.run(args, check=False)
-    if result.returncode != 0:
-        die(f"docker build for {ref!r} failed")
-
-
-def _image_exists(ref: str) -> bool:
-    return subprocess.run(
-        ["docker", "image", "inspect", ref],
-        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
-    ).returncode == 0
-
-
-def _force_remove_container(name: str) -> None:
-    subprocess.run(
-        ["docker", "rm", "-f", name],
-        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
-    )
-
-
-def _docker_exec(name: str, argv: list[str]) -> None:
-    result = subprocess.run(
-        ["docker", "exec", name, *argv], capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        die(f"docker exec in {name} failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}")
-
-
-def _docker_cp(host_path: str, dest: str) -> None:
-    result = subprocess.run(
-        ["docker", "cp", host_path, dest], capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        die(f"docker cp {host_path} -> {dest} failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}")

bot_bottle/backend/macos_container/launch.py

@@ -17,6 +17,7 @@ from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
+from ...agent_provider import runtime_for
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
@@ -169,6 +170,9 @@ def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
         _REPO_DIR,
         dockerfile=plan.dockerfile_path,
     )
+    container_mod.verify_agent_image(
+        plan.image, runtime_for(plan.agent_provider_template).smoke_test,
+    )
     return plan
 
 

bot_bottle/backend/macos_container/util.py

@@ -60,13 +60,21 @@ def dns_server() -> str:
 
 
 def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
-    """Build an OCI image with Apple's BuildKit-backed `container build`."""
+    """Build an OCI image with Apple's BuildKit-backed `container build`.
+
+    Set `BOT_BOTTLE_NO_CACHE=1` (the `start --no-cache` flag) to force
+    `--no-cache`. The npm/curl installers some provider Dockerfiles
+    shell out to can silently no-op on a transient network failure —
+    e.g. an `optionalDependencies` fetch for a platform-native binary —
+    and the builder will then cache that broken layer indefinitely."""
     info(
         f"building image {ref} from {context} with Apple Container "
         "(layer cache keeps repeat builds fast)"
     )
     _ensure_builder_dns()
     args = [_CONTAINER, "build", "-t", ref, "--dns", dns_server()]
+    if os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
+        args.append("--no-cache")
     if dockerfile:
         # `container build` resolves -f relative to the current working
         # directory, not the build context. Anchor a relative Dockerfile to
@@ -78,6 +86,28 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
     subprocess.run(args, check=True)
 
 
+def verify_agent_image(image: str, argv: tuple[str, ...]) -> None:
+    """Run `argv` inside a throwaway container of a freshly built agent
+    image and die loudly if it fails, instead of shipping an image
+    whose CLI only breaks at first real use. No-op when the provider
+    hasn't declared a smoke test (`AgentProviderRuntime.smoke_test`)."""
+    if not argv:
+        return
+    result = subprocess.run(
+        [_CONTAINER, "run", "--rm", "--entrypoint", argv[0], image, *argv[1:]],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        die(
+            f"agent image {image!r} failed its post-build smoke test "
+            f"({' '.join(argv)}): {detail}\n"
+            f"Try rebuilding from scratch: bot-bottle start --no-cache"
+        )
+
+
 def commit_container(container_name: str, image_tag: str) -> None:
     """Snapshot a running Apple Container as a local image.
 

bot_bottle/cli/start.py

@@ -46,6 +46,17 @@ def cmd_start(argv: list[str]) -> int:
     parser = argparse.ArgumentParser(prog=f"{PROG} start", add_help=True)
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--cwd", action="store_true", help="copy host cwd into the running bottle")
+    parser.add_argument(
+        "--no-cache",
+        action="store_true",
+        help=(
+            "rebuild agent/sidecar images from scratch, bypassing the "
+            "build layer cache. Use when an image looks broken after a "
+            "dependency bump — e.g. an installer's optionalDependencies "
+            "fetch silently no-op'd on a transient failure and got baked "
+            "into a cached layer."
+        ),
+    )
     parser.add_argument(
         "--backend",
         choices=known_backend_names(),
@@ -97,6 +108,11 @@ def cmd_start(argv: list[str]) -> int:
     args = parser.parse_args(argv)
 
     dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
+    if args.no_cache or os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
+        # Read by build_image() in each backend's util module — set here
+        # so both the interactive and --headless paths pick it up without
+        # threading a no_cache field through every backend's plan dataclass.
+        os.environ["BOT_BOTTLE_NO_CACHE"] = "1"
 
     manifest = ManifestIndex.resolve(USER_CWD)
     backend_name: str | None = args.backend

bot_bottle/contrib/claude/agent_provider.py

@@ -91,6 +91,7 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="append_file",
     bypass_args=("--dangerously-skip-permissions",),
     resume_args=("--continue",),
+    smoke_test=("claude", "--version"),
 )
 
 

bot_bottle/contrib/codex/agent_provider.py

@@ -61,6 +61,7 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="read_prompt_file",
     bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
     resume_args=("resume", "--last"),
+    smoke_test=(_CODEX_CLI, "--version"),
 )
 
 

bot_bottle/orchestrator/gateway.py

@@ -100,6 +100,7 @@ class DockerGateway(Gateway):
         orchestrator_url: str = "",
         build_context: Path | None = None,
         dockerfile: str | None = GATEWAY_DOCKERFILE,
+        host_port_bindings: tuple[int, ...] = (),
     ) -> None:
         self.image_ref = image_ref
         self.name = name
@@ -110,6 +111,10 @@ class DockerGateway(Gateway):
         self._orchestrator_url = orchestrator_url
         self._build_context = build_context or _REPO_ROOT
         self._dockerfile = dockerfile
+        # Ports published on the host (0.0.0.0). Used by the Firecracker
+        # backend's dev-harness gateway so VMs can reach it via their TAP link;
+        # Docker's DNAT + the nft `ct status dnat accept` rule handle the rest.
+        self._host_port_bindings = host_port_bindings
 
     def image_exists(self) -> bool:
         return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0
@@ -188,6 +193,8 @@ class DockerGateway(Gateway):
             # trust it) — see GATEWAY_CA_VOLUME.
             "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
         ]
+        for port in self._host_port_bindings:
+            argv += ["--publish", f"0.0.0.0:{port}:{port}"]
         if self._orchestrator_url:
             # Makes the gateway's egress / git / supervise daemons multi-tenant:
             # each request resolves source-IP -> policy against the control plane.

bot_bottle/orchestrator/lifecycle.py

@@ -16,6 +16,7 @@ names + the published port).
 
 from __future__ import annotations
 
+import hashlib
 import time
 import urllib.error
 import urllib.request
@@ -29,6 +30,10 @@ from .gateway import GATEWAY_IMAGE, GATEWAY_NETWORK, DockerGateway
 DEFAULT_PORT = 8099
 ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
 ORCHESTRATOR_LABEL = "bot-bottle-orchestrator=1"
+# Baked onto the container as a label so `ensure_running` can tell whether the
+# running process is executing the *current* bind-mounted source — see
+# `_source_hash`.
+ORCHESTRATOR_SOURCE_HASH_LABEL = "bot-bottle-orchestrator-source-hash"
 
 # The repo root is bind-mounted into the control-plane container so
 # `python -m bot_bottle.orchestrator` resolves the package (the orchestrator
@@ -46,9 +51,31 @@ class OrchestratorStartError(RuntimeError):
     """The orchestrator container did not become healthy within the timeout."""
 
 
+def _source_hash(repo_root: Path) -> str:
+    """Content hash of the orchestrator's bind-mounted Python source (the
+    `bot_bottle` package the control-plane process imports). This only
+    changes when the code that would actually run inside the container
+    changes — `ensure_running` recreates the container on a mismatch and
+    otherwise leaves a healthy one alone, so a bottle launch that isn't
+    accompanied by a code change doesn't restart the process and drop every
+    *other* active bottle's in-memory egress tokens (`Orchestrator._tokens`
+    in `service.py`, never persisted to disk by design)."""
+    h = hashlib.sha256()
+    for path in sorted((repo_root / "bot_bottle").rglob("*.py")):
+        h.update(str(path.relative_to(repo_root)).encode())
+        h.update(path.read_bytes())
+    return h.hexdigest()
+
+
 class OrchestratorService:
     """Manages the orchestrator control-plane container + the shared gateway.
-    Callers only need `ensure_running()` + `url`."""
+    Callers only need `ensure_running()` + `url`.
+
+    `orchestrator_name` / `orchestrator_label` let backends run independent
+    orchestrators on the same host without name collisions (e.g. the
+    Firecracker backend uses `bot-bottle-fc-orchestrator` alongside the Docker
+    backend's `bot-bottle-orchestrator`). Subclass and override `_gateway()`
+    to supply a backend-specific gateway variant."""
 
     def __init__(
         self,
@@ -58,12 +85,16 @@ class OrchestratorService:
         image: str = GATEWAY_IMAGE,
         repo_root: Path = _REPO_ROOT,
         host_root: Path | None = None,
+        orchestrator_name: str = ORCHESTRATOR_NAME,
+        orchestrator_label: str = ORCHESTRATOR_LABEL,
     ) -> None:
         self.port = port
         self.network = network
         self.image = image
         self._repo_root = repo_root
         self._host_root = host_root or bot_bottle_root()
+        self._orchestrator_name = orchestrator_name
+        self._orchestrator_label = orchestrator_label
 
     @property
     def url(self) -> str:
@@ -75,7 +106,7 @@ class OrchestratorService:
         """Control-plane URL as the gateway container reaches it — by name over
         docker DNS on the shared network. This is the gateway's
         BOT_BOTTLE_ORCHESTRATOR_URL."""
-        return f"http://{ORCHESTRATOR_NAME}:{self.port}"
+        return f"http://{self._orchestrator_name}:{self.port}"
 
     def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
         try:
@@ -88,14 +119,17 @@ class OrchestratorService:
         proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"])
         return name in proc.stdout.split()
 
-    def _run_orchestrator_container(self) -> None:
+    def _run_orchestrator_container(self, source_hash: str) -> None:
         """Start the control-plane container (idempotent: clears a stale
-        fixed-name container first). Register-only broker → no docker socket."""
-        run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME])
+        fixed-name container first). Register-only broker → no docker socket.
+        Labels the container with `source_hash` so a later `ensure_running`
+        can detect a real code change (see `_source_hash`)."""
+        run_docker(["docker", "rm", "--force", self._orchestrator_name])
         proc = run_docker([
             "docker", "run", "--detach",
-            "--name", ORCHESTRATOR_NAME,
-            "--label", ORCHESTRATOR_LABEL,
+            "--name", self._orchestrator_name,
+            "--label", self._orchestrator_label,
+            "--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={source_hash}",
             "--network", self.network,
             # Host CLI reaches the control plane here; bound to loopback so it
             # is not exposed on the host's external interfaces.
@@ -119,26 +153,46 @@ class OrchestratorService:
     def _gateway(self) -> DockerGateway:
         return DockerGateway(self.image, network=self.network, orchestrator_url=self.internal_url)
 
+    def _orchestrator_source_current(self, current_hash: str) -> bool:
+        """True iff the running orchestrator container was created from the
+        *current* bind-mounted source. Mirrors `DockerGateway`'s
+        image-staleness check, but by content hash rather than image id since
+        the orchestrator runs bind-mounted source, not a built image."""
+        if not self._container_running(self._orchestrator_name):
+            return False
+        proc = run_docker([
+            "docker", "inspect", "--format",
+            "{{ index .Config.Labels \"" + ORCHESTRATOR_SOURCE_HASH_LABEL + "\" }}",
+            self._orchestrator_name,
+        ])
+        if proc.returncode != 0:
+            return True  # can't compare -> don't churn a working container
+        return proc.stdout.strip() == current_hash
+
     def ensure_running(
         self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
     ) -> str:
         """Ensure the control plane + shared gateway are up; return the host
-        control-plane URL. Idempotent — a healthy control plane and a running
-        gateway are left untouched. Raises `OrchestratorStartError` on
-        timeout."""
+        control-plane URL. Idempotent — a healthy control plane running
+        current code and a running gateway are left untouched. Raises
+        `OrchestratorStartError` on timeout."""
         gateway = self._gateway()
         gateway.ensure_built()  # rebuild the bundle image on a source change
         gateway.ensure_running()  # creates the shared network + (re)starts gateway
 
-        # Always (re)create the orchestrator container. It runs the repo's code
-        # bind-mounted, but the Python process loaded that code at startup and
-        # won't reload — so reusing a healthy-but-stale container would keep
-        # running OLD control-plane code (e.g. dropping the tokens field). Cheap
-        # (~seconds); the registry DB persists and the current launch
-        # re-registers its own in-memory state. (The dedicated orchestrator
-        # image follow-up replaces this with image-staleness detection.)
-        log.info("starting orchestrator container", context={"name": ORCHESTRATOR_NAME})
-        self._run_orchestrator_container()
+        # Recreate the orchestrator container only when its bind-mounted
+        # source has actually changed since it started — its Python process
+        # loaded that code at startup and won't reload, so a stale container
+        # would keep running OLD control-plane code. Recreating on *every*
+        # launch (the prior behaviour) would drop every other active
+        # bottle's in-memory egress tokens each time a new bottle starts,
+        # since the orchestrator process holds them only in memory (#381).
+        current_hash = _source_hash(self._repo_root)
+        if self.is_healthy() and self._orchestrator_source_current(current_hash):
+            return self.url
+
+        log.info("starting orchestrator container", context={"name": self._orchestrator_name})
+        self._run_orchestrator_container(current_hash)
 
         deadline = time.monotonic() + startup_timeout
         while time.monotonic() < deadline:
@@ -152,7 +206,7 @@ class OrchestratorService:
 
     def stop(self) -> None:
         """Remove the orchestrator + gateway containers (idempotent)."""
-        run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME])
+        run_docker(["docker", "rm", "--force", self._orchestrator_name])
         self._gateway().stop()
 
 

tests/unit/test_backend_selection.py

@@ -40,7 +40,7 @@ class TestGetBottleBackend(unittest.TestCase):
                 return True
 
         with patch.dict(os.environ, {}, clear=True), \
-             patch.object(backend_mod, "_backends", {
+             patch.object(backend_mod, "_BACKENDS", {
                  "macos-container": _FakeBackend(),
                  "docker": _FakeBackend(),
              }):
@@ -61,7 +61,7 @@ class TestGetBottleBackend(unittest.TestCase):
         with patch.dict(os.environ, {}, clear=True), \
              patch.object(backend_mod.FirecrackerBottleBackend,
                           "is_host_capable", classmethod(lambda cls: False)), \
-             patch.object(backend_mod, "_backends", {
+             patch.object(backend_mod, "_BACKENDS", {
                  "macos-container": _FakeBackend("macos-container", False),
                  "docker": _FakeBackend("docker", True),
              }):
@@ -83,7 +83,7 @@ class TestGetBottleBackend(unittest.TestCase):
         with patch.dict(os.environ, {}, clear=True), \
              patch.object(backend_mod.FirecrackerBottleBackend,
                           "is_host_capable", classmethod(lambda cls: True)), \
-             patch.object(backend_mod, "_backends", {
+             patch.object(backend_mod, "_BACKENDS", {
                  "macos-container": _FakeBackend("macos-container", False),
                  "firecracker": _FakeBackend("firecracker", False),
                  "docker": _FakeBackend("docker", True),
@@ -133,7 +133,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_backends",
+            backend_mod, "_BACKENDS",
             {"docker": _FakeBackend([a]), "firecracker": _FakeBackend([b])},
         ):
             self.assertEqual([a, b], enumerate_active_agents())
@@ -167,7 +167,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_backends",
+            backend_mod, "_BACKENDS",
             {
                 "docker": _FakeBackend([newer, tie_b]),
                 "firecracker": _FakeBackend([missing_metadata, tie_a]),
@@ -187,7 +187,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return []
 
         with patch.object(
-            backend_mod, "_backends",
+            backend_mod, "_BACKENDS",
             {"docker": _FakeBackend(), "firecracker": _FakeBackend()},
         ):
             self.assertEqual([], enumerate_active_agents())
@@ -218,7 +218,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_backends",
+            backend_mod, "_BACKENDS",
             {
                 "docker": _FakeBackend([present], available=True),
                 "firecracker": _FakeBackend([hidden], available=False),
@@ -234,7 +234,7 @@ class TestHasBackend(unittest.TestCase):
                 return False
 
         with patch.object(
-            backend_mod, "_backends", {"docker": _FakeBackend()},
+            backend_mod, "_BACKENDS", {"docker": _FakeBackend()},
         ):
             from bot_bottle.backend import has_backend
             self.assertFalse(has_backend("docker"))

tests/unit/test_docker_launch_committed_image.py

@@ -91,6 +91,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
         with mock.patch.object(launch_mod, "read_committed_image", return_value=committed_tag), \
                 mock.patch.object(launch_mod.docker_mod, "image_exists", return_value=image_present), \
                 mock.patch.object(launch_mod.docker_mod, "build_image", side_effect=_build), \
+                mock.patch.object(launch_mod.docker_mod, "verify_agent_image"), \
                 mock.patch.object(launch_mod, "launch_consolidated", return_value=_CTX), \
                 mock.patch.object(launch_mod, "teardown_consolidated"), \
                 mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \

tests/unit/test_docker_launch_teardown.py

@@ -94,6 +94,7 @@ class TestTeardownWarning(unittest.TestCase):
         )
 
         with mock.patch.object(launch_mod.docker_mod, "build_image"), \
+             mock.patch.object(launch_mod.docker_mod, "verify_agent_image"), \
              mock.patch.object(launch_mod, "launch_consolidated", return_value=ctx), \
              mock.patch.object(launch_mod, "teardown_consolidated"), \
              mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \

tests/unit/test_docker_util_image.py

@@ -29,7 +29,7 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
 class TestCommitContainer(unittest.TestCase):
     def test_runs_docker_commit(self):
         with patch.object(
-            docker_mod, "run_docker", return_value=_ok(),
+            docker_mod.subprocess, "run", return_value=_ok(),
         ) as run, patch.object(docker_mod, "info"):
             docker_mod.commit_container(
                 "bot-bottle-dev-abc12",
@@ -47,7 +47,7 @@ class TestCommitContainer(unittest.TestCase):
 
     def test_dies_on_docker_commit_failure(self):
         with patch.object(
-            docker_mod, "run_docker", return_value=_fail("No such container"),
+            docker_mod.subprocess, "run", return_value=_fail("No such container"),
         ), patch.object(
             docker_mod, "die", side_effect=SystemExit("die"),
         ) as die:
@@ -58,7 +58,7 @@ class TestCommitContainer(unittest.TestCase):
 
     def test_die_message_includes_image_tag(self):
         with patch.object(
-            docker_mod, "run_docker", return_value=_fail("boom"),
+            docker_mod.subprocess, "run", return_value=_fail("boom"),
         ), patch.object(
             docker_mod, "die", side_effect=SystemExit("die"),
         ) as die:

tests/unit/test_macos_container_launch.py

@@ -357,12 +357,17 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
             launch.container_mod, "image_exists", return_value=False,
         ), patch.object(
             launch.container_mod, "build_image", side_effect=fake_build,
-        ):
+        ), patch.object(
+            launch.container_mod, "verify_agent_image",
+        ) as verify:
             updated = launch._build_images(plan)
 
         self.assertEqual("bot-bottle-agent:latest", updated.image)
         self.assertEqual(2, len(calls))
         self.assertEqual("bot-bottle-agent:latest", calls[1][0])
+        verify.assert_called_once_with(
+            "bot-bottle-agent:latest", ("claude", "--version"),
+        )
 
 
 if __name__ == "__main__":

tests/unit/test_orchestrator_lifecycle.py

@@ -10,8 +10,10 @@ from unittest.mock import MagicMock, Mock, patch
 
 from bot_bottle.orchestrator.lifecycle import (
     ORCHESTRATOR_NAME,
+    ORCHESTRATOR_SOURCE_HASH_LABEL,
     OrchestratorService,
     OrchestratorStartError,
+    _source_hash,
 )
 from tests.unit import use_bottle_root
 
@@ -28,6 +30,10 @@ def _health(status: int) -> MagicMock:
     return m
 
 
+def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
+    return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
+
+
 class TestOrchestratorService(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
@@ -46,25 +52,67 @@ class TestOrchestratorService(unittest.TestCase):
         with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
             self.assertFalse(self.svc.is_healthy())
 
-    def test_ensure_running_always_recreates_orchestrator(self) -> None:
-        # Even when a control plane is already healthy, the orchestrator is
-        # recreated so bind-mounted code changes take effect (its process
-        # won't reload). The gateway is ensured too.
-        run = Mock(return_value=Mock(returncode=0, stderr=""))
+    def test_ensure_running_noop_when_healthy_and_source_unchanged(self) -> None:
+        # A healthy control plane already running the *current* bind-mounted
+        # source is left alone — recreating it on every launch would drop
+        # every other active bottle's in-memory egress tokens (#381).
+        current = _source_hash(self.svc._repo_root)
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout=ORCHESTRATOR_NAME)
+            if argv[:2] == ["docker", "inspect"]:
+                return _proc(stdout=current)
+            return _proc()
+
         with patch(_URLOPEN, return_value=_health(200)), \
-                patch(_GATEWAY) as gw_cls, patch(_RUN, run), patch(_SLEEP):
+                patch(_GATEWAY) as gw_cls, patch(_RUN, side_effect=fake), patch(_SLEEP):
             self.assertEqual(self.svc.url, self.svc.ensure_running())
         gw_cls.return_value.ensure_running.assert_called()   # gateway kept up
-        runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
-        self.assertEqual(1, len(runs))                       # orchestrator recreated
+        runs = [c for c in calls if c[:2] == ["docker", "run"]]
+        rms = [c for c in calls if c[:3] == ["docker", "rm", "--force"] and ORCHESTRATOR_NAME in c]
+        self.assertEqual([], runs)   # not recreated
+        self.assertEqual([], rms)
+
+    def test_ensure_running_recreates_when_source_changed(self) -> None:
+        # Healthy, but the running container's label doesn't match the
+        # current source hash (a real code change) — recreate so it takes
+        # effect, same as the gateway's image-staleness check.
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout=ORCHESTRATOR_NAME)
+            if argv[:2] == ["docker", "inspect"]:
+                return _proc(stdout="stale-hash")
+            return _proc()
+
+        with patch(_URLOPEN, return_value=_health(200)), \
+                patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
+            self.assertEqual(self.svc.url, self.svc.ensure_running())
+        runs = [c for c in calls if c[:2] == ["docker", "run"]]
+        self.assertEqual(1, len(runs))
         self.assertIn(ORCHESTRATOR_NAME, runs[0])
+        # the fresh container is labeled with the current hash, not the stale one
+        current = _source_hash(self.svc._repo_root)
+        self.assertIn(f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current}", runs[0])
 
     def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
-        run = Mock(return_value=Mock(returncode=0, stderr=""))
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout="")   # not running
+            return _proc()
+
         with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
-                patch(_GATEWAY), patch(_RUN, run), patch(_SLEEP):
+                patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
             self.assertEqual(self.svc.url, self.svc.ensure_running())
-        runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
+        runs = [c for c in calls if c[:2] == ["docker", "run"]]
         self.assertEqual(1, len(runs))
         argv = runs[0]
         self.assertIn(ORCHESTRATOR_NAME, argv)