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,

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)

bot_bottle/orchestrator/supervisor/queue.py

@@ -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()

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()

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:

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(

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):

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)
 

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__":