feat(firecracker): implement consolidated orchestrator launch (PRD 0070)

Replace the per-bottle Docker sidecar bundle with the shared per-host
orchestrator + gateway, mirroring what the Docker backend already has.

- Add `bot_bottle/backend/firecracker/consolidated_launch.py`:
  `_FirecrackerOrchestratorService` (subclasses `OrchestratorService`,
  overrides `_gateway()` to return a `DockerGateway` with host port
  bindings so Firecracker VMs can reach it via their TAP link);
  `launch_consolidated()` registers the bottle by guest IP (attribution
  key), provisions git-gate into the shared gateway, and returns the
  shared CA + orchestrator URL for teardown; `teardown_consolidated()`
  deregisters and cleans up.

- Rewrite `bot_bottle/backend/firecracker/launch.py`: removes the
  per-bottle sidecar bundle (`_start_sidecar_bundle`, `_stage_git_gate`,
  etc.) and `_mint_certs`; wires `launch_consolidated()` instead. The VM
  still sends to `host_tap_ip:PORT` — Docker's PREROUTING DNAT + the nft
  `ct status dnat accept` rule in the forward chain route the traffic to
  the shared gateway container.

- Extend `DockerGateway` with `host_port_bindings` so the Firecracker
  gateway publishes its ports on the host (`0.0.0.0:PORT`).

- Parameterise `OrchestratorService` with `orchestrator_name` /
  `orchestrator_label` so Docker and Firecracker orchestrators can
  coexist on the same host (`bot-bottle-orchestrator` vs
  `bot-bottle-fc-orchestrator`).

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

bot_bottle/backend/__init__.py

@@ -572,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(
@@ -646,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:
@@ -660,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:
@@ -682,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]:
@@ -701,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
 
@@ -717,12 +678,12 @@ __all__ = [
     "BottleCleanupPlan",
     "BottlePlan",
     "BottleSpec",
-    "CommitCancelled",  # pylint: disable=undefined-all-variable
+    "CommitCancelled",
     "ExecResult",
-    "Freezer",  # pylint: disable=undefined-all-variable
+    "Freezer",
     "enumerate_active_agents",
     "get_bottle_backend",
-    "get_freezer",  # pylint: disable=undefined-all-variable
+    "get_freezer",
     "has_backend",
     "known_backend_names",
 ]

bot_bottle/backend/docker/util.py

@@ -8,9 +8,8 @@ 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
 # from ...workspace import WorkspacePlan
 
@@ -32,7 +31,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
 
 
@@ -46,15 +50,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:
@@ -62,7 +71,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:
@@ -82,7 +96,7 @@ def docker_exec(container: str, argv: list[str], *, user: str = "") -> None:
     if user:
         cmd += ["-u", user]
     cmd += [container, *argv]
-    result = run_docker(cmd)
+    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
     if result.returncode != 0:
         die(
             f"docker exec in {container} failed: "
@@ -92,7 +106,9 @@ def docker_exec(container: str, argv: list[str], *, user: str = "") -> None:
 
 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])
+    result = subprocess.run(
+        ["docker", "cp", src, dest], capture_output=True, text=True, check=False,
+    )
     if result.returncode != 0:
         die(f"docker cp {src} -> {dest} failed: "
             f"{(result.stderr or '').strip() or '<no stderr>'}")
@@ -142,7 +158,12 @@ def verify_agent_image(image: str, argv: tuple[str, ...]) -> None:
     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:]])
+    result = subprocess.run(
+        ["docker", "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(
@@ -190,10 +211,22 @@ def verify_agent_image(image: str, argv: tuple[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

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: