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

bot_bottle/orchestrator/__init__.py

@@ -32,6 +32,7 @@ from .broker import (
     sign_request,
     verify_request,
 )
+from .docker_broker import DockerBroker, DockerBrokerError
 from .service import Orchestrator
 from .control_plane import ControlPlaneServer, dispatch, make_server
 
@@ -43,6 +44,8 @@ __all__ = [
     "LaunchBroker",
     "LaunchRequest",
     "StubBroker",
+    "DockerBroker",
+    "DockerBrokerError",
     "sign_request",
     "verify_request",
     "Orchestrator",

bot_bottle/orchestrator/__main__.py

@@ -16,8 +16,9 @@ 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
 
@@ -31,16 +32,21 @@ 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",
+    )
     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)
+    orchestrator = Orchestrator(registry, broker, secret)
 
     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/docker_broker.py

@@ -0,0 +1,85 @@
+"""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 .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 subprocess.run(
+            argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+            text=True, check=False,
+        )
+
+    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",
+]