diff --git a/bot_bottle/cli/supervise.py b/bot_bottle/cli/supervise.py index ff66ba6..add3f6c 100644 --- a/bot_bottle/cli/supervise.py +++ b/bot_bottle/cli/supervise.py @@ -20,32 +20,19 @@ from datetime import datetime, timezone from pathlib import Path from ..paths import bot_bottle_root -from ..bottle_state import read_metadata -from ..backend.docker.egress_apply import ( - EgressApplyError, - applicator as _docker_applicator, -) -from ..backend.macos_container.egress_apply import ( - applicator as _macos_applicator, -) from ..log import Die, error, info +from ..orchestrator.client import ( + OrchestratorClient, + OrchestratorClientError, + discover_orchestrator_url, +) from ..supervise import ( - COMPONENT_FOR_TOOL, - AuditEntry, Proposal, - Response, - STATUS_APPROVED, - STATUS_MODIFIED, - STATUS_REJECTED, TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK, TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW, - list_all_pending_proposals, - render_diff, - write_audit_entry, - write_response, ) from ._common import PROG @@ -65,25 +52,31 @@ class QueuedProposal: proposal: Proposal -# Errors any remediation engine may raise. Caught by the TUI key -# handlers and surfaced in the status line so a failed apply keeps -# the proposal pending rather than crashing curses. -ApplyError = (EgressApplyError,) +# A failed operator action (orchestrator unreachable, bottle torn down, +# 409) is caught by the TUI key handlers and surfaced in the status line so +# the proposal stays pending rather than crashing curses. +ApplyError = (OrchestratorClientError,) -def apply_routes_change(slug: str, content: str) -> tuple[str, str]: - meta = read_metadata(slug) - backend = meta.backend if meta is not None else "" - if backend == "macos-container": - return _macos_applicator.apply_routes_change(slug, content) - return _docker_applicator.apply_routes_change(slug, content) +# The one per-host orchestrator, discovered lazily on first use. Every +# operator action — list, approve, reject — goes through its HTTP control +# plane (the orchestrator owns the single DB + live policy); there is no +# direct-DB path and no backend branching here. +_client_instance: OrchestratorClient | None = None + + +def _client() -> OrchestratorClient: + global _client_instance # noqa: PLW0603 — CLI-session singleton + if _client_instance is None: + _client_instance = OrchestratorClient(discover_orchestrator_url()) + return _client_instance def discover_pending() -> list[QueuedProposal]: - """Collect pending proposals across bottles.""" + """Collect pending proposals across bottles from the orchestrator.""" out = [ - QueuedProposal(proposal=proposal) - for proposal in list_all_pending_proposals() + QueuedProposal(proposal=Proposal.from_dict(d)) + for d in _client().supervise_pending() ] out.sort(key=lambda q: q.proposal.arrival_timestamp) return out @@ -136,39 +129,27 @@ def approve( notes: str = "", final_file: str | None = None, ) -> None: - """Apply the proposal, write the waiting response, and audit it.""" - status = STATUS_MODIFIED if final_file is not None else STATUS_APPROVED - file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file - - diff_before, diff_after = "", "" - if qp.proposal.tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK): - diff_before, diff_after = apply_routes_change( - qp.proposal.bottle_slug, - file_to_apply, - ) - - response = Response( - proposal_id=qp.proposal.id, - status=status, + """Approve (or, with `final_file`, modify-then-approve) via the + orchestrator: it applies the route change to the bottle's live policy, + writes the response that unblocks the agent, and audits it — one atomic + server-side op. Raises `OrchestratorClientError` on failure.""" + _client().supervise_respond( + qp.proposal.id, + bottle_slug=qp.proposal.bottle_slug, + decision="modify" if final_file is not None else "approve", notes=notes, final_file=final_file, ) - write_response(qp.proposal.bottle_slug, response) - _write_audit( - qp, action=status, notes=notes, - diff_before=diff_before, diff_after=diff_after, - ) + def reject(qp: QueuedProposal, *, reason: str) -> None: - """Write a rejection response and an audit entry.""" - response = Response( - proposal_id=qp.proposal.id, - status=STATUS_REJECTED, + """Reject via the orchestrator (writes the response + audit).""" + _client().supervise_respond( + qp.proposal.id, + bottle_slug=qp.proposal.bottle_slug, + decision="reject", notes=reason, - final_file=None, ) - write_response(qp.proposal.bottle_slug, response) - _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="") def _approve_from_tui( @@ -188,29 +169,6 @@ def _approve_from_tui( return _approval_status(qp, verb) -def _write_audit( - qp: QueuedProposal, - *, - action: str, - notes: str, - diff_before: str, - diff_after: str, -) -> None: - """Audit log for egress tool.""" - component = COMPONENT_FOR_TOOL.get(qp.proposal.tool) - if component is None: - return - write_audit_entry(AuditEntry( - timestamp=datetime.now(timezone.utc).isoformat(), - bottle_slug=qp.proposal.bottle_slug, - component=component, - operator_action=action, - operator_notes=notes, - justification=qp.proposal.justification, - diff=render_diff(diff_before, diff_after, label=component), - )) - - # --- $EDITOR integration -------------------------------------------------- @@ -245,6 +203,14 @@ def cmd_supervise(argv: list[str]) -> int: ) args = parser.parse_args(argv) + # Establish the orchestrator connection up front so a missing control + # plane is a clean one-line error, not a curses crash mid-loop. + try: + _client() + except OrchestratorClientError as e: + error(str(e)) + return 1 + if args.once: return _list_once() try: diff --git a/bot_bottle/orchestrator/client.py b/bot_bottle/orchestrator/client.py index f6bdc71..b34f038 100644 --- a/bot_bottle/orchestrator/client.py +++ b/bot_bottle/orchestrator/client.py @@ -135,10 +135,71 @@ class OrchestratorClient: bottles = payload.get("bottles") return bottles if isinstance(bottles, list) else [] + # --- supervise queue (operator TUI) ------------------------------------ + + def supervise_pending(self) -> list[dict[str, object]]: + """Pending supervise proposals across all bottles + (`GET /supervise/proposals`).""" + payload = self._ok("GET", "/supervise/proposals") + proposals = payload.get("proposals") + return proposals if isinstance(proposals, list) else [] + + def supervise_respond( + self, + proposal_id: str, + *, + bottle_slug: str, + decision: str, + notes: str = "", + final_file: str | None = None, + ) -> None: + """Record an operator decision (`POST /supervise/respond`). `decision` + is approve/modify/reject. Raises `OrchestratorClientError` if the + proposal is gone or the bottle can no longer be applied to (409).""" + body: dict[str, object] = { + "proposal_id": proposal_id, + "bottle_slug": bottle_slug, + "decision": decision, + "notes": notes, + } + if final_file is not None: + body["final_file"] = final_file + self._ok("POST", "/supervise/respond", body) + + +def discover_orchestrator_url(*, timeout: float = 2.0) -> str: + """The URL of the one running per-host orchestrator control plane, probing + the backends' well-known control-plane addresses (both on port 8099): + docker publishes it on loopback; the firecracker infra VM serves it on the + orchestrator TAP. Returns the first that answers `/health`; raises if none + do (no orchestrator up — launch a bottle first).""" + candidates: list[str] = [] + try: # docker: loopback-published control plane + from .lifecycle import DEFAULT_PORT as _DOCKER_PORT + candidates.append(f"http://127.0.0.1:{_DOCKER_PORT}") + except Exception: # noqa: BLE001 — backend optional + candidates.append("http://127.0.0.1:8099") + try: # firecracker: infra VM control plane on the orchestrator TAP + from ..backend.firecracker import netpool + from ..backend.firecracker.infra_vm import CONTROL_PLANE_PORT + candidates.append( + f"http://{netpool.orch_slot().guest_ip}:{CONTROL_PLANE_PORT}") + except Exception: # noqa: BLE001 — backend optional / not firecracker + pass + for url in candidates: + if OrchestratorClient(url, timeout=timeout).health(): + return url + raise OrchestratorClientError( + "no running orchestrator control plane found (tried " + + ", ".join(candidates) + + "); launch a bottle first" + ) + __all__ = [ "OrchestratorClient", "OrchestratorClientError", "RegisteredBottle", "DEFAULT_TIMEOUT_SECONDS", + "discover_orchestrator_url", ] diff --git a/tests/unit/test_supervise_cli.py b/tests/unit/test_supervise_cli.py index 9c680c9..c1d1d2a 100644 --- a/tests/unit/test_supervise_cli.py +++ b/tests/unit/test_supervise_cli.py @@ -1,31 +1,24 @@ -"""Unit: supervise headless paths (PRD 0013 phase 4, PRD 0016). +"""Unit: supervise headless paths — the discovery + approve/reject that the +TUI key handlers call into. -The curses TUI itself isn't exercised here — these tests cover the -discovery + approve/reject paths that the TUI's key handlers call into. +These go through the orchestrator HTTP client now (the operator never +touches the DB directly), so the client is mocked here; the server-side +apply / response / audit is covered in test_orchestrator_service. """ import os import tempfile import unittest from datetime import datetime, timezone -from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch -from bot_bottle import supervise -from tests.unit import use_bottle_root -from bot_bottle.audit_store import AuditStore from bot_bottle.cli import supervise as supervise_cli -from bot_bottle.queue_store import QueueStore from bot_bottle.supervise import ( Proposal, - STATUS_APPROVED, - STATUS_MODIFIED, - STATUS_REJECTED, TOOL_EGRESS_ALLOW, + TOOL_EGRESS_BLOCK, TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW, - read_audit_entries, - read_response, sha256_hex, ) @@ -33,198 +26,114 @@ from bot_bottle.supervise import ( FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc) -def _proposal(slug: str = "dev", tool: str = TOOL_EGRESS_ALLOW) -> Proposal: +def _proposal(slug: str = "dev", tool: str = TOOL_EGRESS_ALLOW, + *, now: datetime = FIXED) -> Proposal: payloads = { - supervise.TOOL_EGRESS_ALLOW: "routes:\n - host: example.com\n", - supervise.TOOL_EGRESS_BLOCK: "routes:\n - host: example.com\n", + TOOL_EGRESS_ALLOW: "routes:\n - host: example.com\n", + TOOL_EGRESS_BLOCK: "routes:\n - host: example.com\n", TOOL_GITLEAKS_ALLOW: "file: tests/test_fixture.py\nline: 3\n", TOOL_EGRESS_TOKEN_ALLOW: "host: api.example.com\ndetector: token\n", } payload = payloads.get(tool, "") return Proposal.new( - bottle_slug=slug, tool=tool, - proposed_file=payload, - justification=f"needed for {slug}", - current_file_hash=sha256_hex(payload), - now=FIXED, + bottle_slug=slug, tool=tool, proposed_file=payload, + justification=f"needed for {slug}", current_file_hash=sha256_hex(payload), + now=now, ) -class _FakeHomeMixin: - """Point bot_bottle_root at a temp dir (via BOT_BOTTLE_ROOT) for the test.""" +class _ClientMixin: + """Install a mock orchestrator client as the CLI-session singleton.""" - def _setup_fake_home(self): - self._tmp = tempfile.TemporaryDirectory(prefix="supervise-test.") - self._restore_home = use_bottle_root(Path(self._tmp.name) / ".bot-bottle") - QueueStore("").migrate() - AuditStore().migrate() - - def _teardown_fake_home(self): - self._restore_home() - self._tmp.cleanup() + def _install_client(self, pending: "list[Proposal] | None" = None) -> MagicMock: + client = MagicMock() + client.supervise_pending.return_value = [ + p.to_dict() for p in (pending or []) + ] + patcher = patch.object(supervise_cli, "_client", return_value=client) + patcher.start() + self.addCleanup(patcher.stop) # type: ignore[attr-defined] + self.addCleanup( # type: ignore[attr-defined] + lambda: setattr(supervise_cli, "_client_instance", None)) + return client -class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase): - def setUp(self): - self._setup_fake_home() - - def tearDown(self): - self._teardown_fake_home() - - def test_empty_when_no_queues(self): +class TestDiscoverPending(_ClientMixin, unittest.TestCase): + def test_empty(self) -> None: + self._install_client([]) self.assertEqual([], supervise_cli.discover_pending()) - def test_walks_all_slug_subdirs(self): - for slug in ("dev", "api"): - supervise.write_proposal(_proposal(slug=slug)) + def test_lists_all_bottles(self) -> None: + self._install_client([_proposal("dev"), _proposal("api")]) pending = supervise_cli.discover_pending() - self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending}) + self.assertEqual( + {"dev", "api"}, {qp.proposal.bottle_slug for qp in pending}) - def test_sorted_by_arrival_across_bottles(self): - early = Proposal.new( - bottle_slug="api", tool=TOOL_EGRESS_ALLOW, - proposed_file="routes:\n - host: early.example.com\n", justification="early", - current_file_hash="h", - now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc), - ) - late = Proposal.new( - bottle_slug="dev", tool=TOOL_EGRESS_ALLOW, - proposed_file="routes:\n - host: late.example.com\n", justification="late", - current_file_hash="h", - now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), - ) - for p in (late, early): - supervise.write_proposal(p) + def test_sorted_by_arrival(self) -> None: + early = _proposal( + "api", now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc)) + late = _proposal( + "dev", now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc)) + self._install_client([late, early]) pending = supervise_cli.discover_pending() self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending]) - def test_excludes_already_responded(self): - p = _proposal() - supervise.write_proposal(p) - supervise.write_response("dev", supervise.Response( - proposal_id=p.id, status=STATUS_APPROVED, notes="", - )) - self.assertEqual([], supervise_cli.discover_pending()) +class TestApproveReject(_ClientMixin, unittest.TestCase): + def _qp(self, tool: str = TOOL_EGRESS_ALLOW) -> "supervise_cli.QueuedProposal": + return supervise_cli.QueuedProposal(proposal=_proposal(tool=tool)) -class TestApproveReject(_FakeHomeMixin, unittest.TestCase): - def setUp(self): - self._setup_fake_home() - - def tearDown(self): - self._teardown_fake_home() - - def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW): - p = _proposal(tool=tool) - supervise.write_proposal(p) - return supervise_cli.QueuedProposal(proposal=p) - - def test_approve_writes_response(self): - qp = self._enqueue() - with patch( - "bot_bottle.cli.supervise.apply_routes_change", - return_value=("routes: []\n", "routes:\n - host: example.com\n"), - ): - supervise_cli.approve(qp) - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual(STATUS_APPROVED, resp.status) - self.assertIsNone(resp.final_file) - - def test_approve_with_final_file_marks_modified(self): - qp = self._enqueue() - with patch( - "bot_bottle.cli.supervise.apply_routes_change", - return_value=("routes: []\n", "routes:\n - host: edited.example.com\n"), - ): - supervise_cli.approve( - qp, - final_file="routes:\n - host: edited.example.com\n", - notes="tweaked", - ) - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual(STATUS_MODIFIED, resp.status) - self.assertEqual("routes:\n - host: edited.example.com\n", resp.final_file) - self.assertEqual("tweaked", resp.notes) - - def test_reject_writes_rejection(self): - qp = self._enqueue() - supervise_cli.reject(qp, reason="nope") - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual(STATUS_REJECTED, resp.status) - self.assertEqual("nope", resp.notes) - - def test_approve_egress_block_writes_audit_log(self): - qp = self._enqueue(tool=supervise.TOOL_EGRESS_BLOCK) - with patch( - "bot_bottle.cli.supervise.apply_routes_change", - return_value=("routes: []\n", "routes:\n - host: example.com\n"), - ) as apply_routes_change: - supervise_cli.approve(qp) - apply_routes_change.assert_called_once_with( - "dev", - "routes:\n - host: example.com\n", + def test_approve_calls_respond(self) -> None: + client = self._install_client() + qp = self._qp() + supervise_cli.approve(qp) + client.supervise_respond.assert_called_once_with( + qp.proposal.id, bottle_slug="dev", decision="approve", + notes="", final_file=None, ) - entries = read_audit_entries("egress", "dev") - self.assertEqual(1, len(entries)) - self.assertEqual(STATUS_APPROVED, entries[0].operator_action) - self.assertEqual("needed for dev", entries[0].justification) - def test_approve_gitleaks_allow_leaves_response_for_gate(self): - qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) - supervise_cli.approve(qp, notes="dummy fixture") - # Gate polls the DB for the response; TUI must not archive it. - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual(STATUS_APPROVED, resp.status) - self.assertEqual("dummy fixture", resp.notes) + def test_modify_sets_decision_and_final_file(self) -> None: + client = self._install_client() + qp = self._qp() + edited = "routes:\n - host: edited.example.com\n" + supervise_cli.approve(qp, final_file=edited, notes="tweaked") + client.supervise_respond.assert_called_once_with( + qp.proposal.id, bottle_slug="dev", decision="modify", + notes="tweaked", final_file=edited, + ) - def test_tui_gitleaks_allow_requires_reason(self): - qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) + def test_reject_calls_respond(self) -> None: + client = self._install_client() + qp = self._qp() + supervise_cli.reject(qp, reason="nope") + client.supervise_respond.assert_called_once_with( + qp.proposal.id, bottle_slug="dev", decision="reject", notes="nope", + ) + + def test_tui_report_only_requires_reason(self) -> None: + self._install_client() + qp = self._qp(tool=TOOL_GITLEAKS_ALLOW) with patch.object(supervise_cli, "_prompt", return_value=""): status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] self.assertEqual("approve aborted (empty reason)", status) - def test_tui_gitleaks_allow_writes_reason(self): - qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) + def test_tui_report_only_writes_reason(self) -> None: + client = self._install_client() + qp = self._qp(tool=TOOL_GITLEAKS_ALLOW) with patch.object(supervise_cli, "_prompt", return_value="test fixture"): status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] self.assertIn("approved gitleaks-allow", status) - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual("test fixture", resp.notes) + client.supervise_respond.assert_called_once() + self.assertEqual( + "test fixture", client.supervise_respond.call_args.kwargs["notes"]) - def test_approve_token_allow_leaves_response_for_egress(self): - qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) - supervise_cli.approve(qp, notes="false positive") - # The egress addon polls the DB for the response; the TUI must - # not archive it (the addon archives after reading). - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual(STATUS_APPROVED, resp.status) - self.assertEqual("false positive", resp.notes) - - def test_token_allow_writes_no_audit_log(self): - qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) - supervise_cli.approve(qp, notes="false positive") - self.assertEqual([], read_audit_entries("egress", "dev")) - - def test_tui_token_allow_requires_reason(self): - qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) - with patch.object(supervise_cli, "_prompt", return_value=""): - status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] - self.assertEqual("approve aborted (empty reason)", status) - - def test_tui_token_allow_writes_reason(self): - qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) - with patch.object(supervise_cli, "_prompt", return_value="legit"): - status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] - self.assertIn("approved egress-token-allow", status) - resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) - self.assertEqual("legit", resp.notes) - - def test_suffix_for_token_allow_is_txt(self): - self.assertEqual(".txt", supervise_cli._suffix_for_tool(TOOL_EGRESS_TOKEN_ALLOW)) + def test_suffix_for_token_allow_is_txt(self) -> None: + self.assertEqual( + ".txt", supervise_cli._suffix_for_tool(TOOL_EGRESS_TOKEN_ALLOW)) class TestEditInEditor(unittest.TestCase): - def test_runs_editor_returns_edited_content(self): + def test_runs_editor_returns_edited_content(self) -> None: original_editor = os.environ.get("EDITOR") try: with tempfile.NamedTemporaryFile( @@ -245,7 +154,7 @@ class TestEditInEditor(unittest.TestCase): else: os.environ["EDITOR"] = original_editor - def test_returns_none_when_unchanged(self): + def test_returns_none_when_unchanged(self) -> None: original_editor = os.environ.get("EDITOR") try: with tempfile.NamedTemporaryFile(