feat(orchestrator): slice 5 — build the consolidated sidecar bundle image (#352)

Answers "where do we build the consolidated sidecar": nowhere, until now.

  * sidecar.py — `Sidecar.ensure_built()` (default no-op) + `DockerSidecar`
    now defaults its image to the real bundle (`bot-bottle-sidecars`) and
    `ensure_built()` builds it from `Dockerfile.sidecars` when
    `docker image inspect` shows it's missing (no-op when present or when no
    dockerfile is configured, e.g. a pre-pulled image). `image_exists()`
    added.
  * service.py — `ensure_sidecar()` now builds then runs.
  * __main__.py — `--sidecar` runs the consolidated bundle (build-if-missing).

Scope note: this builds + launches the bundle *container*; making the
running instance functional across bottles needs the per-bottle,
source-IP-keyed multi-tenant config + registration/reload, and routing
agent bottles to it — the next slices (added to PRD 0070's roadmap).

Tests: unit (docker mocked) — image_exists, ensure_built builds when
missing / no-op when present / no-op without a dockerfile / raises on build
failure; ensure_sidecar builds-then-runs; integration (gated, no heavy
build) — image_exists reflects real docker state. Full suite green (only
pre-existing /bin/sleep errors).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck

bot_bottle/docker_cmd.py

@@ -0,0 +1,27 @@
+"""Lean, framework-free `docker` subprocess primitive.
+
+Deliberately a top-level module with a single stdlib import so it can be
+reused from anywhere without cost. It is intentionally *not* placed in
+`backend.docker.util`: importing that module runs `backend/__init__.py`,
+which eagerly loads all three bottle backends (docker + firecracker +
+macos) plus the manifest/egress/git-gate/supervise framework — ~76 modules
+— which would drag the whole backend layer into the deliberately-lean
+orchestrator. This primitive stays free of that so both the orchestrator's
+docker components and (in time) `backend.docker.util` can share it."""
+
+from __future__ import annotations
+
+import subprocess
+
+
+def run_docker(argv: list[str]) -> subprocess.CompletedProcess[str]:
+    """Run a `docker` command, capturing stdout/stderr as text. Never raises
+    on a non-zero exit — callers inspect `returncode` / `stderr` so they can
+    stay fail-closed or tolerate idempotent no-ops (e.g. removing an
+    already-absent container)."""
+    return subprocess.run(
+        argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False,
+    )
+
+
+__all__ = ["run_docker"]

bot_bottle/orchestrator/__init__.py

@@ -10,10 +10,15 @@ backend-neutral "consolidation core" that needs no VM packaging:
   * `broker`        — the signed, structured launch-request contract + a
                       `LaunchBroker` (stub for the harness) that verifies
                       provenance before acting.
+  * `sidecar`       — the consolidated per-host sidecar: a `Sidecar`
+                      lifecycle contract (idempotent singleton) + a
+                      `DockerSidecar` impl. One sidecar shared by all
+                      bottles instead of one per bottle.
   * `service`       — the `Orchestrator`: owns the registry, brokers the
-                      launch lifecycle (launch/teardown), attributes.
+                      launch lifecycle (launch/teardown), manages the
+                      shared sidecar, attributes.
   * `control_plane` — the HTTP control-plane RPC (launch / teardown /
-                      list / attribute / health).
+                      list / attribute / sidecar / health).
 
 The actual backend-native launch (a real docker/firecracker broker) and
 the egress/git/supervise data plane land in later slices once this core is
@@ -32,6 +37,8 @@ from .broker import (
     sign_request,
     verify_request,
 )
+from .docker_broker import DockerBroker, DockerBrokerError
+from .sidecar import DockerSidecar, Sidecar, SidecarError
 from .service import Orchestrator
 from .control_plane import ControlPlaneServer, dispatch, make_server
 
@@ -43,6 +50,11 @@ __all__ = [
     "LaunchBroker",
     "LaunchRequest",
     "StubBroker",
+    "DockerBroker",
+    "DockerBrokerError",
+    "Sidecar",
+    "DockerSidecar",
+    "SidecarError",
     "sign_request",
     "verify_request",
     "Orchestrator",

bot_bottle/orchestrator/__main__.py

@@ -16,10 +16,12 @@ import secrets
 from pathlib import Path
 
 from .. import log
-from .broker import StubBroker
+from .broker import LaunchBroker, StubBroker
 from .control_plane import make_server
+from .docker_broker import DockerBroker
 from .registry import RegistryStore, default_db_path
 from .service import Orchestrator
+from .sidecar import DockerSidecar, Sidecar
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -31,16 +33,32 @@ def main(argv: list[str] | None = None) -> int:
         "--db", type=Path, default=None,
         help=f"registry DB path (default: {default_db_path()})",
     )
+    parser.add_argument(
+        "--broker", choices=("stub", "docker"), default="stub",
+        help="launch broker: 'stub' records requests; 'docker' runs containers",
+    )
+    parser.add_argument(
+        "--sidecar", action="store_true",
+        help="run one consolidated per-host sidecar bundle (build-if-missing)",
+    )
     args = parser.parse_args(argv)
 
     registry = RegistryStore(args.db)
     registry.migrate()
 
-    # Dev-harness wiring: an ephemeral signing secret + a stub broker that
-    # records launches instead of starting anything. A real backend broker
-    # (docker, then firecracker) drops in here later.
+    # An ephemeral signing secret ties the orchestrator (signer) to its
+    # broker (verifier). 'stub' records launches instead of starting
+    # anything; 'docker' runs real containers (firecracker drops in later).
     secret = secrets.token_bytes(32)
-    orchestrator = Orchestrator(registry, StubBroker(secret), secret)
+    broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret)
+    sidecar: Sidecar | None = DockerSidecar() if args.sidecar else None
+    orchestrator = Orchestrator(registry, broker, secret, sidecar)
+
+    # One persistent per-host sidecar, shared by every bottle: build the
+    # bundle image if missing, then bring the singleton up (idempotent).
+    if sidecar is not None:
+        orchestrator.ensure_sidecar()
+        log.info("consolidated sidecar ensured", context={"name": sidecar.name})
 
     server = make_server(orchestrator, host=args.host, port=args.port)
     bound_host, bound_port = server.server_address[0], server.server_address[1]

bot_bottle/orchestrator/control_plane.py

@@ -5,6 +5,7 @@ The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
 vsock / unix-socket portability caveats):
 
     GET    /health              -> 200 {"status": "ok"}
+    GET    /sidecar             -> 200 {"configured", ["name", "running"]}
     GET    /bottles             -> 200 {"bottles": [ <redacted record>, ... ]}
     POST   /bottles             -> 201 {"bottle_id", "identity_token"}    (launch)
                                    body: {"source_ip", ["image_ref"], ["metadata"]}
@@ -58,6 +59,9 @@ def dispatch(  # pylint: disable=too-many-return-statements
     if method == "GET" and route == "/health":
         return 200, {"status": "ok"}
 
+    if method == "GET" and route == "/sidecar":
+        return 200, orch.sidecar_status()
+
     if method == "GET" and route == "/bottles":
         return 200, {"bottles": [r.redacted() for r in orch.registry.all()]}
 

bot_bottle/orchestrator/docker_broker.py

@@ -0,0 +1,83 @@
+"""Docker launch broker (PRD 0070) — the first *real* LaunchBroker.
+
+On a verified launch request it starts a Docker container; on teardown it
+removes it. This proves the orchestrator -> backend seam on the cheapest
+backend (the sidecar bundle is already containers). Only the request's
+static ids/flags reach `docker`, so nothing free-form crosses the boundary.
+
+Slice 3 launches a single container from the request's `image_ref`, named
+after the bottle id and labelled for cleanup. Wiring the full agent +
+sidecar bundle (networks, mounts, the consolidated sidecar) is a later
+slice — this is the seam, not the finished launcher.
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+from ..docker_cmd import run_docker
+from .broker import LaunchBroker, LaunchRequest
+
+CONTAINER_PREFIX = "bot-bottle-orch-"
+BOTTLE_ID_LABEL = "bot-bottle-bottle-id"
+
+
+class DockerBrokerError(Exception):
+    """A brokered docker launch/teardown failed (non-zero `docker` exit)."""
+
+
+def container_name(bottle_id: str) -> str:
+    """Deterministic container name for a bottle."""
+    return f"{CONTAINER_PREFIX}{bottle_id}"
+
+
+def run_argv(req: LaunchRequest) -> list[str]:
+    """`docker run` argv for a launch request — built only from its static
+    fields, so it can't be coerced into an arbitrary command."""
+    return [
+        "docker", "run", "--detach",
+        "--name", container_name(req.bottle_id),
+        "--label", f"{BOTTLE_ID_LABEL}={req.bottle_id}",
+        req.image_ref,
+    ]
+
+
+def rm_argv(req: LaunchRequest) -> list[str]:
+    """`docker rm --force` argv to tear a bottle's container down."""
+    return ["docker", "rm", "--force", container_name(req.bottle_id)]
+
+
+class DockerBroker(LaunchBroker):
+    """A `LaunchBroker` that runs / removes a Docker container per bottle.
+    Provenance + schema verification are inherited from `LaunchBroker`."""
+
+    def _docker(self, argv: list[str]) -> subprocess.CompletedProcess[str]:
+        return run_docker(argv)
+
+    def _launch(self, req: LaunchRequest) -> None:
+        if not req.image_ref:
+            raise DockerBrokerError(f"launch request for {req.bottle_id} has no image_ref")
+        proc = self._docker(run_argv(req))
+        if proc.returncode != 0:
+            raise DockerBrokerError(
+                f"docker run failed for {req.bottle_id}: {proc.stderr.strip()}"
+            )
+
+    def _teardown(self, req: LaunchRequest) -> None:
+        proc = self._docker(rm_argv(req))
+        # Idempotent: an already-absent container is a successful teardown.
+        if proc.returncode != 0 and "No such container" not in proc.stderr:
+            raise DockerBrokerError(
+                f"docker rm failed for {req.bottle_id}: {proc.stderr.strip()}"
+            )
+
+
+__all__ = [
+    "DockerBroker",
+    "DockerBrokerError",
+    "container_name",
+    "run_argv",
+    "rm_argv",
+    "CONTAINER_PREFIX",
+    "BOTTLE_ID_LABEL",
+]

bot_bottle/orchestrator/service.py

@@ -19,17 +19,25 @@ from __future__ import annotations
 
 from .broker import LaunchBroker, LaunchRequest, sign_request
 from .registry import BottleRecord, RegistryStore
+from .sidecar import Sidecar
 
 
 class Orchestrator:
-    """Owns the registry + brokers launches. Backend-neutral."""
+    """Owns the registry + brokers launches, and manages the single
+    consolidated per-host sidecar. Backend-neutral (broker and sidecar
+    abstract the backend-native pieces)."""
 
     def __init__(
-        self, registry: RegistryStore, broker: LaunchBroker, sign_secret: bytes
+        self,
+        registry: RegistryStore,
+        broker: LaunchBroker,
+        sign_secret: bytes,
+        sidecar: Sidecar | None = None,
     ) -> None:
         self.registry = registry
         self._broker = broker
         self._secret = sign_secret
+        self._sidecar = sidecar
 
     def launch_bottle(
         self,
@@ -72,5 +80,24 @@ class Orchestrator:
         """Fail-closed attribution (delegates to the registry)."""
         return self.registry.attribute(source_ip, identity_token)
 
+    # --- consolidated sidecar ----------------------------------------------
+
+    def ensure_sidecar(self) -> None:
+        """Ensure the single per-host sidecar is built and up (idempotent).
+        No-op when no sidecar is configured."""
+        if self._sidecar is not None:
+            self._sidecar.ensure_built()
+            self._sidecar.ensure_running()
+
+    def sidecar_status(self) -> dict[str, object]:
+        """Report the shared sidecar for the control plane / console."""
+        if self._sidecar is None:
+            return {"configured": False}
+        return {
+            "configured": True,
+            "name": self._sidecar.name,
+            "running": self._sidecar.is_running(),
+        }
+
 
 __all__ = ["Orchestrator"]

bot_bottle/orchestrator/sidecar.py

@@ -0,0 +1,139 @@
+"""The consolidated per-host sidecar (PRD 0070).
+
+The core consolidation win: **one** persistent sidecar per host, shared by
+every bottle, instead of a sidecar bundle per bottle. It's safe to share
+because the attribution invariant (source IP + identity token, see
+`registry`) lets the sidecar attribute each request to the right bottle —
+so per-bottle policy lives in one long-lived process keyed on who's calling.
+
+`Sidecar` is the backend-neutral lifecycle contract (mirrors `LaunchBroker`):
+ensure the single instance is up, report it, tear it down. `DockerSidecar`
+is the docker implementation; a firecracker sidecar VM slots in later.
+
+The defining behaviour is **idempotent singleton**: `ensure_running` starts
+the instance if absent and is a no-op if it's already up, so N bottle
+launches never spawn N sidecars.
+"""
+
+from __future__ import annotations
+
+import abc
+import os
+from pathlib import Path
+
+from ..docker_cmd import run_docker
+
+SIDECAR_NAME = "bot-bottle-orch-sidecar"
+SIDECAR_LABEL = "bot-bottle-orch-sidecar=1"
+
+# The real sidecar-bundle image + its Dockerfile. Kept as a local constant
+# rather than imported from backend.docker.sidecar_bundle, which would drag
+# the whole backend layer into the lean orchestrator (see #359); unify when
+# that lands. Env override matches the backend's BOT_BOTTLE_SIDECAR_IMAGE.
+SIDECAR_BUNDLE_IMAGE = os.environ.get("BOT_BOTTLE_SIDECAR_IMAGE", "bot-bottle-sidecars:latest")
+SIDECAR_DOCKERFILE = "Dockerfile.sidecars"
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+class SidecarError(Exception):
+    """The shared sidecar failed to build/start/stop (non-zero `docker` exit)."""
+
+
+class Sidecar(abc.ABC):
+    """Lifecycle of the single per-host sidecar. Backend-neutral."""
+
+    name: str
+
+    def ensure_built(self) -> None:
+        """Ensure the sidecar's image / rootfs exists, building it if needed.
+        Default: nothing to build (e.g. a stub or a pre-pulled image)."""
+        return
+
+    @abc.abstractmethod
+    def ensure_running(self) -> None:
+        """Start the sidecar if it isn't already up. Idempotent: a no-op
+        when it's already running (that's the whole point — one per host).
+        Assumes the image exists — call `ensure_built()` first."""
+
+    @abc.abstractmethod
+    def is_running(self) -> bool:
+        """True iff the sidecar instance is currently up."""
+
+    @abc.abstractmethod
+    def stop(self) -> None:
+        """Remove the sidecar. Idempotent — absent is success."""
+
+
+class DockerSidecar(Sidecar):
+    """The consolidated sidecar as a single, fixed-name Docker container.
+
+    `image_ref` defaults to the real sidecar-bundle image; `ensure_built`
+    builds it from `Dockerfile.sidecars` when it's missing. (Note: slice 5
+    builds + launches the bundle container; wiring its per-bottle,
+    source-IP-keyed config is a later slice — see PRD 0070.)"""
+
+    def __init__(
+        self,
+        image_ref: str = SIDECAR_BUNDLE_IMAGE,
+        *,
+        name: str = SIDECAR_NAME,
+        build_context: Path | None = None,
+        dockerfile: str | None = SIDECAR_DOCKERFILE,
+    ) -> None:
+        self.image_ref = image_ref
+        self.name = name
+        self._build_context = build_context or _REPO_ROOT
+        self._dockerfile = dockerfile
+
+    def image_exists(self) -> bool:
+        return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0
+
+    def ensure_built(self) -> None:
+        """Build the bundle image from its Dockerfile when it's missing.
+        No-op when the image is already present, or when no dockerfile is
+        configured (e.g. a pre-pulled image)."""
+        if self._dockerfile is None or self.image_exists():
+            return
+        proc = run_docker([
+            "docker", "build",
+            "-t", self.image_ref,
+            "-f", str(self._build_context / self._dockerfile),
+            str(self._build_context),
+        ])
+        if proc.returncode != 0:
+            raise SidecarError(f"sidecar image build failed: {proc.stderr.strip()}")
+
+    def is_running(self) -> bool:
+        proc = run_docker([
+            "docker", "ps",
+            "--filter", f"name=^{self.name}$",
+            "--filter", "status=running",
+            "--format", "{{.Names}}",
+        ])
+        return self.name in proc.stdout.split()
+
+    def ensure_running(self) -> None:
+        if self.is_running():
+            return
+        # Clear any stale (stopped) container holding the fixed name, then
+        # start fresh. `rm --force` on an absent name is a tolerated no-op.
+        run_docker(["docker", "rm", "--force", self.name])
+        proc = run_docker([
+            "docker", "run", "--detach",
+            "--name", self.name,
+            "--label", SIDECAR_LABEL,
+            self.image_ref,
+        ])
+        if proc.returncode != 0:
+            raise SidecarError(f"sidecar failed to start: {proc.stderr.strip()}")
+
+    def stop(self) -> None:
+        proc = run_docker(["docker", "rm", "--force", self.name])
+        if proc.returncode != 0 and "No such container" not in proc.stderr:
+            raise SidecarError(f"sidecar failed to stop: {proc.stderr.strip()}")
+
+
+__all__ = [
+    "Sidecar", "DockerSidecar", "SidecarError",
+    "SIDECAR_NAME", "SIDECAR_LABEL", "SIDECAR_BUNDLE_IMAGE",
+]

tests/integration/test_orchestrator_docker_broker.py

@@ -0,0 +1,56 @@
+"""Integration: the Docker launch broker starts and removes a real container.
+
+Gated on a reachable Docker daemon (skips cleanly otherwise). Uses a tiny
+image; the container may exit immediately — we only assert it exists after
+launch and is gone after teardown.
+"""
+
+from __future__ import annotations
+
+import secrets
+import subprocess
+import unittest
+
+from bot_bottle.orchestrator.broker import LaunchRequest, sign_request
+from bot_bottle.orchestrator.docker_broker import DockerBroker, container_name
+from tests._docker import skip_unless_docker
+
+IMAGE = "busybox"
+
+
+@skip_unless_docker()
+class TestDockerBrokerIntegration(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+        self.broker = DockerBroker(self.secret)
+        self.bottle_id = "itest" + secrets.token_hex(4)
+        self.name = container_name(self.bottle_id)
+        self.addCleanup(
+            lambda: subprocess.run(
+                ["docker", "rm", "--force", self.name],
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+            )
+        )
+
+    def _exists(self) -> bool:
+        proc = subprocess.run(
+            ["docker", "ps", "-a", "--filter", f"name=^{self.name}$",
+             "--format", "{{.Names}}"],
+            stdout=subprocess.PIPE, text=True, check=False,
+        )
+        return self.name in proc.stdout.split()
+
+    def _submit(self, req: LaunchRequest) -> None:
+        self.broker.submit(sign_request(req, self.secret))
+
+    def test_launch_creates_then_teardown_removes(self) -> None:
+        self._submit(
+            LaunchRequest(op="launch", bottle_id=self.bottle_id, image_ref=IMAGE)
+        )
+        self.assertTrue(self._exists(), "container should exist after launch")
+        self._submit(LaunchRequest(op="teardown", bottle_id=self.bottle_id))
+        self.assertFalse(self._exists(), "container should be gone after teardown")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/integration/test_orchestrator_docker_sidecar.py

@@ -0,0 +1,45 @@
+"""Integration: the consolidated Docker sidecar is an idempotent singleton.
+
+Gated on a reachable Docker daemon. Uses a unique name so it never
+collides with a real per-host sidecar or a leftover from another run.
+"""
+
+from __future__ import annotations
+
+import secrets
+import subprocess
+import unittest
+
+from bot_bottle.orchestrator.sidecar import DockerSidecar
+from tests._docker import skip_unless_docker
+
+IMAGE = "busybox"
+
+
+@skip_unless_docker()
+class TestDockerSidecarIntegration(unittest.TestCase):
+    def setUp(self) -> None:
+        self.name = "bot-bottle-orch-sidecar-itest-" + secrets.token_hex(4)
+        self.sc = DockerSidecar(IMAGE, name=self.name)
+        self.addCleanup(self.sc.stop)
+
+    def _count(self) -> int:
+        proc = subprocess.run(
+            ["docker", "ps", "-a", "--filter", f"name=^{self.name}$",
+             "--format", "{{.Names}}"],
+            stdout=subprocess.PIPE, text=True, check=False,
+        )
+        return sum(1 for n in proc.stdout.split() if n == self.name)
+
+    def test_ensure_is_idempotent_singleton(self) -> None:
+        self.sc.ensure_running()
+        self.assertEqual(1, self._count())
+        # A second ensure must not spawn a second container — one per host.
+        self.sc.ensure_running()
+        self.assertEqual(1, self._count())
+        self.sc.stop()
+        self.assertEqual(0, self._count())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/integration/test_orchestrator_docker_sidecar_build.py

@@ -0,0 +1,36 @@
+"""Integration: DockerSidecar.image_exists reflects real docker state.
+
+Gated on a reachable Docker daemon. Deliberately does NOT build the full
+sidecar bundle (a heavy, slow image build) — it exercises the `image_exists`
+primitive that `ensure_built` gates on, against a tiny pulled image and a
+name that can't exist.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import unittest
+
+from bot_bottle.orchestrator.sidecar import DockerSidecar
+from tests._docker import skip_unless_docker
+
+IMAGE = "busybox"
+
+
+@skip_unless_docker()
+class TestDockerSidecarImageExists(unittest.TestCase):
+    def test_image_exists_true_for_present_false_for_absent(self) -> None:
+        # Ensure the tiny image is present (build_if_missing is disabled here
+        # so this never triggers a Dockerfile.sidecars build).
+        subprocess.run(
+            ["docker", "pull", IMAGE],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+        )
+        self.assertTrue(DockerSidecar(IMAGE, dockerfile=None).image_exists())
+        self.assertFalse(
+            DockerSidecar("bot-bottle-nonexistent:doesnotexist", dockerfile=None).image_exists()
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_control_plane.py

@@ -118,6 +118,11 @@ class TestDispatch(unittest.TestCase):
         status, _ = dispatch(self.orch, "GET", "/health/", b"")
         self.assertEqual(200, status)
 
+    def test_sidecar_status_unconfigured(self) -> None:
+        status, payload = dispatch(self.orch, "GET", "/sidecar", b"")
+        self.assertEqual(200, status)
+        self.assertEqual(False, payload["configured"])
+
 
 class TestServerRoundTrip(unittest.TestCase):
     def test_http_register_health_attribute(self) -> None:

tests/unit/test_orchestrator_docker_broker.py

@@ -0,0 +1,100 @@
+"""Unit tests for the Docker launch broker (PRD 0070). Docker is mocked."""
+
+from __future__ import annotations
+
+import secrets
+import unittest
+from unittest.mock import Mock, patch
+
+from bot_bottle.orchestrator.broker import (
+    BrokerAuthError,
+    LaunchRequest,
+    sign_request,
+)
+from bot_bottle.orchestrator.docker_broker import (
+    BOTTLE_ID_LABEL,
+    DockerBroker,
+    DockerBrokerError,
+    container_name,
+    rm_argv,
+    run_argv,
+)
+
+
+class TestArgv(unittest.TestCase):
+    def test_run_argv_uses_only_static_fields(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
+        argv = run_argv(req)
+        self.assertEqual(["docker", "run"], argv[:2])
+        self.assertIn("--name", argv)
+        self.assertIn(container_name("b1"), argv)
+        self.assertIn(f"{BOTTLE_ID_LABEL}=b1", argv)
+        self.assertEqual("busybox", argv[-1])  # image is the terminal arg
+
+    def test_rm_argv(self) -> None:
+        req = LaunchRequest(op="teardown", bottle_id="b1")
+        self.assertEqual(["docker", "rm", "--force", container_name("b1")], rm_argv(req))
+
+    def test_container_name_is_prefixed(self) -> None:
+        name = container_name("b1")
+        self.assertTrue(name.startswith("bot-bottle-orch-"))
+        self.assertTrue(name.endswith("b1"))
+
+
+class TestDockerBroker(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+        self.broker = DockerBroker(self.secret)
+
+    def _submit(self, req: LaunchRequest) -> None:
+        self.broker.submit(sign_request(req, self.secret))
+
+    def test_launch_invokes_docker_run(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
+        with patch.object(self.broker, "_docker", return_value=Mock(returncode=0, stderr="")) as m:
+            self._submit(req)
+        m.assert_called_once()
+        self.assertEqual(run_argv(req), m.call_args.args[0])
+
+    def test_launch_without_image_raises_and_skips_docker(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", image_ref="")
+        with patch.object(self.broker, "_docker") as m:
+            with self.assertRaises(DockerBrokerError):
+                self._submit(req)
+            m.assert_not_called()
+
+    def test_launch_docker_failure_raises(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
+        with patch.object(self.broker, "_docker", return_value=Mock(returncode=1, stderr="boom")):
+            with self.assertRaises(DockerBrokerError):
+                self._submit(req)
+
+    def test_teardown_invokes_docker_rm(self) -> None:
+        req = LaunchRequest(op="teardown", bottle_id="b1")
+        with patch.object(self.broker, "_docker", return_value=Mock(returncode=0, stderr="")) as m:
+            self._submit(req)
+        self.assertEqual(rm_argv(req), m.call_args.args[0])
+
+    def test_teardown_is_idempotent_on_missing_container(self) -> None:
+        req = LaunchRequest(op="teardown", bottle_id="b1")
+        absent = Mock(returncode=1, stderr="Error: No such container: bot-bottle-orch-b1")
+        with patch.object(self.broker, "_docker", return_value=absent):
+            self._submit(req)  # must not raise
+
+    def test_teardown_other_failure_raises(self) -> None:
+        req = LaunchRequest(op="teardown", bottle_id="b1")
+        with patch.object(self.broker, "_docker", return_value=Mock(returncode=1, stderr="daemon down")):
+            with self.assertRaises(DockerBrokerError):
+                self._submit(req)
+
+    def test_forged_token_never_touches_docker(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
+        forged = sign_request(req, secrets.token_bytes(16))
+        with patch.object(self.broker, "_docker") as m:
+            with self.assertRaises(BrokerAuthError):
+                self.broker.submit(forged)
+            m.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_service.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker
 from bot_bottle.orchestrator.registry import RegistryStore
 from bot_bottle.orchestrator.service import Orchestrator
+from bot_bottle.orchestrator.sidecar import Sidecar
 
 
 class _FailingBroker(LaunchBroker):
@@ -23,6 +24,29 @@ class _FailingBroker(LaunchBroker):
         pass
 
 
+class _FakeSidecar(Sidecar):
+    """In-memory sidecar for wiring tests."""
+
+    def __init__(self) -> None:
+        self.name = "fake-sidecar"
+        self.ensured = 0
+        self.built = 0
+        self._running = False
+
+    def ensure_built(self) -> None:
+        self.built += 1
+
+    def ensure_running(self) -> None:
+        self.ensured += 1
+        self._running = True
+
+    def is_running(self) -> bool:
+        return self._running
+
+    def stop(self) -> None:
+        self._running = False
+
+
 class TestOrchestrator(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
@@ -68,6 +92,25 @@ class TestOrchestrator(unittest.TestCase):
             orch.launch_bottle("10.243.0.9")
         self.assertEqual([], self.store.all())  # no orphan
 
+    def test_sidecar_unconfigured_by_default(self) -> None:
+        self.assertEqual({"configured": False}, self.orch.sidecar_status())
+        self.orch.ensure_sidecar()  # no-op, must not raise
+
+    def test_sidecar_wired_and_ensured(self) -> None:
+        sc = _FakeSidecar()
+        orch = Orchestrator(self.store, self.broker, self.secret, sidecar=sc)
+        self.assertEqual(
+            {"configured": True, "name": "fake-sidecar", "running": False},
+            orch.sidecar_status(),
+        )
+        orch.ensure_sidecar()
+        self.assertEqual(1, sc.built)   # ensure_sidecar builds first,
+        self.assertEqual(1, sc.ensured)  # then runs
+        self.assertEqual(
+            {"configured": True, "name": "fake-sidecar", "running": True},
+            orch.sidecar_status(),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_sidecar.py

@@ -0,0 +1,129 @@
+"""Unit tests for the consolidated Docker sidecar (PRD 0070). Docker mocked."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import Mock, patch
+
+from bot_bottle.orchestrator.sidecar import (
+    SIDECAR_NAME,
+    DockerSidecar,
+    SidecarError,
+)
+
+_RUN_DOCKER = "bot_bottle.orchestrator.sidecar.run_docker"
+
+
+def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
+    return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+class TestDockerSidecar(unittest.TestCase):
+    def setUp(self) -> None:
+        self.sc = DockerSidecar("bot-bottle-sidecars:latest")
+
+    def test_default_name(self) -> None:
+        self.assertEqual(SIDECAR_NAME, self.sc.name)
+
+    def test_is_running_reads_docker_ps(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name + "\n")):
+            self.assertTrue(self.sc.is_running())
+        with patch(_RUN_DOCKER, return_value=_proc(stdout="")):
+            self.assertFalse(self.sc.is_running())
+
+    def test_ensure_running_is_noop_when_already_up(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name)) as m:
+            self.sc.ensure_running()
+        # Only the is_running() ps probe — no rm / run.
+        self.assertEqual(1, m.call_count)
+
+    def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()
+
+        runs = [c for c in calls if c[:2] == ["docker", "run"]]
+        self.assertEqual(1, len(runs))
+        self.assertIn(self.sc.name, runs[0])
+        self.assertIn("bot-bottle-sidecars:latest", runs[0])
+
+    def test_ensure_running_raises_on_docker_failure(self) -> None:
+        def fake(argv: list[str]) -> Mock:
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout="")
+            if argv[:2] == ["docker", "run"]:
+                return _proc(returncode=1, stderr="boom")
+            return _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            with self.assertRaises(SidecarError):
+                self.sc.ensure_running()
+
+    def test_stop_is_idempotent_on_missing(self) -> None:
+        absent = _proc(returncode=1, stderr="Error: No such container: x")
+        with patch(_RUN_DOCKER, return_value=absent):
+            self.sc.stop()  # must not raise
+
+    def test_stop_raises_on_other_failure(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="daemon down")):
+            with self.assertRaises(SidecarError):
+                self.sc.stop()
+
+
+class TestDockerSidecarBuild(unittest.TestCase):
+    def setUp(self) -> None:
+        self.sc = DockerSidecar()  # defaults to the real bundle image + dockerfile
+
+    def test_image_exists_reads_docker_inspect(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=0)):
+            self.assertTrue(self.sc.image_exists())
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1)):
+            self.assertFalse(self.sc.image_exists())
+
+    def test_ensure_built_is_noop_when_image_present(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=0)) as m:
+            self.sc.ensure_built()
+        self.assertEqual(1, m.call_count)  # only the image-inspect probe
+
+    def test_ensure_built_builds_from_dockerfile_when_missing(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:3] == ["docker", "image", "inspect"]:
+                return _proc(returncode=1)  # missing
+            return _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_built()
+
+        builds = [c for c in calls if c[:2] == ["docker", "build"]]
+        self.assertEqual(1, len(builds))
+        self.assertIn(self.sc.image_ref, builds[0])
+        self.assertIn("-f", builds[0])
+        self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0]))
+
+    def test_ensure_built_noop_when_no_dockerfile(self) -> None:
+        sc = DockerSidecar("busybox", dockerfile=None)
+        with patch(_RUN_DOCKER) as m:
+            sc.ensure_built()
+        m.assert_not_called()
+
+    def test_ensure_built_raises_on_build_failure(self) -> None:
+        def fake(argv: list[str]) -> Mock:
+            if argv[:3] == ["docker", "image", "inspect"]:
+                return _proc(returncode=1)
+            return _proc(returncode=1, stderr="build boom")
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            with self.assertRaises(SidecarError):
+                self.sc.ensure_built()
+
+
+if __name__ == "__main__":
+    unittest.main()