feat(orchestrator): slice 4 — consolidated per-host sidecar (#352)

The core consolidation win: one persistent sidecar per host, shared by
every bottle, instead of a sidecar bundle per bottle. Safe to share
because the attribution invariant (source IP + identity token) lets the
sidecar map each request to the right bottle.

  * orchestrator/sidecar.py — a backend-neutral `Sidecar` lifecycle
    contract (mirrors LaunchBroker) + a `DockerSidecar` impl. 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 launches never spawn
    N sidecars; `stop` is idempotent.
  * orchestrator/dockerutil.py — a shared `run_docker` helper; DockerBroker
    now uses it too (DRY with slice 3).
  * service.py — the Orchestrator holds an optional `Sidecar`, exposes
    `ensure_sidecar()` + `sidecar_status()`.
  * control_plane.py — `GET /sidecar` reports it; __main__ gains
    `--sidecar-image` and ensures the single sidecar on startup.

Tests: unit (docker mocked) — is_running, ensure idempotent (no-op when up,
starts when absent), failure raises, stop idempotent; Orchestrator sidecar
wiring/status; control-plane /sidecar; integration (gated) — ensure is a
real idempotent singleton (one container after two ensures), stop removes.
Full suite green (only pre-existing /bin/sleep errors); integration
verified locally against real docker.

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

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()