feat(orchestrator): slice 2 — launch lifecycle + signed launch-broker (#352)

Second slice of PRD 0070, still a backend-neutral dev-harness:

  * orchestrator/broker.py — the launch-broker contract. A LaunchRequest is
    structured (static ids/flags only — bottle id, pool slot, a
    content-addressed image_ref; never a path/argv) and signed as a compact
    HS256 JWT so the broker verifies PROVENANCE before acting: a compromised
    co-located component can't forge a launch without the shared secret.
    verify_request is fail-closed (bad sig / malformed / off-schema -> raise).
    Stdlib only (no runtime deps). Ships a StubBroker that records verified
    requests for the harness/tests.
  * orchestrator/service.py — the Orchestrator: owns the registry and brokers
    the lifecycle. launch_bottle mints the bottle + sends a signed launch,
    rolling the registry entry back if the launch fails (no orphans);
    teardown_bottle brokers teardown then deregisters; attribute delegates.
  * control_plane.py — POST /bottles now launches, DELETE tears down (both go
    through the Orchestrator + broker). dispatch/server take an Orchestrator.
  * __main__.py wires an ephemeral secret + StubBroker for the harness.

Tests: broker sign/verify round-trip, tamper/wrong-secret/malformed/off-schema
rejection, StubBroker fail-closed; Orchestrator launch->registry->attribute,
teardown, rollback-on-broker-failure; control-plane updated for launch/teardown.
Full suite green (only the pre-existing /bin/sleep errors); harness does
launch -> attribute -> teardown over HTTP.

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

bot_bottle/orchestrator/__init__.py

@@ -7,23 +7,45 @@ backend-neutral "consolidation core" that needs no VM packaging:
 
   * `registry`      — the SQLite runtime-state store + fail-closed
                       attribution (source IP + per-bottle identity token).
-  * `control_plane` — the HTTP control-plane RPC (register / deregister /
-                      list / attribute / health), with live reload.
+  * `broker`        — the signed, structured launch-request contract + a
+                      `LaunchBroker` (stub for the harness) that verifies
+                      provenance before acting.
+  * `service`       — the `Orchestrator`: owns the registry, brokers the
+                      launch lifecycle (launch/teardown), attributes.
+  * `control_plane` — the HTTP control-plane RPC (launch / teardown /
+                      list / attribute / health).
 
-Launch/teardown, the launch broker, and the actual egress/git/supervise
-data plane land in later slices once this core is proven (see the PRD's
-sequencing: plain-process dev-harness -> docker orchestrator -> firecracker).
+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
+proven (see the PRD sequencing: plain-process dev-harness -> docker
+orchestrator -> firecracker).
 """
 
 from __future__ import annotations
 
 from .registry import BottleRecord, RegistryStore, new_identity_token
+from .broker import (
+    BrokerAuthError,
+    LaunchBroker,
+    LaunchRequest,
+    StubBroker,
+    sign_request,
+    verify_request,
+)
+from .service import Orchestrator
 from .control_plane import ControlPlaneServer, dispatch, make_server
 
 __all__ = [
     "BottleRecord",
     "RegistryStore",
     "new_identity_token",
+    "BrokerAuthError",
+    "LaunchBroker",
+    "LaunchRequest",
+    "StubBroker",
+    "sign_request",
+    "verify_request",
+    "Orchestrator",
     "ControlPlaneServer",
     "dispatch",
     "make_server",

bot_bottle/orchestrator/__main__.py

@@ -12,11 +12,14 @@ container packaging. Wrapping this exact service in a backend-native unit
 from __future__ import annotations
 
 import argparse
+import secrets
 from pathlib import Path
 
 from .. import log
+from .broker import StubBroker
 from .control_plane import make_server
 from .registry import RegistryStore, default_db_path
+from .service import Orchestrator
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -33,7 +36,13 @@ def main(argv: list[str] | None = None) -> int:
     registry = RegistryStore(args.db)
     registry.migrate()
 
-    server = make_server(registry, host=args.host, port=args.port)
+    # 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.
+    secret = secrets.token_bytes(32)
+    orchestrator = Orchestrator(registry, StubBroker(secret), secret)
+
+    server = make_server(orchestrator, host=args.host, port=args.port)
     bound_host, bound_port = server.server_address[0], server.server_address[1]
     log.info(
         "orchestrator control plane listening",

bot_bottle/orchestrator/broker.py

@@ -0,0 +1,176 @@
+"""Launch broker contract (PRD 0070).
+
+A VM/container can't spawn its own host-network siblings, so the
+orchestrator brokers agent launches through a small privileged component.
+The request it sends is:
+
+  * **structured** — static flags + ids only (bottle id, pool slot, a
+    content-addressed image ref), never a free-form path or argv, so a
+    request can't be coerced into launching an arbitrary payload; and
+  * **signed** — wrapped as a compact JWS/JWT the broker verifies before
+    acting, so a *compromised co-located component* (an agent-facing
+    sidecar, say) can't forge a launch. This is the concrete form of the
+    "structured requests only" rule in the PRD security review.
+
+Signing is **HS256** over a secret shared by the orchestrator (signer) and
+the broker (verifier) — appropriate here because both are trusted host
+components provisioned together; the secret is exactly what an
+agent-facing component does not have. (Stdlib only — the project takes no
+runtime deps; a cross-host deployment could swap in an asymmetric alg.)
+"""
+
+from __future__ import annotations
+
+import abc
+import base64
+import hashlib
+import hmac
+import json
+import secrets
+import time
+from dataclasses import dataclass
+
+_JWT_HEADER = {"alg": "HS256", "typ": "JWT"}
+_ALLOWED_OPS = ("launch", "teardown")
+
+
+class BrokerAuthError(Exception):
+    """A broker request failed provenance or schema verification —
+    bad/absent signature, malformed token, or a payload that doesn't match
+    the fixed launch-request shape. Fail-closed: the broker must not act."""
+
+
+@dataclass(frozen=True)
+class LaunchRequest:
+    """The structured, un-coercible launch/teardown request. Ids + flags
+    only — `image_ref` is a content-addressed id, never a path/argv."""
+
+    op: str  # one of _ALLOWED_OPS
+    bottle_id: str
+    source_ip: str = ""
+    image_ref: str = ""
+    slot: int | None = None
+
+
+# --- minimal JWS/JWT (HS256), stdlib only ----------------------------------
+
+def _b64url(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+def _b64url_decode(text: str) -> bytes:
+    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
+
+
+def _mac(signing_input: str, secret: bytes) -> str:
+    digest = hmac.new(secret, signing_input.encode("ascii"), hashlib.sha256).digest()
+    return _b64url(digest)
+
+
+def sign_request(req: LaunchRequest, secret: bytes) -> str:
+    """Serialize `req` to signed-JWT form. Adds a per-signature `jti`
+    (replay id) and `iat` (issued-at)."""
+    claims: dict[str, object] = {
+        "op": req.op,
+        "bottle_id": req.bottle_id,
+        "source_ip": req.source_ip,
+        "image_ref": req.image_ref,
+        "slot": req.slot,
+        "jti": secrets.token_hex(8),
+        "iat": int(time.time()),
+    }
+    header = _b64url(json.dumps(_JWT_HEADER, sort_keys=True, separators=(",", ":")).encode())
+    payload = _b64url(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode())
+    signing_input = f"{header}.{payload}"
+    return f"{signing_input}.{_mac(signing_input, secret)}"
+
+
+def verify_request(token: str, secret: bytes) -> LaunchRequest:
+    """Verify the signature and the request shape; return the parsed
+    request. Raises `BrokerAuthError` on any failure (fail-closed)."""
+    parts = token.split(".")
+    if len(parts) != 3:
+        raise BrokerAuthError("malformed token")
+    header_b, payload_b, sig = parts
+    if not hmac.compare_digest(_mac(f"{header_b}.{payload_b}", secret), sig):
+        raise BrokerAuthError("bad signature")
+    try:
+        header = json.loads(_b64url_decode(header_b))
+        claims = json.loads(_b64url_decode(payload_b))
+    except (ValueError, json.JSONDecodeError) as e:
+        raise BrokerAuthError("malformed token payload") from e
+    if not isinstance(header, dict) or header.get("alg") != "HS256":
+        raise BrokerAuthError("unexpected header/alg")
+    if not isinstance(claims, dict):
+        raise BrokerAuthError("claims are not an object")
+
+    op = claims.get("op")
+    bottle_id = claims.get("bottle_id")
+    source_ip = claims.get("source_ip", "")
+    image_ref = claims.get("image_ref", "")
+    slot = claims.get("slot")
+    if (
+        not isinstance(op, str) or op not in _ALLOWED_OPS
+        or not isinstance(bottle_id, str) or not bottle_id
+        or not isinstance(source_ip, str) or not isinstance(image_ref, str)
+        or not (slot is None or isinstance(slot, int))
+    ):
+        raise BrokerAuthError("request does not match the launch-request schema")
+    return LaunchRequest(
+        op=op, bottle_id=bottle_id, source_ip=source_ip, image_ref=image_ref, slot=slot
+    )
+
+
+# --- the broker itself ------------------------------------------------------
+
+class LaunchBroker(abc.ABC):
+    """Verifies a signed request came from the orchestrator, then performs
+    the backend-native launch/teardown. Subclasses implement `_launch` /
+    `_teardown`; verification is shared and fail-closed."""
+
+    def __init__(self, secret: bytes) -> None:
+        self._secret = secret
+
+    def submit(self, token: str) -> LaunchRequest:
+        """Verify `token` and perform its op. Returns the verified request;
+        raises `BrokerAuthError` if provenance/shape fails."""
+        req = verify_request(token, self._secret)
+        if req.op == "launch":
+            self._launch(req)
+        else:
+            self._teardown(req)
+        return req
+
+    @abc.abstractmethod
+    def _launch(self, req: LaunchRequest) -> None:
+        ...
+
+    @abc.abstractmethod
+    def _teardown(self, req: LaunchRequest) -> None:
+        ...
+
+
+class StubBroker(LaunchBroker):
+    """Dev-harness broker: records verified requests without launching
+    anything. Exercises the full sign -> verify -> act contract in-process."""
+
+    def __init__(self, secret: bytes) -> None:
+        super().__init__(secret)
+        self.launched: list[LaunchRequest] = []
+        self.torn_down: list[LaunchRequest] = []
+
+    def _launch(self, req: LaunchRequest) -> None:
+        self.launched.append(req)
+
+    def _teardown(self, req: LaunchRequest) -> None:
+        self.torn_down.append(req)
+
+
+__all__ = [
+    "BrokerAuthError",
+    "LaunchRequest",
+    "LaunchBroker",
+    "StubBroker",
+    "sign_request",
+    "verify_request",
+]

bot_bottle/orchestrator/control_plane.py

@@ -2,23 +2,25 @@
 
 The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
 **HTTP** — the universal transport chosen in 0070 (works on every host; no
-vsock / unix-socket portability caveats). Endpoints mutate the live
-registry, so register/deregister are the *live-reload* path — no restart:
+vsock / unix-socket portability caveats):
 
     GET    /health              -> 200 {"status": "ok"}
     GET    /bottles             -> 200 {"bottles": [ <redacted record>, ... ]}
-    POST   /bottles             -> 201 {"bottle_id", "identity_token"}
-                                   body: {"source_ip", ["bottle_id"], ["metadata"]}
-    DELETE /bottles/<bottle_id> -> 200 {"deregistered": true} | 404
+    POST   /bottles             -> 201 {"bottle_id", "identity_token"}    (launch)
+                                   body: {"source_ip", ["image_ref"], ["metadata"]}
+    DELETE /bottles/<bottle_id> -> 200 {"torn_down": true} | 404          (teardown)
     POST   /attribute           -> 200 {"bottle_id"} | 403
                                    body: {"source_ip", "identity_token"}
 
+`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
+tear down) the bottle in the registry AND broker the backend-native launch
+via the orchestrator. Register/deregister without a launch are internal to
+`Orchestrator`, not exposed here.
+
 Routing/handling is the pure function `dispatch()` so it is unit-testable
 without a socket; `Handler` / `ControlPlaneServer` / `make_server` are a
-thin stdlib adapter around it.
-
-Note the listing redacts identity tokens — they are never returned except
-once, to the caller that registers the bottle.
+thin stdlib adapter around it. Listing redacts identity tokens — they are
+returned only once, to the caller that launches the bottle.
 """
 
 from __future__ import annotations
@@ -30,7 +32,7 @@ import socketserver
 import typing
 from urllib.parse import urlsplit
 
-from .registry import RegistryStore
+from .service import Orchestrator
 
 # JSON body payload type (parsed request / rendered response).
 Json = dict[str, object]
@@ -47,17 +49,17 @@ def _parse_json_object(body: bytes) -> Json:
 
 
 def dispatch(  # pylint: disable=too-many-return-statements
-    registry: RegistryStore, method: str, path: str, body: bytes
+    orch: Orchestrator, method: str, path: str, body: bytes
 ) -> tuple[int, Json]:
     """Route one control-plane request to a (status, payload) pair. Pure —
-    no I/O beyond the registry — so it is fully testable without a socket."""
+    no I/O beyond the orchestrator — so it is fully testable without a socket."""
     route = urlsplit(path).path.rstrip("/") or "/"
 
     if method == "GET" and route == "/health":
         return 200, {"status": "ok"}
 
     if method == "GET" and route == "/bottles":
-        return 200, {"bottles": [r.redacted() for r in registry.all()]}
+        return 200, {"bottles": [r.redacted() for r in orch.registry.all()]}
 
     if method == "POST" and route == "/bottles":
         try:
@@ -67,19 +69,19 @@ def dispatch(  # pylint: disable=too-many-return-statements
         source_ip = data.get("source_ip")
         if not isinstance(source_ip, str) or not source_ip:
             return 400, {"error": "source_ip (string) is required"}
-        bottle_id = data.get("bottle_id")
+        image_ref = data.get("image_ref")
         metadata = data.get("metadata")
-        rec = registry.register(
+        rec = orch.launch_bottle(
             source_ip,
-            bottle_id=bottle_id if isinstance(bottle_id, str) else None,
+            image_ref=image_ref if isinstance(image_ref, str) else "",
             metadata=metadata if isinstance(metadata, str) else "",
         )
         return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
 
     if method == "DELETE" and route.startswith("/bottles/"):
         bottle_id = route[len("/bottles/"):]
-        if registry.deregister(bottle_id):
-            return 200, {"deregistered": True}
+        if orch.teardown_bottle(bottle_id):
+            return 200, {"torn_down": True}
         return 404, {"error": "no such bottle"}
 
     if method == "POST" and route == "/attribute":
@@ -91,7 +93,7 @@ def dispatch(  # pylint: disable=too-many-return-statements
         token = data.get("identity_token")
         if not isinstance(source_ip, str) or not isinstance(token, str):
             return 400, {"error": "source_ip and identity_token (strings) required"}
-        rec = registry.attribute(source_ip, token)
+        rec = orch.attribute(source_ip, token)
         if rec is None:
             return 403, {"error": "unattributed"}
         return 200, {"bottle_id": rec.bottle_id}
@@ -114,7 +116,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
         assert isinstance(server, ControlPlaneServer)
         length = int(self.headers.get("Content-Length") or 0)
         body = self.rfile.read(length) if length > 0 else b""
-        status, payload = dispatch(server.registry, method, self.path, body)
+        status, payload = dispatch(server.orchestrator, method, self.path, body)
         data = json.dumps(payload).encode()
         self.send_response(status)
         self.send_header("Content-Type", "application/json")
@@ -133,22 +135,22 @@ class Handler(http.server.BaseHTTPRequestHandler):
 
 
 class ControlPlaneServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
-    """Threading HTTP server that carries the registry for its handlers."""
+    """Threading HTTP server that carries the orchestrator for its handlers."""
 
     daemon_threads = True
     allow_reuse_address = True
 
-    def __init__(self, address: tuple[str, int], registry: RegistryStore) -> None:
-        self.registry = registry
+    def __init__(self, address: tuple[str, int], orchestrator: Orchestrator) -> None:
+        self.orchestrator = orchestrator
         super().__init__(address, Handler)
 
 
 def make_server(
-    registry: RegistryStore, host: str = "127.0.0.1", port: int = 0
+    orchestrator: Orchestrator, host: str = "127.0.0.1", port: int = 0
 ) -> ControlPlaneServer:
     """Build (but do not start) a control-plane server. `port=0` binds an
     ephemeral port — read `server.server_address` for the actual one."""
-    return ControlPlaneServer((host, port), registry)
+    return ControlPlaneServer((host, port), orchestrator)
 
 
 __all__ = ["dispatch", "Handler", "ControlPlaneServer", "make_server", "Json"]

bot_bottle/orchestrator/service.py

@@ -0,0 +1,76 @@
+"""The per-host orchestrator core (PRD 0070).
+
+`Orchestrator` is the single backend-neutral object the control plane talks
+to: it owns the registry (runtime state) and brokers agent launches. It
+never branches on backend — the `LaunchBroker` abstracts the backend-native
+launch, so this same object drives docker / firecracker / apple once a real
+broker is wired in.
+
+Launch lifecycle:
+
+  * `launch_bottle` mints the bottle (registry: source IP + identity
+    token), sends a *signed, structured* launch request through the broker,
+    and returns the record. If the broker rejects/fails, the registry entry
+    is rolled back so a failed launch leaves no orphan.
+  * `teardown_bottle` sends a signed teardown request, then deregisters.
+"""
+
+from __future__ import annotations
+
+from .broker import LaunchBroker, LaunchRequest, sign_request
+from .registry import BottleRecord, RegistryStore
+
+
+class Orchestrator:
+    """Owns the registry + brokers launches. Backend-neutral."""
+
+    def __init__(
+        self, registry: RegistryStore, broker: LaunchBroker, sign_secret: bytes
+    ) -> None:
+        self.registry = registry
+        self._broker = broker
+        self._secret = sign_secret
+
+    def launch_bottle(
+        self,
+        source_ip: str,
+        *,
+        image_ref: str = "",
+        slot: int | None = None,
+        metadata: str = "",
+    ) -> BottleRecord:
+        """Register a bottle and broker its launch. Rolls the registry entry
+        back if the launch doesn't take, so a failure leaves no orphan."""
+        rec = self.registry.register(source_ip, metadata=metadata)
+        req = LaunchRequest(
+            op="launch",
+            bottle_id=rec.bottle_id,
+            source_ip=source_ip,
+            image_ref=image_ref,
+            slot=slot,
+        )
+        launched = False
+        try:
+            self._broker.submit(sign_request(req, self._secret))
+            launched = True
+        finally:
+            if not launched:
+                self.registry.deregister(rec.bottle_id)
+        return rec
+
+    def teardown_bottle(self, bottle_id: str) -> bool:
+        """Broker teardown then deregister. False if the bottle is unknown."""
+        rec = self.registry.get(bottle_id)
+        if rec is None:
+            return False
+        req = LaunchRequest(op="teardown", bottle_id=bottle_id, source_ip=rec.source_ip)
+        self._broker.submit(sign_request(req, self._secret))
+        self.registry.deregister(bottle_id)
+        return True
+
+    def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
+        """Fail-closed attribution (delegates to the registry)."""
+        return self.registry.attribute(source_ip, identity_token)
+
+
+__all__ = ["Orchestrator"]