feat(orchestrator): host control server transport (#468)

Chunk 1 of the host-control-server stack: close the PRD's **transport**
gap. Today LaunchBroker.submit(token) is an in-process method call from
OrchestratorCore; this makes it a real out-of-process service reached
over HTTP.

- host_server.py: the host control server. A pure dispatch() (POST
  /broker verifies a signed token via the existing verify_request +
  _launch/_teardown path, GET /health) wrapped by a thin http.server
  adapter, mirroring orchestrator/server.py. Only the signed token
  crosses the wire; provenance/schema failures are fail-closed 401s that
  never touch the backend, a backend launch failure is a 502.
- broker_client.py: BrokerClient — a drop-in submit(token) that POSTs the
  signed token to the host controller. A 401 re-raises as BrokerAuthError
  so the launch path's rollback is identical local or remote.
- broker.py: SubmitBroker Protocol — the one method OrchestratorCore
  depends on, satisfied by both LaunchBroker and BrokerClient, so the
  core is unchanged (service.py annotation only).
- __main__.py: wire `--broker http` behind the shared-secret env var
  (BOT_BOTTLE_BROKER_SECRET, hex) — a chunk-1 stopgap the durable
  TrustDomain key (chunk 2, #476) replaces.

Tested: pure-dispatch cases, BrokerClient with HTTP mocked, and a
real-socket sign -> POST -> verify -> act round-trip (incl. fail-closed
forged token). pyright clean; pylint 9.86.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/unit/test_orchestrator_broker_client.py

@@ -0,0 +1,84 @@
+"""Unit: orchestrator-side broker client (issue #468, chunk 1). HTTP mocked."""
+
+from __future__ import annotations
+
+import io
+import json
+import unittest
+import urllib.error
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.orchestrator.broker import BrokerAuthError, LaunchRequest
+from bot_bottle.orchestrator.broker_client import BrokerClient, BrokerClientError
+
+_URLOPEN = "bot_bottle.orchestrator.broker_client.urllib.request.urlopen"
+
+
+def _resp(payload: object) -> MagicMock:
+    m = MagicMock()
+    m.__enter__.return_value.read.return_value = json.dumps(payload).encode()
+    return m
+
+
+def _http_error(code: int, payload: object = None) -> urllib.error.HTTPError:
+    body = json.dumps(payload).encode() if payload is not None else b""
+    return urllib.error.HTTPError(
+        "http://host/broker", code, "err", {}, io.BytesIO(body))  # type: ignore[arg-type]
+
+
+class TestSubmit(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = BrokerClient("http://host:8091")
+
+    def test_returns_the_verified_request(self) -> None:
+        echo = {
+            "op": "launch", "bottle_id": "b1", "source_ip": "10.0.0.1",
+            "image_ref": "img", "slot": 3,
+        }
+        with patch(_URLOPEN, return_value=_resp(echo)):
+            got = self.c.submit("tok")
+        self.assertEqual(
+            LaunchRequest(op="launch", bottle_id="b1", source_ip="10.0.0.1",
+                          image_ref="img", slot=3),
+            got,
+        )
+
+    def test_posts_token_to_broker_endpoint(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"op": "teardown", "bottle_id": "b1"})) as m:
+            self.c.submit("signed-token")
+        request = m.call_args.args[0]
+        self.assertEqual("POST", request.get_method())
+        self.assertTrue(request.full_url.endswith("/broker"))
+        self.assertEqual({"token": "signed-token"}, json.loads(request.data))
+
+    def test_401_raises_broker_auth_error(self) -> None:
+        # A fail-closed provenance/schema rejection surfaces as the SAME exception
+        # the in-process broker raises, so the launch path's rollback is identical.
+        with patch(_URLOPEN, side_effect=_http_error(401, {"error": "bad signature"})):
+            with self.assertRaises(BrokerAuthError):
+                self.c.submit("forged")
+
+    def test_502_raises_broker_client_error(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})):
+            with self.assertRaises(BrokerClientError):
+                self.c.submit("tok")
+
+    def test_unreachable_raises_broker_client_error(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(BrokerClientError):
+                self.c.submit("tok")
+
+    def test_malformed_success_body_raises(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"op": "launch"})):  # missing bottle_id
+            with self.assertRaises(BrokerClientError):
+                self.c.submit("tok")
+
+    def test_empty_error_body_is_tolerated(self) -> None:
+        # An error with no readable JSON body still classifies by status code.
+        with patch(_URLOPEN, side_effect=_http_error(401)):
+            with self.assertRaises(BrokerAuthError):
+                self.c.submit("forged")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_host_server.py

@@ -0,0 +1,156 @@
+"""Unit tests for the host control server (issue #468, chunk 1).
+
+Mostly exercises the pure `dispatch()` (socket-free, like the orchestrator
+server tests), plus a real-socket round-trip through `BrokerClient` that proves
+the full sign -> POST -> verify -> act seam over HTTP.
+"""
+
+from __future__ import annotations
+
+import json
+import secrets
+import threading
+import unittest
+
+from bot_bottle.orchestrator.broker import (
+    BrokerAuthError,
+    LaunchBroker,
+    LaunchRequest,
+    StubBroker,
+    sign_request,
+)
+from bot_bottle.orchestrator.broker_client import BrokerClient
+from bot_bottle.orchestrator.host_server import (
+    broker_secret_from_env,
+    dispatch,
+    make_host_server,
+)
+
+
+def _body(obj: object) -> bytes:
+    return json.dumps(obj).encode()
+
+
+class _RaisingBroker(LaunchBroker):
+    """A broker whose backend launch always fails — exercises the 502 path (an
+    operational backend failure, distinct from a fail-closed provenance 401)."""
+
+    def _launch(self, req: LaunchRequest) -> None:
+        raise RuntimeError("docker down")
+
+    def _teardown(self, req: LaunchRequest) -> None:
+        raise RuntimeError("docker down")
+
+
+class TestDispatch(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+        self.broker = StubBroker(self.secret)
+
+    def _token(self, **kwargs: object) -> str:
+        return sign_request(LaunchRequest(**kwargs), self.secret)  # type: ignore[arg-type]
+
+    def test_health(self) -> None:
+        status, payload = dispatch(self.broker, "GET", "/health", b"")
+        self.assertEqual(200, status)
+        self.assertEqual("ok", payload["status"])
+
+    def test_broker_launch_verifies_and_acts(self) -> None:
+        token = self._token(
+            op="launch", bottle_id="b1", source_ip="10.243.0.1",
+            image_ref="img", slot=2,
+        )
+        status, payload = dispatch(self.broker, "POST", "/broker", _body({"token": token}))
+        self.assertEqual(200, status)
+        self.assertEqual("launch", payload["op"])
+        self.assertEqual("b1", payload["bottle_id"])
+        self.assertEqual("img", payload["image_ref"])
+        self.assertEqual(2, payload["slot"])
+        self.assertEqual(["b1"], [r.bottle_id for r in self.broker.launched])
+
+    def test_broker_teardown_acts(self) -> None:
+        token = self._token(op="teardown", bottle_id="b1")
+        status, _ = dispatch(self.broker, "POST", "/broker", _body({"token": token}))
+        self.assertEqual(200, status)
+        self.assertEqual(["b1"], [r.bottle_id for r in self.broker.torn_down])
+
+    def test_forged_token_is_401_and_nothing_acted(self) -> None:
+        forged = sign_request(
+            LaunchRequest(op="launch", bottle_id="b1"), secrets.token_bytes(16))
+        status, payload = dispatch(self.broker, "POST", "/broker", _body({"token": forged}))
+        self.assertEqual(401, status)
+        self.assertIn("broker auth failed", str(payload["error"]))
+        self.assertEqual([], self.broker.launched)  # fail-closed: never launched
+
+    def test_backend_failure_is_502(self) -> None:
+        broker = _RaisingBroker(self.secret)
+        token = self._token(op="launch", bottle_id="b1", image_ref="img")
+        status, payload = dispatch(broker, "POST", "/broker", _body({"token": token}))
+        self.assertEqual(502, status)
+        self.assertIn("backend launch failed", str(payload["error"]))
+
+    def test_missing_token_is_400(self) -> None:
+        status, _ = dispatch(self.broker, "POST", "/broker", _body({}))
+        self.assertEqual(400, status)
+
+    def test_bad_json_is_400(self) -> None:
+        status, _ = dispatch(self.broker, "POST", "/broker", b"{not json")
+        self.assertEqual(400, status)
+
+    def test_unknown_route_404(self) -> None:
+        status, _ = dispatch(self.broker, "GET", "/nope", b"")
+        self.assertEqual(404, status)
+
+    def test_trailing_slash_normalized(self) -> None:
+        status, _ = dispatch(self.broker, "GET", "/health/", b"")
+        self.assertEqual(200, status)
+
+
+class TestBrokerSecretFromEnv(unittest.TestCase):
+    def test_reads_hex_secret(self) -> None:
+        s = secrets.token_bytes(16)
+        self.assertEqual(s, broker_secret_from_env({"BOT_BOTTLE_BROKER_SECRET": s.hex()}))
+
+    def test_unset_is_none(self) -> None:
+        self.assertIsNone(broker_secret_from_env({}))
+
+    def test_invalid_hex_is_none(self) -> None:
+        self.assertIsNone(broker_secret_from_env({"BOT_BOTTLE_BROKER_SECRET": "not-hex"}))
+
+
+class TestSeamRoundTrip(unittest.TestCase):
+    """The whole point of chunk 1: a request signed by the orchestrator side is
+    POSTed to a real host control server, verified there, and acted on — over
+    HTTP, not an in-process call."""
+
+    def _serve(self, broker: LaunchBroker) -> BrokerClient:
+        server = make_host_server(broker, "127.0.0.1", 0)
+        self.addCleanup(server.server_close)
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        self.addCleanup(server.shutdown)
+        host, port = server.server_address[0], server.server_address[1]
+        return BrokerClient(f"http://{host}:{port}")
+
+    def test_sign_post_verify_act_over_http(self) -> None:
+        secret = secrets.token_bytes(16)
+        broker = StubBroker(secret)
+        client = self._serve(broker)
+        req = LaunchRequest(
+            op="launch", bottle_id="b1", source_ip="10.0.0.1", image_ref="img", slot=1)
+        got = client.submit(sign_request(req, secret))
+        self.assertEqual(req, got)  # the controller echoes the verified request
+        self.assertEqual(["b1"], [r.bottle_id for r in broker.launched])
+
+    def test_forged_token_raises_broker_auth_error_over_http(self) -> None:
+        secret = secrets.token_bytes(16)
+        broker = StubBroker(secret)
+        client = self._serve(broker)
+        forged = sign_request(
+            LaunchRequest(op="launch", bottle_id="b1"), secrets.token_bytes(16))
+        with self.assertRaises(BrokerAuthError):
+            client.submit(forged)
+        self.assertEqual([], broker.launched)  # fail-closed across the wire
+
+
+if __name__ == "__main__":
+    unittest.main()