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

tests/unit/test_orchestrator_broker.py

@@ -0,0 +1,86 @@
+"""Unit tests for the orchestrator launch broker (PRD 0070)."""
+
+from __future__ import annotations
+
+import secrets
+import unittest
+
+from bot_bottle.orchestrator.broker import (
+    BrokerAuthError,
+    LaunchRequest,
+    StubBroker,
+    sign_request,
+    verify_request,
+)
+
+
+class TestSignVerify(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+
+    def test_launch_round_trip(self) -> None:
+        req = LaunchRequest(
+            op="launch", bottle_id="b1", source_ip="10.243.0.1",
+            image_ref="sha256:abc", slot=3,
+        )
+        self.assertEqual(req, verify_request(sign_request(req, self.secret), self.secret))
+
+    def test_teardown_round_trip(self) -> None:
+        req = LaunchRequest(op="teardown", bottle_id="b1")
+        got = verify_request(sign_request(req, self.secret), self.secret)
+        self.assertEqual(("teardown", "b1"), (got.op, got.bottle_id))
+
+    def test_wrong_secret_rejected(self) -> None:
+        token = sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret)
+        with self.assertRaises(BrokerAuthError):
+            verify_request(token, secrets.token_bytes(16))
+
+    def test_tampered_payload_rejected(self) -> None:
+        token = sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret)
+        header, payload, sig = token.split(".")
+        flipped = payload[:-1] + ("A" if payload[-1] != "A" else "B")
+        with self.assertRaises(BrokerAuthError):
+            verify_request(f"{header}.{flipped}.{sig}", self.secret)
+
+    def test_malformed_tokens_rejected(self) -> None:
+        for bad in ("", "a.b", "a.b.c.d", "not-a-token"):
+            with self.assertRaises(BrokerAuthError):
+                verify_request(bad, self.secret)
+
+    def test_unknown_op_rejected_despite_valid_signature(self) -> None:
+        # A correctly-signed token whose op isn't in the allow-list must
+        # still be refused — the schema guard is independent of provenance.
+        token = sign_request(LaunchRequest(op="rm-rf", bottle_id="b1"), self.secret)
+        with self.assertRaises(BrokerAuthError):
+            verify_request(token, self.secret)
+
+
+class TestStubBroker(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+        self.broker = StubBroker(self.secret)
+
+    def test_submit_launch_records_verified_request(self) -> None:
+        req = LaunchRequest(op="launch", bottle_id="b1", source_ip="10.243.0.1")
+        got = self.broker.submit(sign_request(req, self.secret))
+        self.assertEqual(req, got)
+        self.assertEqual(["b1"], [r.bottle_id for r in self.broker.launched])
+        self.assertEqual([], self.broker.torn_down)
+
+    def test_submit_teardown_records(self) -> None:
+        self.broker.submit(
+            sign_request(LaunchRequest(op="teardown", bottle_id="b1"), self.secret)
+        )
+        self.assertEqual(["b1"], [r.bottle_id for r in self.broker.torn_down])
+
+    def test_submit_forged_token_is_fail_closed(self) -> None:
+        forged = sign_request(
+            LaunchRequest(op="launch", bottle_id="b1"), secrets.token_bytes(16)
+        )
+        with self.assertRaises(BrokerAuthError):
+            self.broker.submit(forged)
+        self.assertEqual([], self.broker.launched)  # nothing acted on
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_control_plane.py

@@ -7,53 +7,62 @@ server tests), plus one real-socket round-trip to prove the handler wiring.
 from __future__ import annotations
 
 import json
+import secrets
 import tempfile
 import threading
 import unittest
 import urllib.request
 from pathlib import Path
 
+from bot_bottle.orchestrator.broker import StubBroker
 from bot_bottle.orchestrator.control_plane import dispatch, make_server
 from bot_bottle.orchestrator.registry import RegistryStore
+from bot_bottle.orchestrator.service import Orchestrator
 
 
 def _body(obj: object) -> bytes:
     return json.dumps(obj).encode()
 
 
+def _orchestrator(db_path: Path) -> Orchestrator:
+    store = RegistryStore(db_path)
+    store.migrate()
+    secret = secrets.token_bytes(16)
+    return Orchestrator(store, StubBroker(secret), secret)
+
+
 class TestDispatch(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
-        self.store = RegistryStore(Path(self._tmp.name) / "r.db")
-        self.store.migrate()
+        self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
 
     def tearDown(self) -> None:
         self._tmp.cleanup()
 
     def test_health(self) -> None:
-        status, payload = dispatch(self.store, "GET", "/health", b"")
+        status, payload = dispatch(self.orch, "GET", "/health", b"")
         self.assertEqual(200, status)
         self.assertEqual("ok", payload["status"])
 
     def test_register_returns_id_and_token(self) -> None:
         status, payload = dispatch(
-            self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})
+            self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})
         )
         self.assertEqual(201, status)
         self.assertTrue(payload["bottle_id"])
         self.assertTrue(payload["identity_token"])
 
     def test_register_requires_source_ip(self) -> None:
-        status, _ = dispatch(self.store, "POST", "/bottles", _body({}))
+        status, _ = dispatch(self.orch, "POST", "/bottles", _body({}))
         self.assertEqual(400, status)
 
     def test_register_rejects_bad_json(self) -> None:
-        status, _ = dispatch(self.store, "POST", "/bottles", b"{not json")
+        status, _ = dispatch(self.orch, "POST", "/bottles", b"{not json")
         self.assertEqual(400, status)
 
     def test_list_redacts_identity_token(self) -> None:
-        dispatch(self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}))
-        status, payload = dispatch(self.store, "GET", "/bottles", b"")
+        dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}))
+        status, payload = dispatch(self.orch, "GET", "/bottles", b"")
         self.assertEqual(200, status)
         bottles = payload["bottles"]
         assert isinstance(bottles, list)
@@ -65,48 +74,48 @@ class TestDispatch(unittest.TestCase):
 
     def test_attribute_ok_and_forbidden(self) -> None:
         _, reg = dispatch(
-            self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.5"})
+            self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.5"})
         )
         token = reg["identity_token"]
         ok_status, ok = dispatch(
-            self.store, "POST", "/attribute",
+            self.orch, "POST", "/attribute",
             _body({"source_ip": "10.243.0.5", "identity_token": token}),
         )
         self.assertEqual(200, ok_status)
         self.assertEqual(reg["bottle_id"], ok["bottle_id"])
 
         bad_status, _ = dispatch(
-            self.store, "POST", "/attribute",
+            self.orch, "POST", "/attribute",
             _body({"source_ip": "10.243.0.5", "identity_token": "nope"}),
         )
         self.assertEqual(403, bad_status)
 
     def test_attribute_requires_both_fields(self) -> None:
         status, _ = dispatch(
-            self.store, "POST", "/attribute", _body({"source_ip": "10.243.0.5"})
+            self.orch, "POST", "/attribute", _body({"source_ip": "10.243.0.5"})
         )
         self.assertEqual(400, status)
 
     def test_delete(self) -> None:
         _, reg = dispatch(
-            self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})
+            self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})
         )
         bid = reg["bottle_id"]
-        status, payload = dispatch(self.store, "DELETE", f"/bottles/{bid}", b"")
+        status, payload = dispatch(self.orch, "DELETE", f"/bottles/{bid}", b"")
         self.assertEqual(200, status)
-        self.assertEqual(True, payload["deregistered"])
-        self.assertEqual([], self.store.all())
+        self.assertEqual(True, payload["torn_down"])
+        self.assertEqual([], self.orch.registry.all())
 
     def test_delete_missing_404(self) -> None:
-        status, _ = dispatch(self.store, "DELETE", "/bottles/ghost", b"")
+        status, _ = dispatch(self.orch, "DELETE", "/bottles/ghost", b"")
         self.assertEqual(404, status)
 
     def test_unknown_route_404(self) -> None:
-        status, _ = dispatch(self.store, "GET", "/nope", b"")
+        status, _ = dispatch(self.orch, "GET", "/nope", b"")
         self.assertEqual(404, status)
 
     def test_trailing_slash_normalized(self) -> None:
-        status, _ = dispatch(self.store, "GET", "/health/", b"")
+        status, _ = dispatch(self.orch, "GET", "/health/", b"")
         self.assertEqual(200, status)
 
 
@@ -114,9 +123,8 @@ class TestServerRoundTrip(unittest.TestCase):
     def test_http_register_health_attribute(self) -> None:
         tmp = tempfile.TemporaryDirectory()
         self.addCleanup(tmp.cleanup)
-        store = RegistryStore(Path(tmp.name) / "r.db")
-        store.migrate()
-        server = make_server(store, "127.0.0.1", 0)
+        orch = _orchestrator(Path(tmp.name) / "r.db")
+        server = make_server(orch, "127.0.0.1", 0)
         self.addCleanup(server.server_close)
         thread = threading.Thread(target=server.serve_forever, daemon=True)
         thread.start()

tests/unit/test_orchestrator_service.py

@@ -0,0 +1,73 @@
+"""Unit tests for the Orchestrator launch lifecycle (PRD 0070)."""
+
+from __future__ import annotations
+
+import secrets
+import tempfile
+import unittest
+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
+
+
+class _FailingBroker(LaunchBroker):
+    """Verifies the token like any broker, then fails the launch — to
+    exercise the orchestrator's registry rollback."""
+
+    def _launch(self, req: LaunchRequest) -> None:
+        raise RuntimeError("launch failed")
+
+    def _teardown(self, req: LaunchRequest) -> None:
+        pass
+
+
+class TestOrchestrator(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.secret = secrets.token_bytes(16)
+        self.store = RegistryStore(Path(self._tmp.name) / "r.db")
+        self.store.migrate()
+        self.broker = StubBroker(self.secret)
+        self.orch = Orchestrator(self.store, self.broker, self.secret)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_launch_registers_and_brokers_signed_request(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.1", image_ref="sha256:abc", slot=2)
+        self.assertIsNotNone(self.store.get(rec.bottle_id))
+        self.assertEqual(1, len(self.broker.launched))
+        req = self.broker.launched[0]
+        self.assertEqual("launch", req.op)
+        self.assertEqual(rec.bottle_id, req.bottle_id)
+        self.assertEqual("10.243.0.1", req.source_ip)
+        self.assertEqual("sha256:abc", req.image_ref)
+        self.assertEqual(2, req.slot)
+
+    def test_launch_then_attribute(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.3")
+        got = self.orch.attribute("10.243.0.3", rec.identity_token)
+        assert got is not None
+        self.assertEqual(rec.bottle_id, got.bottle_id)
+        self.assertIsNone(self.orch.attribute("10.243.0.3", "wrong-token"))
+
+    def test_teardown_brokers_and_deregisters(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.1")
+        self.assertTrue(self.orch.teardown_bottle(rec.bottle_id))
+        self.assertIsNone(self.store.get(rec.bottle_id))
+        self.assertEqual(["teardown"], [r.op for r in self.broker.torn_down])
+
+    def test_teardown_unknown_is_false(self) -> None:
+        self.assertFalse(self.orch.teardown_bottle("ghost"))
+
+    def test_launch_rolls_back_registry_on_broker_failure(self) -> None:
+        orch = Orchestrator(self.store, _FailingBroker(self.secret), self.secret)
+        with self.assertRaises(RuntimeError):
+            orch.launch_bottle("10.243.0.9")
+        self.assertEqual([], self.store.all())  # no orphan
+
+
+if __name__ == "__main__":
+    unittest.main()