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
@@ -33,6 +38,7 @@ from .broker import (
     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
 
@@ -46,6 +52,9 @@ __all__ = [
     "StubBroker",
     "DockerBroker",
     "DockerBrokerError",
+    "Sidecar",
+    "DockerSidecar",
+    "SidecarError",
     "sign_request",
     "verify_request",
     "Orchestrator",

bot_bottle/orchestrator/__main__.py

@@ -21,6 +21,7 @@ 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:
@@ -36,6 +37,10 @@ def main(argv: list[str] | None = None) -> int:
         "--broker", choices=("stub", "docker"), default="stub",
         help="launch broker: 'stub' records requests; 'docker' runs containers",
     )
+    parser.add_argument(
+        "--sidecar-image", default=None,
+        help="if set, run one consolidated per-host sidecar from this image",
+    )
     args = parser.parse_args(argv)
 
     registry = RegistryStore(args.db)
@@ -46,7 +51,13 @@ def main(argv: list[str] | None = None) -> int:
     # anything; 'docker' runs real containers (firecracker drops in later).
     secret = secrets.token_bytes(32)
     broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret)
-    orchestrator = Orchestrator(registry, broker, secret)
+    sidecar: Sidecar | None = DockerSidecar(args.sidecar_image) if args.sidecar_image else None
+    orchestrator = Orchestrator(registry, broker, secret, sidecar)
+
+    # One persistent per-host sidecar, shared by every bottle (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

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import subprocess
 
+from ..docker_cmd import run_docker
 from .broker import LaunchBroker, LaunchRequest
 
 CONTAINER_PREFIX = "bot-bottle-orch-"
@@ -51,10 +52,7 @@ class DockerBroker(LaunchBroker):
     Provenance + schema verification are inherited from `LaunchBroker`."""
 
     def _docker(self, argv: list[str]) -> subprocess.CompletedProcess[str]:
-        return subprocess.run(
-            argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            text=True, check=False,
-        )
+        return run_docker(argv)
 
     def _launch(self, req: LaunchRequest) -> None:
         if not req.image_ref:

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,23 @@ class Orchestrator:
         """Fail-closed attribution (delegates to the registry)."""
         return self.registry.attribute(source_ip, identity_token)
 
+    # --- consolidated sidecar ----------------------------------------------
+
+    def ensure_sidecar(self) -> None:
+        """Bring the single per-host sidecar up (idempotent). No-op when no
+        sidecar is configured."""
+        if self._sidecar is not None:
+            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,88 @@
+"""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
+
+from ..docker_cmd import run_docker
+
+SIDECAR_NAME = "bot-bottle-orch-sidecar"
+SIDECAR_LABEL = "bot-bottle-orch-sidecar=1"
+
+
+class SidecarError(Exception):
+    """The shared sidecar failed to start/stop (non-zero `docker` exit)."""
+
+
+class Sidecar(abc.ABC):
+    """Lifecycle of the single per-host sidecar. Backend-neutral."""
+
+    name: str
+
+    @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)."""
+
+    @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."""
+
+    def __init__(self, image_ref: str, *, name: str = SIDECAR_NAME) -> None:
+        self.image_ref = image_ref
+        self.name = name
+
+    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"]

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/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_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,25 @@ 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._running = False
+
+    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 +88,24 @@ 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.ensured)
+        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,79 @@
+"""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()
+
+
+if __name__ == "__main__":
+    unittest.main()