feat(orchestrator): slice 3 — real Docker launch broker (#352)

The first concrete LaunchBroker, proving the orchestrator -> backend seam
on the cheapest backend (the sidecar bundle is already containers):

  * orchestrator/docker_broker.py — DockerBroker runs a container on a
    verified launch (`docker run --detach --name <bottle> --label ...
    <image_ref>`) and removes it on teardown (`docker rm --force`,
    idempotent on an already-absent container). The argv is built only from
    the request's static ids/flags, so nothing free-form reaches docker;
    provenance/schema verification is inherited from LaunchBroker.submit.
  * __main__.py gains `--broker {stub,docker}` so the harness can drive real
    containers.

Slice 3 launches a single container from image_ref (the seam); the full
agent + sidecar bundle is a later slice.

Tests: unit (docker mocked) — argv from static fields, launch/teardown call
the right commands, missing-image and docker-failure raise, teardown
idempotent on missing, forged token never touches docker; integration
(gated on a reachable daemon) — launch creates a real container, teardown
removes it. 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_broker.py

@@ -0,0 +1,55 @@
+"""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, op: str, **kw: str) -> None:
+        req = LaunchRequest(op=op, bottle_id=self.bottle_id, **kw)
+        self.broker.submit(sign_request(req, self.secret))
+
+    def test_launch_creates_then_teardown_removes(self) -> None:
+        self._submit("launch", image_ref=IMAGE)
+        self.assertTrue(self._exists(), "container should exist after launch")
+        self._submit("teardown")
+        self.assertFalse(self._exists(), "container should be gone after teardown")
+
+
+if __name__ == "__main__":
+    unittest.main()

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