From 9085d6f7132dcb11d6141c0cf8b87dbdf0d0d364 Mon Sep 17 00:00:00 2001 From: didericis Date: Thu, 16 Jul 2026 18:51:55 -0400 Subject: [PATCH] feat(supervise): orchestrator-side operator-approval API (Step 2a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator owns the single DB *and* the live policy, so operator decisions belong there — applied server-side, reached over HTTP. Adds: GET /supervise/proposals -> pending proposals across bottles POST /supervise/respond -> apply + record an operator decision `supervise_respond` is one atomic server-side op on the one DB: approve/ modify on an egress tool rewrites the bottle's policy (so the gateway serves the new routes on its next /resolve — the live "apply" that was a documented TODO), then writes the queued Response (unblocking the agent's MCP call) and an audit entry. reject records the response + audit only. Fails closed (409) when the proposal is unknown or the bottle was torn down before the operator acted (an egress apply would have no target). This is the server half of unifying every backend onto one HTTP path for supervise; the host TUI (direct-DB today) moves onto this client next. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck --- bot_bottle/orchestrator/control_plane.py | 37 ++++++ bot_bottle/orchestrator/service.py | 105 ++++++++++++++++ tests/unit/test_orchestrator_control_plane.py | 73 +++++++++++ tests/unit/test_orchestrator_service.py | 117 ++++++++++++++++++ 4 files changed, 332 insertions(+) diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index c590915..064528f 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -16,6 +16,10 @@ vsock / unix-socket portability caveats): POST /attribute -> 200 {"bottle_id"} | 403 POST /resolve -> 200 {"bottle_id","policy"} | 403 body: {"source_ip","identity_token"} + GET /supervise/proposals -> 200 {"proposals": [ , ...]} + POST /supervise/respond -> 200 {"responded": true} | 409 (operator) + body: {"proposal_id","bottle_slug", + "decision", ["notes"],["final_file"]} `POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or tear down) the bottle in the registry AND broker the backend-native launch @@ -127,6 +131,39 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches return 403, {"error": "unattributed"} return 200, {"bottle_id": rec.bottle_id} + if method == "GET" and route == "/supervise/proposals": + # Operator TUI: pending supervise proposals across all bottles. + return 200, {"proposals": orch.supervise_pending()} + + if method == "POST" and route == "/supervise/respond": + # Operator decision: apply (approve/modify rewrites egress policy), + # write the queued response, audit — all server-side on the one DB. + try: + data = _parse_json_object(body) + except ValueError as e: + return 400, {"error": f"invalid JSON: {e}"} + proposal_id = data.get("proposal_id") + bottle_slug = data.get("bottle_slug") + decision = data.get("decision") + if not (isinstance(proposal_id, str) and proposal_id): + return 400, {"error": "proposal_id (string) is required"} + if not (isinstance(bottle_slug, str) and bottle_slug): + return 400, {"error": "bottle_slug (string) is required"} + if not (isinstance(decision, str) and decision): + return 400, {"error": "decision (string) is required"} + notes = data.get("notes") + final_file = data.get("final_file") + ok, err = orch.supervise_respond( + proposal_id, + bottle_slug=bottle_slug, + decision=decision, + notes=notes if isinstance(notes, str) else "", + final_file=final_file if isinstance(final_file, str) else None, + ) + if ok: + return 200, {"responded": True} + return 409, {"error": err} + if method == "POST" and route == "/resolve": # The per-request lookup the multi-tenant gateway makes: returns the # bottle's policy. Requires a matching (source_ip, identity_token) diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index c44e5be..5c34815 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -17,9 +17,37 @@ Launch lifecycle: from __future__ import annotations +import json +from datetime import datetime, timezone + from .broker import LaunchBroker, LaunchRequest, sign_request from .registry import BottleRecord, RegistryStore from .gateway import Gateway +from ..supervise import ( + AuditEntry, + COMPONENT_FOR_TOOL, + Response, + STATUS_APPROVED, + STATUS_MODIFIED, + STATUS_REJECTED, + TOOL_EGRESS_ALLOW, + TOOL_EGRESS_BLOCK, + list_all_pending_proposals, + read_proposal, + render_diff, + write_audit_entry, + write_response, +) + + +# Operator decision → Response.status. The apply half (egress tools) runs +# for approve/modify only. +_RESPOND_STATUS = { + "approve": STATUS_APPROVED, + "modify": STATUS_MODIFIED, + "reject": STATUS_REJECTED, +} +_APPLY_TOOLS = (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK) class Orchestrator: @@ -117,6 +145,83 @@ class Orchestrator: the bottle is unknown.""" return self.registry.set_policy(bottle_id, policy) + # --- supervise queue (operator approvals) ------------------------------ + # + # The orchestrator owns the single DB *and* the live policy, so operator + # decisions are applied here, server-side, and reached over HTTP by the + # host TUI (no direct-DB access, one path for every backend). + + def supervise_pending(self) -> list[dict[str, object]]: + """All pending proposals across bottles, FIFO, as JSON dicts + (`Proposal.to_dict`, round-trippable via `Proposal.from_dict`).""" + return [p.to_dict() for p in list_all_pending_proposals()] + + def _record_for_slug(self, slug: str) -> BottleRecord | None: + """The live registry record whose metadata carries `slug`, or None + (e.g. the bottle was torn down before the operator responded).""" + for rec in self.registry.all(): + try: + meta = json.loads(rec.metadata) if rec.metadata else {} + except ValueError: + meta = {} + if isinstance(meta, dict) and meta.get("slug") == slug: + return rec + return None + + def supervise_respond( + self, + proposal_id: str, + *, + bottle_slug: str, + decision: str, + notes: str = "", + final_file: str | None = None, + ) -> tuple[bool, str]: + """Record an operator decision on a queued proposal, applying it + server-side. `decision` is approve/modify/reject. + + Approve/modify on an egress tool rewrites the bottle's policy so the + gateway serves the new routes on its next `/resolve` (the live apply); + then the queued Response is written (unblocking the agent's MCP call) + and an audit entry recorded — all against the one DB. Returns + (ok, error): ok=False with a message when the proposal or decision is + unknown, or the bottle is gone so an approval can't be applied.""" + status = _RESPOND_STATUS.get(decision) + if status is None: + return False, f"unknown decision {decision!r}" + try: + proposal = read_proposal(bottle_slug, proposal_id) + except FileNotFoundError: + return False, "no such proposal" + + diff_before, diff_after = "", "" + if status in (STATUS_APPROVED, STATUS_MODIFIED) and proposal.tool in _APPLY_TOOLS: + new_policy = final_file if final_file is not None else proposal.proposed_file + rec = self._record_for_slug(bottle_slug) + if rec is None: + return False, ( + f"bottle {bottle_slug!r} is no longer registered; " + "cannot apply the route change" + ) + diff_before, diff_after = rec.policy, new_policy + self.set_policy(rec.bottle_id, new_policy) + + write_response(bottle_slug, Response( + proposal_id=proposal_id, status=status, notes=notes, final_file=final_file, + )) + component = COMPONENT_FOR_TOOL.get(proposal.tool) + if component is not None: + write_audit_entry(AuditEntry( + timestamp=datetime.now(timezone.utc).isoformat(), + bottle_slug=bottle_slug, + component=component, + operator_action=status, + operator_notes=notes, + justification=proposal.justification, + diff=render_diff(diff_before, diff_after, label=component), + )) + return True, "" + # --- consolidated gateway ---------------------------------------------- def ensure_gateway(self) -> None: diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 6816990..9fd4fb0 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -13,11 +13,19 @@ import threading import unittest import urllib.request from pathlib import Path +from unittest.mock import patch 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 +from bot_bottle.store_manager import StoreManager +from bot_bottle.supervise import ( + Proposal, + TOOL_EGRESS_ALLOW, + sha256_hex, + write_proposal, +) def _body(obj: object) -> bytes: @@ -235,5 +243,70 @@ class TestServerRoundTrip(unittest.TestCase): self.assertEqual(reg["bottle_id"], attr["bottle_id"]) +class TestDispatchSupervise(unittest.TestCase): + """The /supervise/* routes over the pure dispatch().""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + db = root / "db" / "bot-bottle.db" + db.parent.mkdir(parents=True) + self._env = patch.dict("os.environ", { + "BOT_BOTTLE_ROOT": str(root), + "SUPERVISE_DB_PATH": str(db), + }) + self._env.start() + self.store = RegistryStore(db) + self.store.migrate() + StoreManager(db).migrate() + secret = secrets.token_bytes(16) + self.orch = Orchestrator(self.store, StubBroker(secret), secret) + + def tearDown(self) -> None: + self._env.stop() + self._tmp.cleanup() + + def _queue(self, slug: str, proposed: str) -> str: + self.store.register( + "10.243.0.1", metadata=json.dumps({"slug": slug}), policy="routes: []\n") + p = Proposal.new( + bottle_slug=slug, tool=TOOL_EGRESS_ALLOW, proposed_file=proposed, + justification="need it", current_file_hash=sha256_hex(proposed)) + write_proposal(p) + return p.id + + def test_list_pending(self) -> None: + pid = self._queue("demo", "routes:\n - host: google.com\n") + status, payload = dispatch(self.orch, "GET", "/supervise/proposals", b"") + self.assertEqual(200, status) + proposals = payload["proposals"] + assert isinstance(proposals, list) + self.assertEqual(pid, proposals[0]["id"]) + + def test_respond_approve_applies_and_clears(self) -> None: + pid = self._queue("demo", "routes:\n - host: google.com\n") + status, payload = dispatch( + self.orch, "POST", "/supervise/respond", + _body({"proposal_id": pid, "bottle_slug": "demo", "decision": "approve"}), + ) + self.assertEqual(200, status) + self.assertTrue(payload["responded"]) + _, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"") + self.assertEqual([], listing["proposals"]) + + def test_respond_requires_fields(self) -> None: + status, _ = dispatch( + self.orch, "POST", "/supervise/respond", _body({"decision": "approve"})) + self.assertEqual(400, status) + + def test_respond_unknown_proposal_conflicts(self) -> None: + status, payload = dispatch( + self.orch, "POST", "/supervise/respond", + _body({"proposal_id": "ghost", "bottle_slug": "demo", "decision": "approve"}), + ) + self.assertEqual(409, status) + self.assertIn("no such proposal", str(payload["error"])) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index c8139c2..8957ba0 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -2,15 +2,26 @@ from __future__ import annotations +import json import secrets import tempfile import unittest from pathlib import Path +from unittest.mock import patch from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker from bot_bottle.orchestrator.registry import RegistryStore from bot_bottle.orchestrator.service import Orchestrator from bot_bottle.orchestrator.gateway import Gateway +from bot_bottle.store_manager import StoreManager +from bot_bottle.supervise import ( + Proposal, + STATUS_APPROVED, + TOOL_EGRESS_ALLOW, + read_response, + sha256_hex, + write_proposal, +) class _FailingBroker(LaunchBroker): @@ -155,5 +166,111 @@ class TestOrchestrator(unittest.TestCase): ) +class TestOrchestratorSupervise(unittest.TestCase): + """Operator-approval flow: the orchestrator applies the decision + server-side against the single DB (queue + policy + audit).""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + db = root / "db" / "bot-bottle.db" + db.parent.mkdir(parents=True) + # One DB for registry + supervise queue + audit (as in the VM). + self._env = patch.dict("os.environ", { + "BOT_BOTTLE_ROOT": str(root), + "SUPERVISE_DB_PATH": str(db), + }) + self._env.start() + self.store = RegistryStore(db) + self.store.migrate() + StoreManager(db).migrate() + secret = secrets.token_bytes(16) + self.orch = Orchestrator(self.store, StubBroker(secret), secret) + + def tearDown(self) -> None: + self._env.stop() + self._tmp.cleanup() + + def _register(self, slug: str, policy: str) -> str: + rec = self.store.register( + "10.243.0.1", metadata=json.dumps({"slug": slug}), policy=policy) + return rec.bottle_id + + def _queue(self, slug: str, proposed: str) -> str: + p = Proposal.new( + bottle_slug=slug, tool=TOOL_EGRESS_ALLOW, proposed_file=proposed, + justification="need it", current_file_hash=sha256_hex(proposed)) + write_proposal(p) + return p.id + + def test_pending_lists_queued_proposal(self) -> None: + self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + pending = self.orch.supervise_pending() + self.assertEqual(1, len(pending)) + self.assertEqual(pid, pending[0]["id"]) + self.assertEqual("demo", pending[0]["bottle_slug"]) + + def test_approve_applies_policy_writes_response_and_clears_pending(self) -> None: + bottle_id = self._register("demo", "routes:\n - host: existing.com\n") + new_routes = "routes:\n - host: google.com\n" + pid = self._queue("demo", new_routes) + ok, err = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="approve") + self.assertTrue(ok, err) + # policy live-applied so /resolve serves the new routes + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual(new_routes, rec.policy) + # response written -> agent unblocks, proposal no longer pending + self.assertEqual(STATUS_APPROVED, read_response("demo", pid).status) + self.assertEqual([], self.orch.supervise_pending()) + + def test_modify_applies_final_file_not_proposed(self) -> None: + bottle_id = self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + edited = "routes:\n - host: example.com\n" + ok, _ = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="modify", final_file=edited) + self.assertTrue(ok) + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual(edited, rec.policy) + + def test_reject_leaves_policy_unchanged(self) -> None: + bottle_id = self._register("demo", "routes:\n - host: existing.com\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + ok, _ = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="reject", notes="no") + self.assertTrue(ok) + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual("routes:\n - host: existing.com\n", rec.policy) + self.assertEqual("rejected", read_response("demo", pid).status) + + def test_unknown_proposal_is_error(self) -> None: + ok, err = self.orch.supervise_respond( + "ghost", bottle_slug="demo", decision="approve") + self.assertFalse(ok) + self.assertIn("no such proposal", err) + + def test_unknown_decision_is_error(self) -> None: + self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + ok, err = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="bogus") + self.assertFalse(ok) + self.assertIn("unknown decision", err) + + def test_approve_when_bottle_gone_cannot_apply(self) -> None: + # Proposal queued but the bottle was torn down before the operator + # acted: an egress apply has no target, so respond fails closed. + pid = self._queue("ghost-bottle", "routes:\n - host: google.com\n") + ok, err = self.orch.supervise_respond( + pid, bottle_slug="ghost-bottle", decision="approve") + self.assertFalse(ok) + self.assertIn("no longer registered", err) + + if __name__ == "__main__": unittest.main()