From 8abccf7ffecab3cbfae3d1026d5acc5d1505a310 Mon Sep 17 00:00:00 2001 From: didericis Date: Fri, 24 Jul 2026 14:52:17 -0400 Subject: [PATCH] refactor(supervise): make Supervisor a service class the orchestrator calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turn the loose queue/audit functions in orchestrator/supervisor/queue.py into methods on a concrete Supervisor service. The Orchestrator now owns an injectable `self._supervisor` and calls `self._supervisor.write_proposal(...)` etc., instead of module-level free functions — the supervise dependency is explicit and mockable, and a Supervisor can be scoped to a `db_path` (defaults to the host DB) so tests can point it at a temp database. - Supervisor (in the package __init__) drops the ABC and gains write_proposal, read_proposal, list_pending_proposals, list_all_pending_proposals, write_response, read_response, archive_all_proposals, write_audit_entry, read_audit_entries, plus the launch-time prepare(). - render_diff / sha256_hex are pure and stateless, so they move to orchestrator/supervisor/util.py (re-exported from the facade) rather than becoming methods. - queue.py is deleted; service.py + the supervise tests call through a Supervisor instance. Full unit suite green (2243). Co-Authored-By: Claude Opus 4.8 --- bot_bottle/orchestrator/service.py | 30 +++-- .../orchestrator/supervisor/__init__.py | 105 ++++++++++++------ bot_bottle/orchestrator/supervisor/queue.py | 93 ---------------- bot_bottle/orchestrator/supervisor/util.py | 31 ++++++ tests/unit/test_orchestrator_control_plane.py | 4 +- tests/unit/test_orchestrator_service.py | 9 +- tests/unit/test_supervise.py | 56 +++++----- tests/unit/test_supervise_edge.py | 30 +++-- tests/unit/test_supervise_server.py | 35 +++--- 9 files changed, 183 insertions(+), 210 deletions(-) delete mode 100644 bot_bottle/orchestrator/supervisor/queue.py create mode 100644 bot_bottle/orchestrator/supervisor/util.py diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index bf6a14d..452af25 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -39,15 +39,9 @@ from .supervisor import ( STATUS_REJECTED, TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK, - archive_all_proposals, - list_all_pending_proposals, - read_proposal, - read_response, + Supervisor, render_diff, sha256_hex, - write_audit_entry, - write_proposal, - write_response, ) @@ -71,10 +65,14 @@ class Orchestrator: registry: RegistryStore, broker: LaunchBroker, sign_secret: bytes, + supervisor: Supervisor | None = None, ) -> None: self.registry = registry self._broker = broker self._secret = sign_secret + # The supervise service (queue + audit I/O). Injectable so tests can + # point it at a temp DB; defaults to the host DB. + self._supervisor = supervisor or Supervisor() # Per-bottle egress auth tokens (env_name -> value), keyed by bottle_id. # Held **in memory only** — never written to the registry DB — so the # gateway can inject each bottle's upstream credential without secrets @@ -134,7 +132,7 @@ class Orchestrator: self.registry.deregister(bottle_id) self._tokens.pop(bottle_id, None) # Reap any supervise proposals the gone bottle never acknowledged. - archive_all_proposals(bottle_id) + self._supervisor.archive_all_proposals(bottle_id) return True def reconcile( @@ -160,7 +158,7 @@ class Orchestrator: for rec in reaped: self._tokens.pop(rec.bottle_id, None) # Reap any supervise proposals the gone bottle never acknowledged. - archive_all_proposals(rec.bottle_id) + self._supervisor.archive_all_proposals(rec.bottle_id) return [rec.bottle_id for rec in reaped] def tokens_for(self, bottle_id: str) -> dict[str, str]: @@ -220,7 +218,7 @@ class Orchestrator: justification=justification, current_file_hash=sha256_hex(proposed_file), ) - write_proposal(proposal) + self._supervisor.write_proposal(proposal) return proposal.id def supervise_poll_response(self, bottle_id: str, proposal_id: str) -> dict[str, object]: @@ -241,10 +239,10 @@ class Orchestrator: Reads are scoped to `bottle_id` (the queue key), so a caller can never read another bottle's proposal even with a guessed id.""" try: - response = read_response(bottle_id, proposal_id) + response = self._supervisor.read_response(bottle_id, proposal_id) except FileNotFoundError: try: - read_proposal(bottle_id, proposal_id) + self._supervisor.read_proposal(bottle_id, proposal_id) except FileNotFoundError: return {"status": POLL_STATUS_UNKNOWN} return {"status": POLL_STATUS_PENDING} @@ -263,7 +261,7 @@ class Orchestrator: orchestrator-assigned bottle_id, which is opaque to an operator). The CLI renders the label but still responds against `bottle_slug`.""" out: list[dict[str, object]] = [] - for p in list_all_pending_proposals(): + for p in self._supervisor.list_all_pending_proposals(): d = p.to_dict() d["bottle_label"] = self._label_for(p.bottle_slug) out.append(d) @@ -326,7 +324,7 @@ class Orchestrator: if status is None: return False, f"unknown decision {decision!r}" try: - proposal = read_proposal(bottle_slug, proposal_id) + proposal = self._supervisor.read_proposal(bottle_slug, proposal_id) except FileNotFoundError: return False, "no such proposal" @@ -342,12 +340,12 @@ class Orchestrator: diff_before, diff_after = rec.policy, new_policy self.set_policy(rec.bottle_id, new_policy) - write_response(bottle_slug, Response( + self._supervisor.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( + self._supervisor.write_audit_entry(AuditEntry( timestamp=datetime.now(timezone.utc).isoformat(), bottle_slug=bottle_slug, component=component, diff --git a/bot_bottle/orchestrator/supervisor/__init__.py b/bot_bottle/orchestrator/supervisor/__init__.py index 48b96b0..8a2d912 100644 --- a/bot_bottle/orchestrator/supervisor/__init__.py +++ b/bot_bottle/orchestrator/supervisor/__init__.py @@ -1,52 +1,91 @@ -"""The orchestrator's view of the supervise plane. +"""The orchestrator's supervise service. -Bundles the orchestrator-owned supervise surface: the queue/audit I/O and diff -rendering (`queue`), and the `Supervisor` lifecycle (host-side database -staging). For convenience it re-exports the neutral vocabulary -(`bot_bottle.supervisor`'s types + `SupervisePlan`) so orchestrator-side -callers — service, CLI, tests — import from one place. +`Supervisor` is the service object the `Orchestrator` owns and calls: it wraps +the supervise queue + audit stores (queue proposals, record operator responses, +write audit entries, stage the DB at launch). Binding those operations to one +object — optionally scoped to a `db_path` — makes the orchestrator's supervise +dependency explicit and injectable in tests. -The data plane must NOT import this package; it imports `bot_bottle.supervisor` -(neutral) directly and reaches the queue over the control-plane RPC. +For convenience this package also re-exports the neutral vocabulary +(`bot_bottle.supervisor`'s types + `SupervisePlan`) and the pure `render_diff` +/ `sha256_hex` helpers (`util`), so orchestrator-side callers import from one +place. The data plane must NOT import this package — it imports +`bot_bottle.supervisor` (neutral) directly and reaches the queue over the +control-plane RPC. """ from __future__ import annotations -from abc import ABC from pathlib import Path from ...supervisor.types import * # noqa: F401,F403 — re-export the neutral vocabulary from ...supervisor.plan import SupervisePlan from ..store.store_manager import StoreManager -from .queue import ( - archive_all_proposals, - list_all_pending_proposals, - list_pending_proposals, - read_audit_entries, - read_proposal, - read_response, - render_diff, - sha256_hex, - write_audit_entry, - write_proposal, - write_response, -) +from ..store.queue_store import QueueStore +from ...store.audit_store import AuditStore +from ...supervisor.types import AuditEntry, Proposal, Response +from .util import render_diff, sha256_hex # noqa: F401 — re-exported for callers -class Supervisor(ABC): - """Per-bottle supervise lifecycle. Encapsulates host-side database staging; - the gateway's start/stop lifecycle is backend-specific. +class Supervisor: + """Host-side supervise service: the queue + audit I/O and launch-time DB + staging the orchestrator drives. Stateless apart from an optional `db_path` + (None → the host DB, resolved per store as before), so a test can point a + `Supervisor` at a temp database.""" - `prepare` migrates the orchestrator's `bot-bottle.db` and returns the - neutral `SupervisePlan`. It touches the orchestrator store manager, so it - lives here rather than in the neutral `bot_bottle.supervisor` package; the - backend (which drives launch) may import it — backend → orchestrator is an - allowed direction, unlike gateway → orchestrator.""" + def __init__(self, db_path: Path | None = None) -> None: + self._db_path = db_path + + def _queue(self, queue_key: str) -> QueueStore: + return QueueStore(queue_key, self._db_path) + + def _audit(self) -> AuditStore: + return AuditStore(self._db_path) + + # --- Queue I/O --------------------------------------------------------- + + def write_proposal(self, proposal: Proposal) -> Path: + """Persist `proposal` in the queue database, mode 0o600.""" + return self._queue(proposal.bottle_slug).write_proposal(proposal) + + def read_proposal(self, bottle_slug: str, proposal_id: str) -> Proposal: + return self._queue(bottle_slug).read_proposal(proposal_id) + + def list_pending_proposals(self, bottle_slug: str) -> list[Proposal]: + """A single bottle's undecided proposals, FIFO by arrival.""" + return self._queue(bottle_slug).list_pending_proposals() + + def list_all_pending_proposals(self) -> list[Proposal]: + """All pending proposals across bottles, sorted FIFO.""" + return self._queue("").list_all_pending_proposals() + + def write_response(self, bottle_slug: str, response: Response) -> Path: + return self._queue(bottle_slug).write_response(response) + + def read_response(self, bottle_slug: str, proposal_id: str) -> Response: + return self._queue(bottle_slug).read_response(proposal_id) + + def archive_all_proposals(self, bottle_slug: str) -> None: + """Archive every proposal + response for `bottle_slug` (bottle teardown + / reconcile cleanup). Idempotent.""" + self._queue(bottle_slug).archive_all() + + # --- Audit log --------------------------------------------------------- + + def write_audit_entry(self, entry: AuditEntry) -> Path: + """Append `entry` to the host supervise audit table.""" + return self._audit().write_audit_entry(entry) + + def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]: + """Load all audit entries for the given component+slug.""" + return self._audit().read_audit_entries(component, slug) + + # --- Launch-time staging ---------------------------------------------- def prepare(self, slug: str, stage_dir: Path) -> SupervisePlan: - """Stage the host database. Returns the plan; `internal_network` must be - set by the launch step before .start runs.""" + """Stage the host database and return the plan. Called by the backend at + launch; `internal_network` is set by the launch step before `.start`.""" del stage_dir - mgr = StoreManager.instance() + mgr = StoreManager(self._db_path) if self._db_path else StoreManager.instance() mgr.migrate() return SupervisePlan(slug=slug, db_path=mgr.db_path) diff --git a/bot_bottle/orchestrator/supervisor/queue.py b/bot_bottle/orchestrator/supervisor/queue.py deleted file mode 100644 index 17b0c7f..0000000 --- a/bot_bottle/orchestrator/supervisor/queue.py +++ /dev/null @@ -1,93 +0,0 @@ -"""Orchestrator-side supervise queue + audit I/O and diff rendering. - -The host-side library the orchestrator uses to drive the supervise flow: read / -write proposals and responses against the queue store, append audit entries, -and render operator diffs. Only the orchestrator opens `bot-bottle.db` (PRD -0070), so this lives under `orchestrator/` — the data plane reaches it over the -control-plane RPC, never by importing this module. -""" - -from __future__ import annotations - -import difflib -import hashlib -from pathlib import Path - -from ...supervisor.types import AuditEntry, Proposal, Response -from ..store.queue_store import QueueStore -from ...store.audit_store import AuditStore - - -# --- Queue I/O ------------------------------------------------------------- - - -def write_proposal(proposal: Proposal) -> Path: - """Persist `proposal` in the queue database, mode 0o600. - Directory is created if missing.""" - return QueueStore(proposal.bottle_slug).write_proposal(proposal) - - -def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal: - return QueueStore(bottle_slug).read_proposal(proposal_id) - - -def list_pending_proposals(bottle_slug: str) -> list[Proposal]: - """All proposals for `bottle_slug` that do not yet have a matching - response. Sorted by `arrival_timestamp` so the operator sees the queue - FIFO.""" - return QueueStore(bottle_slug).list_pending_proposals() - - -def list_all_pending_proposals() -> list[Proposal]: - """All pending proposals across bottles, sorted FIFO.""" - return QueueStore("").list_all_pending_proposals() - - -def write_response(bottle_slug: str, response: Response) -> Path: - return QueueStore(bottle_slug).write_response(response) - - -def read_response(bottle_slug: str, proposal_id: str) -> Response: - return QueueStore(bottle_slug).read_response(proposal_id) - - -def archive_all_proposals(bottle_slug: str) -> None: - """Archive every proposal + response for `bottle_slug` (bottle teardown / - reconcile cleanup). Idempotent.""" - QueueStore(bottle_slug).archive_all() - - -# --- Audit log ------------------------------------------------------------- - - -def write_audit_entry(entry: AuditEntry) -> Path: - """Append `entry` to the host supervise audit table.""" - return AuditStore().write_audit_entry(entry) - - -def read_audit_entries(component: str, slug: str) -> list[AuditEntry]: - """Load all audit entries for the given component+slug.""" - return AuditStore().read_audit_entries(component, slug) - - -# --- Diff rendering -------------------------------------------------------- - - -def render_diff(before: str, after: str, *, label: str = "config") -> str: - """Unified diff suitable for the audit log + TUI. Empty diff (no changes) - renders as the empty string.""" - diff = difflib.unified_diff( - before.splitlines(keepends=True), - after.splitlines(keepends=True), - fromfile=f"{label} (current)", - tofile=f"{label} (proposed)", - lineterm="", - ) - parts = list(diff) - if not parts: - return "" - return "".join(p if p.endswith("\n") else p + "\n" for p in parts).rstrip("\n") - - -def sha256_hex(content: str) -> str: - return hashlib.sha256(content.encode("utf-8")).hexdigest() diff --git a/bot_bottle/orchestrator/supervisor/util.py b/bot_bottle/orchestrator/supervisor/util.py new file mode 100644 index 0000000..3776b56 --- /dev/null +++ b/bot_bottle/orchestrator/supervisor/util.py @@ -0,0 +1,31 @@ +"""Pure supervise helpers — diff rendering and content hashing. + +Stateless utilities used alongside the `Supervisor` service. Kept as plain +functions (not methods) because they carry no state and are also handy to +callers that only need to hash or diff. +""" + +from __future__ import annotations + +import difflib +import hashlib + + +def render_diff(before: str, after: str, *, label: str = "config") -> str: + """Unified diff suitable for the audit log + TUI. Empty diff (no changes) + renders as the empty string.""" + diff = difflib.unified_diff( + before.splitlines(keepends=True), + after.splitlines(keepends=True), + fromfile=f"{label} (current)", + tofile=f"{label} (proposed)", + lineterm="", + ) + parts = list(diff) + if not parts: + return "" + return "".join(p if p.endswith("\n") else p + "\n" for p in parts).rstrip("\n") + + +def sha256_hex(content: str) -> str: + return hashlib.sha256(content.encode("utf-8")).hexdigest() diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 0bacd90..66119c3 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -29,7 +29,7 @@ from bot_bottle.orchestrator.supervisor import ( Proposal, TOOL_EGRESS_ALLOW, sha256_hex, - write_proposal, + Supervisor, ) @@ -421,7 +421,7 @@ class TestDispatchSupervise(unittest.TestCase): 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) + Supervisor().write_proposal(p) return p.id def test_list_pending(self) -> None: diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index a83f8c1..d090f3a 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -20,9 +20,8 @@ from bot_bottle.orchestrator.supervisor import ( Proposal, STATUS_APPROVED, TOOL_EGRESS_ALLOW, - read_response, sha256_hex, - write_proposal, + Supervisor, ) @@ -185,7 +184,7 @@ class TestOrchestratorSupervise(unittest.TestCase): 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) + Supervisor().write_proposal(p) return p.id def test_pending_lists_queued_proposal(self) -> None: @@ -208,7 +207,7 @@ class TestOrchestratorSupervise(unittest.TestCase): 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(STATUS_APPROVED, Supervisor().read_response("demo", pid).status) self.assertEqual([], self.orch.supervise_pending()) def test_pending_carries_human_label(self) -> None: @@ -261,7 +260,7 @@ class TestOrchestratorSupervise(unittest.TestCase): 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) + self.assertEqual("rejected", Supervisor().read_response("demo", pid).status) def test_unknown_proposal_is_error(self) -> None: ok, err = self.orch.supervise_respond( diff --git a/tests/unit/test_supervise.py b/tests/unit/test_supervise.py index da2b5f0..c609dbe 100644 --- a/tests/unit/test_supervise.py +++ b/tests/unit/test_supervise.py @@ -17,19 +17,15 @@ from bot_bottle.orchestrator.supervisor import ( STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED, + Supervisor, TOOL_EGRESS_ALLOW, TOOL_GITLEAKS_ALLOW, - list_pending_proposals, - read_audit_entries, - read_proposal, - read_response, render_diff, sha256_hex, - write_audit_entry, - write_proposal, - write_response, ) +_SV = Supervisor() + FIXED_TS = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc) @@ -123,26 +119,26 @@ class TestQueueIO(unittest.TestCase): def test_write_and_read_proposal(self): p = _proposal() - path = write_proposal(p) + path = _SV.write_proposal(p) self.assertTrue(path.exists()) self.assertEqual(host_db_path(), path) self.assertEqual(0o600, path.stat().st_mode & 0o777) - loaded = read_proposal(self.slug, p.id) + loaded = _SV.read_proposal(self.slug, p.id) self.assertEqual(p, loaded) def test_list_pending_excludes_responded(self): a = _proposal(justification="first") b = _proposal(justification="second") - write_proposal(a) - write_proposal(b) - write_response(self.slug, Response( + _SV.write_proposal(a) + _SV.write_proposal(b) + _SV.write_response(self.slug, Response( proposal_id=a.id, status=STATUS_APPROVED, notes="", )) - pending = list_pending_proposals(self.slug) + pending = _SV.list_pending_proposals(self.slug) self.assertEqual([b.id], [p.id for p in pending]) def test_list_pending_returns_empty_for_missing_slug(self): - self.assertEqual([], list_pending_proposals("nope")) + self.assertEqual([], _SV.list_pending_proposals("nope")) def test_list_pending_sorted_by_arrival(self): # Fabricate two with explicit timestamps. @@ -159,15 +155,15 @@ class TestQueueIO(unittest.TestCase): now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), ) # Write in reverse order. - write_proposal(b) - write_proposal(a) - ordered = list_pending_proposals(self.slug) + _SV.write_proposal(b) + _SV.write_proposal(a) + ordered = _SV.list_pending_proposals(self.slug) self.assertEqual([a.id, b.id], [p.id for p in ordered]) def test_write_and_read_response(self): r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no") - write_response(self.slug, r) - self.assertEqual(r, read_response(self.slug, "xyz")) + _SV.write_response(self.slug, r) + self.assertEqual(r, _SV.read_response(self.slug, "xyz")) class TestAuditLog(unittest.TestCase): @@ -193,15 +189,15 @@ class TestAuditLog(unittest.TestCase): justification="agent needed gh-api token", diff="--- before\n+++ after\n", ) - path = write_audit_entry(e) + path = _SV.write_audit_entry(e) self.assertEqual(host_db_path(), path) self.assertEqual(0o600, path.stat().st_mode & 0o777) - loaded = read_audit_entries("cred-proxy", "dev") + loaded = _SV.read_audit_entries("cred-proxy", "dev") self.assertEqual([e], loaded) def test_appends_one_line_per_entry(self): for i in range(3): - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp=f"2026-05-25T12:00:0{i}+00:00", bottle_slug="dev", component="egress", @@ -210,7 +206,7 @@ class TestAuditLog(unittest.TestCase): justification="", diff="", )) - entries = read_audit_entries("egress", "dev") + entries = _SV.read_audit_entries("egress", "dev") self.assertEqual(3, len(entries)) self.assertEqual( ["2026-05-25T12:00:00+00:00", "2026-05-25T12:00:01+00:00", @@ -219,7 +215,7 @@ class TestAuditLog(unittest.TestCase): ) def test_separate_logs_per_component_slug(self): - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp="t", bottle_slug="dev", component="cred-proxy", @@ -228,7 +224,7 @@ class TestAuditLog(unittest.TestCase): justification="", diff="", )) - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp="t", bottle_slug="dev", component="egress", @@ -237,7 +233,7 @@ class TestAuditLog(unittest.TestCase): justification="", diff="", )) - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp="t", bottle_slug="other", component="cred-proxy", @@ -246,12 +242,12 @@ class TestAuditLog(unittest.TestCase): justification="", diff="", )) - self.assertEqual(1, len(read_audit_entries("cred-proxy", "dev"))) - self.assertEqual(1, len(read_audit_entries("egress", "dev"))) - self.assertEqual(1, len(read_audit_entries("cred-proxy", "other"))) + self.assertEqual(1, len(_SV.read_audit_entries("cred-proxy", "dev"))) + self.assertEqual(1, len(_SV.read_audit_entries("egress", "dev"))) + self.assertEqual(1, len(_SV.read_audit_entries("cred-proxy", "other"))) def test_read_audit_entries_missing_log_returns_empty(self): - self.assertEqual([], read_audit_entries("cred-proxy", "no-such-bottle")) + self.assertEqual([], _SV.read_audit_entries("cred-proxy", "no-such-bottle")) class TestDiffAndHash(unittest.TestCase): diff --git a/tests/unit/test_supervise_edge.py b/tests/unit/test_supervise_edge.py index 8548040..d29135b 100644 --- a/tests/unit/test_supervise_edge.py +++ b/tests/unit/test_supervise_edge.py @@ -17,14 +17,12 @@ from bot_bottle.orchestrator.supervisor import ( AuditEntry, Proposal, STATUS_APPROVED, + Supervisor, TOOL_EGRESS_ALLOW, - list_pending_proposals, - read_audit_entries, - read_proposal, - read_response, - write_audit_entry, ) +_SV = Supervisor() + def _proposal() -> Proposal: return Proposal.new( @@ -47,21 +45,21 @@ class TestReadMalformed(unittest.TestCase): with patch.dict("os.environ", {"HOME": d}): QueueStore("slug").migrate() with self.assertRaises(FileNotFoundError): - read_proposal("slug", "p") + _SV.read_proposal("slug", "p") def test_read_response_missing_row(self) -> None: with tempfile.TemporaryDirectory() as d: with patch.dict("os.environ", {"HOME": d}): QueueStore("slug").migrate() with self.assertRaises(FileNotFoundError): - read_response("slug", "p") + _SV.read_response("slug", "p") def test_list_pending_reads_db_only(self) -> None: with tempfile.TemporaryDirectory() as d: with patch.dict("os.environ", {"HOME": d}): QueueStore("slug").migrate() - supervise.write_proposal(_proposal()) - pending = list_pending_proposals("slug") + _SV.write_proposal(_proposal()) + pending = _SV.list_pending_proposals("slug") self.assertEqual(1, len(pending)) self.assertEqual("slug", pending[0].bottle_slug) @@ -70,26 +68,26 @@ class TestReadMalformed(unittest.TestCase): with patch.dict("os.environ", {"HOME": d}): QueueStore("slug").migrate() p = _proposal() - supervise.write_proposal(p) - supervise.write_response("slug", supervise.Response( + _SV.write_proposal(p) + _SV.write_response("slug", supervise.Response( proposal_id=p.id, status=STATUS_APPROVED, notes="", )) - self.assertEqual([], list_pending_proposals("slug")) + self.assertEqual([], _SV.list_pending_proposals("slug")) class TestReadAuditEntries(unittest.TestCase): def test_missing_log_returns_empty(self) -> None: with tempfile.TemporaryDirectory() as home, \ patch.dict("os.environ", {"HOME": home}): - self.assertEqual([], read_audit_entries("egress", "nope")) + self.assertEqual([], _SV.read_audit_entries("egress", "nope")) def test_reads_entries_from_db(self) -> None: with tempfile.TemporaryDirectory() as home, \ patch.dict("os.environ", {"HOME": home}): AuditStore().migrate() - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp="t", bottle_slug="slug", component="egress", @@ -98,7 +96,7 @@ class TestReadAuditEntries(unittest.TestCase): justification="", diff="", )) - write_audit_entry(AuditEntry( + _SV.write_audit_entry(AuditEntry( timestamp="t", bottle_slug="other", component="egress", @@ -107,7 +105,7 @@ class TestReadAuditEntries(unittest.TestCase): justification="", diff="", )) - entries = read_audit_entries("egress", "slug") + entries = _SV.read_audit_entries("egress", "slug") self.assertEqual(1, len(entries)) self.assertEqual("approve", entries[0].operator_action) diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index a082998..fdaff90 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -50,6 +50,11 @@ from bot_bottle.gateway.supervise_server import ( validate_proposed_file, ) +# The orchestrator's supervise service, backing the fake resolver's queue reads +# and writes (the fake stands in for the RPC; state still goes through the real +# queue store). +_SV = _sv.Supervisor() + # Fixed caller identity for the handler tests. The control plane attributes by # (source_ip, identity_token); the fake resolver ignores them and answers for a # fixed bottle, since attribution itself is covered by the orchestrator tests. @@ -89,7 +94,7 @@ class _FakeSuperviseResolver: bottle_slug=self.bottle_id, tool=tool, proposed_file=proposed_file, justification=justification, current_file_hash=_sv.sha256_hex(proposed_file), ) - _sv.write_proposal(proposal) + _SV.write_proposal(proposal) return proposal.id def poll_supervise( @@ -101,10 +106,10 @@ class _FakeSuperviseResolver: if self.bottle_id is None or self.poll_none: return None try: - response = _sv.read_response(self.bottle_id, proposal_id) + response = _SV.read_response(self.bottle_id, proposal_id) except FileNotFoundError: try: - _sv.read_proposal(self.bottle_id, proposal_id) + _SV.read_proposal(self.bottle_id, proposal_id) except FileNotFoundError: return {"status": _sv.POLL_STATUS_UNKNOWN} return {"status": _sv.POLL_STATUS_PENDING} @@ -377,10 +382,10 @@ class TestHandleToolsCall(unittest.TestCase): matching response — the operator half, out of band.""" def runner(): for _ in range(200): - pending = _sv.list_pending_proposals("dev") + pending = _SV.list_pending_proposals("dev") if pending: p = pending[0] - _sv.write_response("dev", _sv.Response( + _SV.write_response("dev", _sv.Response( proposal_id=p.id, status=status, notes=notes, )) return @@ -487,7 +492,7 @@ class TestHandleToolsCall(unittest.TestCase): responder.join() # A decided proposal drops off the operator's pending list (a response # row exists) — poll itself no longer archives (issue #469 review). - self.assertEqual([], _sv.list_pending_proposals("dev")) + self.assertEqual([], _SV.list_pending_proposals("dev")) def test_pending_response_times_out_without_archive(self): result = _tools_call( @@ -505,7 +510,7 @@ class TestHandleToolsCall(unittest.TestCase): text = result["content"][0]["text"] # type: ignore[index] self.assertIn("status: pending", text) self.assertIn("proposal remains queued", text) - self.assertEqual(1, len(_sv.list_pending_proposals("dev"))) + self.assertEqual(1, len(_SV.list_pending_proposals("dev"))) _ALLOW: dict[str, object] = { "name": _sv.TOOL_EGRESS_ALLOW, @@ -747,7 +752,7 @@ class TestNonBlockingSupervise(unittest.TestCase): justification="need example.com", current_file_hash=_sv.sha256_hex(self._ROUTES), ) - _sv.write_proposal(p) + _SV.write_proposal(p) return p # --- pending response carries the id --- @@ -771,7 +776,7 @@ class TestNonBlockingSupervise(unittest.TestCase): self.assertFalse(result["isError"]) # type: ignore[index] text = result["content"][0]["text"] # type: ignore[index] self.assertIn("status: pending", text) - pending = _sv.list_pending_proposals("dev") + pending = _SV.list_pending_proposals("dev") self.assertEqual(1, len(pending)) # still queued, not archived self.assertIn(pending[0].id, text) # agent got the id to poll @@ -779,7 +784,7 @@ class TestNonBlockingSupervise(unittest.TestCase): def test_check_returns_approved_idempotently(self): p = self._seed_proposal() - _sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok")) + _SV.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok")) result = _check(self.resolver, {"arguments": {"proposal_id": p.id}}) self.assertFalse(result["isError"]) text = result["content"][0]["text"] # type: ignore[index] @@ -792,7 +797,7 @@ class TestNonBlockingSupervise(unittest.TestCase): def test_check_rejected_sets_isError(self): p = self._seed_proposal() - _sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_REJECTED, notes="no")) + _SV.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_REJECTED, notes="no")) result = _check(self.resolver, {"arguments": {"proposal_id": p.id}}) self.assertTrue(result["isError"]) self.assertIn("status: rejected", result["content"][0]["text"]) # type: ignore[index] @@ -804,7 +809,7 @@ class TestNonBlockingSupervise(unittest.TestCase): text = result["content"][0]["text"] # type: ignore[index] self.assertIn("status: pending", text) self.assertIn(p.id, text) - self.assertEqual(1, len(_sv.list_pending_proposals("dev"))) # not archived + self.assertEqual(1, len(_SV.list_pending_proposals("dev"))) # not archived def test_check_unknown_id_is_error(self): result = _check(self.resolver, {"arguments": {"proposal_id": "no-such-proposal"}}) @@ -848,15 +853,15 @@ class TestNonBlockingSupervise(unittest.TestCase): }, ServerConfig(response_timeout_seconds=0.05), ) - pid = _sv.list_pending_proposals("dev")[0].id + pid = _SV.list_pending_proposals("dev")[0].id self.assertIn(pid, result["content"][0]["text"]) # type: ignore[index] # 2. operator decides out-of-band - _sv.write_response("dev", _sv.Response(proposal_id=pid, status=_sv.STATUS_APPROVED, notes="ok")) + _SV.write_response("dev", _sv.Response(proposal_id=pid, status=_sv.STATUS_APPROVED, notes="ok")) # 3. agent resumes by polling — no re-proposing poll = _check(self.resolver, {"arguments": {"proposal_id": pid}}) self.assertFalse(poll["isError"]) self.assertIn("status: approved", poll["content"][0]["text"]) # type: ignore[index] - self.assertEqual([], _sv.list_pending_proposals("dev")) # resolved (response exists) + self.assertEqual([], _SV.list_pending_proposals("dev")) # resolved (response exists) if __name__ == "__main__":