feat(orchestrator): slice 1 — registry + attribution + HTTP control plane (#352)

First implementation slice of PRD 0070, the backend-neutral consolidation
core as a plain-process dev-harness (no VM packaging yet):

  * orchestrator/registry.py — SQLite (WAL) runtime-state store on the
    existing DbStore/TableMigrations base. Live bottle registry keyed by
    source IP + per-bottle identity token, with fail-closed attribution:
    a request resolves to a bottle only when its source IP AND identity
    token both match exactly one active record (unknown/ambiguous IP,
    empty token, or token mismatch all deny). Tokens are 256-bit urandom.
  * orchestrator/control_plane.py — the HTTP control plane (the universal
    transport chosen in 0070): register / deregister / list / attribute /
    health. Routing is a pure dispatch() so it is socket-free testable;
    Handler/ControlPlaneServer/make_server are a thin stdlib adapter.
    register/deregister are the live-reload path; listing redacts tokens.
  * orchestrator/__main__.py — `python -m bot_bottle.orchestrator` harness.

Launch/teardown, the launch broker, and the egress/git/supervise data
plane come in later slices. 24 unit tests (attribution matrix, persistence,
dispatch, one real-socket round-trip).

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

tests/unit/test_orchestrator_control_plane.py

@@ -0,0 +1,150 @@
+"""Unit tests for the orchestrator HTTP control plane (PRD 0070).
+
+Mostly exercises the pure `dispatch()` (socket-free, like the supervise
+server tests), plus one real-socket round-trip to prove the handler wiring.
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import threading
+import unittest
+import urllib.request
+from pathlib import Path
+
+from bot_bottle.orchestrator.control_plane import dispatch, make_server
+from bot_bottle.orchestrator.registry import RegistryStore
+
+
+def _body(obj: object) -> bytes:
+    return json.dumps(obj).encode()
+
+
+class TestDispatch(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.store = RegistryStore(Path(self._tmp.name) / "r.db")
+        self.store.migrate()
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_health(self) -> None:
+        status, payload = dispatch(self.store, "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.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({}))
+        self.assertEqual(400, status)
+
+    def test_register_rejects_bad_json(self) -> None:
+        status, _ = dispatch(self.store, "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"")
+        self.assertEqual(200, status)
+        bottles = payload["bottles"]
+        assert isinstance(bottles, list)
+        self.assertEqual(1, len(bottles))
+        first = bottles[0]
+        assert isinstance(first, dict)
+        self.assertNotIn("identity_token", first)
+        self.assertEqual("10.243.0.1", first["source_ip"])
+
+    def test_attribute_ok_and_forbidden(self) -> None:
+        _, reg = dispatch(
+            self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.5"})
+        )
+        token = reg["identity_token"]
+        ok_status, ok = dispatch(
+            self.store, "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",
+            _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.assertEqual(400, status)
+
+    def test_delete(self) -> None:
+        _, reg = dispatch(
+            self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})
+        )
+        bid = reg["bottle_id"]
+        status, payload = dispatch(self.store, "DELETE", f"/bottles/{bid}", b"")
+        self.assertEqual(200, status)
+        self.assertEqual(True, payload["deregistered"])
+        self.assertEqual([], self.store.all())
+
+    def test_delete_missing_404(self) -> None:
+        status, _ = dispatch(self.store, "DELETE", "/bottles/ghost", b"")
+        self.assertEqual(404, status)
+
+    def test_unknown_route_404(self) -> None:
+        status, _ = dispatch(self.store, "GET", "/nope", b"")
+        self.assertEqual(404, status)
+
+    def test_trailing_slash_normalized(self) -> None:
+        status, _ = dispatch(self.store, "GET", "/health/", b"")
+        self.assertEqual(200, status)
+
+
+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)
+        self.addCleanup(server.server_close)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(server.shutdown)
+
+        host, port = server.server_address[0], server.server_address[1]
+        base = f"http://{host}:{port}"
+
+        reg = json.load(urllib.request.urlopen(
+            urllib.request.Request(
+                f"{base}/bottles", data=_body({"source_ip": "10.243.0.7"}),
+                method="POST", headers={"Content-Type": "application/json"},
+            ), timeout=5,
+        ))
+        self.assertTrue(reg["bottle_id"])
+
+        health = json.load(urllib.request.urlopen(f"{base}/health", timeout=5))
+        self.assertEqual("ok", health["status"])
+
+        attr = json.load(urllib.request.urlopen(
+            urllib.request.Request(
+                f"{base}/attribute",
+                data=_body({"source_ip": "10.243.0.7", "identity_token": reg["identity_token"]}),
+                method="POST", headers={"Content-Type": "application/json"},
+            ), timeout=5,
+        ))
+        self.assertEqual(reg["bottle_id"], attr["bottle_id"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_registry.py

@@ -0,0 +1,102 @@
+"""Unit tests for the orchestrator bottle registry + attribution (PRD 0070)."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from bot_bottle.orchestrator.registry import (
+    BottleRecord,
+    RegistryStore,
+    new_identity_token,
+)
+
+
+class TestRegistryStore(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = Path(self._tmp.name) / "registry.db"
+        self.store = RegistryStore(self.db)
+        self.store.migrate()
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_register_mints_id_and_token(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        self.assertTrue(rec.bottle_id)
+        self.assertTrue(rec.identity_token)
+        self.assertEqual("active", rec.state)
+        self.assertEqual(rec, self.store.get(rec.bottle_id))
+
+    def test_identity_tokens_are_unique(self) -> None:
+        tokens = {new_identity_token() for _ in range(200)}
+        self.assertEqual(200, len(tokens))
+        a = self.store.register("10.243.0.1")
+        b = self.store.register("10.243.0.3")
+        self.assertNotEqual(a.identity_token, b.identity_token)
+
+    def test_all_and_deregister(self) -> None:
+        a = self.store.register("10.243.0.1")
+        b = self.store.register("10.243.0.3")
+        self.assertEqual({a.bottle_id, b.bottle_id}, {r.bottle_id for r in self.store.all()})
+        self.assertTrue(self.store.deregister(a.bottle_id))
+        self.assertEqual([b.bottle_id], [r.bottle_id for r in self.store.all()])
+
+    def test_deregister_missing_is_false(self) -> None:
+        self.assertFalse(self.store.deregister("nope"))
+
+    def test_explicit_id_replaces(self) -> None:
+        self.store.register("10.243.0.1", bottle_id="fixed", metadata="a")
+        self.store.register("10.243.0.9", bottle_id="fixed", metadata="b")
+        rec = self.store.get("fixed")
+        assert rec is not None
+        self.assertEqual("10.243.0.9", rec.source_ip)
+        self.assertEqual("b", rec.metadata)
+        self.assertEqual(1, len(self.store.all()))
+
+    def test_attribute_success_requires_ip_and_token(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        got = self.store.attribute("10.243.0.1", rec.identity_token)
+        assert got is not None
+        self.assertEqual(rec.bottle_id, got.bottle_id)
+
+    def test_attribute_wrong_token_denied(self) -> None:
+        self.store.register("10.243.0.1")
+        self.assertIsNone(self.store.attribute("10.243.0.1", "wrong-token"))
+
+    def test_attribute_unknown_ip_denied(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        self.assertIsNone(self.store.attribute("10.243.9.9", rec.identity_token))
+
+    def test_attribute_empty_token_denied(self) -> None:
+        self.store.register("10.243.0.1")
+        self.assertIsNone(self.store.attribute("10.243.0.1", ""))
+
+    def test_attribute_ambiguous_ip_denied(self) -> None:
+        # Two active bottles on one source IP is a misconfiguration — deny
+        # rather than guess (fail-closed), even with a valid token.
+        a = self.store.register("10.243.0.1", bottle_id="a")
+        self.store.register("10.243.0.1", bottle_id="b")
+        self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token))
+
+    def test_state_persists_across_reopen(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        reopened = RegistryStore(self.db)
+        got = reopened.get(rec.bottle_id)
+        assert got is not None
+        self.assertEqual(rec, got)
+        # Attribution works against the reopened (durable) store too.
+        self.assertIsNotNone(reopened.attribute("10.243.0.1", rec.identity_token))
+
+    def test_redacted_hides_token(self) -> None:
+        rec = BottleRecord(
+            bottle_id="x", source_ip="10.243.0.1", identity_token="secret"
+        )
+        self.assertNotIn("identity_token", rec.redacted())
+        self.assertEqual("10.243.0.1", rec.redacted()["source_ip"])
+
+
+if __name__ == "__main__":
+    unittest.main()