feat(supervise): host TUI drives approvals over HTTP, not the DB (Step 2b/2c)

The `bot-bottle supervise` operator TUI read and wrote the queue DB
directly and tried a backend-specific live "apply" (which was unwired —
it raised). It now talks only to the orchestrator control plane:

- OrchestratorClient gains supervise_pending() + supervise_respond().
- discover_orchestrator_url() finds the one running per-host control
  plane by health-probing the backends' well-known :8099 addresses
  (docker publishes on loopback; the firecracker infra VM serves it on
  the orchestrator TAP) — no backend branching in the TUI.
- discover_pending/approve/reject call the client; the server does the
  apply + response + audit atomically. The dead direct-DB apply/audit
  helpers and the docker/macos applicator imports are gone.
- A missing control plane is now a clean one-line error up front, not a
  mid-curses crash.

CLI tests move to mocking the client (the DB-write behaviour they used to
assert is now server-side, covered by test_orchestrator_service). Docker's
orchestrator-container DB wiring lands next so its /supervise endpoints hit
the same shared DB.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
This commit is contained in:
2026-07-16 18:57:47 -04:00
parent 9085d6f713
commit 27fe03b612
3 changed files with 188 additions and 252 deletions
+46 -80
View File
@@ -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:
+61
View File
@@ -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",
]
+81 -172
View File
@@ -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(