refactor(supervise): drop obsolete queue helpers (archive_proposal, wait_for_response, audit_dir/path)
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / integration-docker (pull_request) Successful in 17s
lint / lint (push) Failing after 59s
test / unit (pull_request) Successful in 2m3s
test / integration-firecracker (pull_request) Successful in 3m28s
test / coverage (pull_request) Successful in 16s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / integration-docker (pull_request) Successful in 17s
lint / lint (push) Failing after 59s
test / unit (pull_request) Successful in 2m3s
test / integration-firecracker (pull_request) Successful in 3m28s
test / coverage (pull_request) Successful in 16s
test / publish-infra (pull_request) Has been skipped
Remove the supervise queue helpers that no live path uses anymore — they're
mechanisms the PRD-0070 / idempotent-poll migration superseded:
- archive_proposal (+ QueueStore.archive_proposal): the poll stopped archiving
on read (#469 review), so single-proposal archiving has no caller; decided
proposals drop off the pending list via their response row and are reaped in
bulk by archive_all_proposals on teardown/reconcile.
- wait_for_response: the data plane polls the queue over the control-plane RPC
now, so nothing block-waits on it.
- audit_dir / audit_log_path: file-based audit paths, replaced by the SQLite
AuditStore.
Also drops the tests that covered those obsolete helpers. Kept the prod-unused
read accessors list_pending_proposals and read_audit_entries: those aren't dead
mechanisms — they're the read side of live write paths that many tests use to
verify real queue/audit behavior.
Full unit suite green (2243).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -167,25 +167,6 @@ class QueueStore(DbStore):
|
||||
raise FileNotFoundError(proposal_id)
|
||||
return self._row_to_response(row)
|
||||
|
||||
def archive_proposal(self, proposal_id: str) -> None:
|
||||
if not self.db_path.is_file():
|
||||
return
|
||||
with self._connection() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE supervise_proposals SET archived = 1
|
||||
WHERE queue_key = ? AND id = ?
|
||||
""",
|
||||
(self.queue_key, proposal_id),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE supervise_responses SET archived = 1
|
||||
WHERE queue_key = ? AND proposal_id = ?
|
||||
""",
|
||||
(self.queue_key, proposal_id),
|
||||
)
|
||||
|
||||
def archive_all(self) -> None:
|
||||
"""Archive every proposal + response for this queue key. Used to reap a
|
||||
bottle's supervise records when it is torn down or reconciled away —
|
||||
|
||||
@@ -20,9 +20,6 @@ from ...supervisor.plan import SupervisePlan
|
||||
from ..store.store_manager import StoreManager
|
||||
from .queue import (
|
||||
archive_all_proposals,
|
||||
archive_proposal,
|
||||
audit_dir,
|
||||
audit_log_path,
|
||||
list_all_pending_proposals,
|
||||
list_pending_proposals,
|
||||
read_audit_entries,
|
||||
@@ -30,7 +27,6 @@ from .queue import (
|
||||
read_response,
|
||||
render_diff,
|
||||
sha256_hex,
|
||||
wait_for_response,
|
||||
write_audit_entry,
|
||||
write_proposal,
|
||||
write_response,
|
||||
|
||||
@@ -11,31 +11,13 @@ from __future__ import annotations
|
||||
|
||||
import difflib
|
||||
import hashlib
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from ...supervisor.types import (
|
||||
AuditEntry,
|
||||
DEFAULT_POLL_INTERVAL_SEC,
|
||||
Proposal,
|
||||
Response,
|
||||
)
|
||||
from ...paths import bot_bottle_root
|
||||
from ...supervisor.types import AuditEntry, Proposal, Response
|
||||
from ..store.queue_store import QueueStore
|
||||
from ...store.audit_store import AuditStore
|
||||
|
||||
|
||||
# --- Paths -----------------------------------------------------------------
|
||||
|
||||
|
||||
def audit_dir() -> Path:
|
||||
return bot_bottle_root() / "audit"
|
||||
|
||||
|
||||
def audit_log_path(component: str, slug: str) -> Path:
|
||||
return audit_dir() / f"{component}-{slug}.log"
|
||||
|
||||
|
||||
# --- Queue I/O -------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -69,36 +51,6 @@ def read_response(bottle_slug: str, proposal_id: str) -> Response:
|
||||
return QueueStore(bottle_slug).read_response(proposal_id)
|
||||
|
||||
|
||||
def wait_for_response(
|
||||
bottle_slug: str,
|
||||
proposal_id: str,
|
||||
*,
|
||||
poll_interval: float = DEFAULT_POLL_INTERVAL_SEC,
|
||||
deadline: float | None = None,
|
||||
) -> Response:
|
||||
"""Block until a response file appears for `proposal_id`, then return it.
|
||||
`deadline` is an absolute time.monotonic() value after which the wait raises
|
||||
TimeoutError. None waits forever — the natural shape, since the operator's
|
||||
response time is unbounded.
|
||||
|
||||
Polls SQLite so the implementation stays portable and stdlib-only."""
|
||||
store = QueueStore(bottle_slug)
|
||||
while True:
|
||||
try:
|
||||
return store.read_response(proposal_id)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
if deadline is not None and time.monotonic() >= deadline:
|
||||
raise TimeoutError(f"no response for proposal {proposal_id!r}")
|
||||
time.sleep(poll_interval)
|
||||
|
||||
|
||||
def archive_proposal(bottle_slug: str, proposal_id: str) -> None:
|
||||
"""Mark both proposal and response rows processed.
|
||||
Idempotent — missing rows are silently skipped."""
|
||||
QueueStore(bottle_slug).archive_proposal(proposal_id)
|
||||
|
||||
|
||||
def archive_all_proposals(bottle_slug: str) -> None:
|
||||
"""Archive every proposal + response for `bottle_slug` (bottle teardown /
|
||||
reconcile cleanup). Idempotent."""
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
"""Unit: supervise queue + audit log + diff helpers (PRD 0013)."""
|
||||
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
@@ -21,14 +19,12 @@ from bot_bottle.orchestrator.supervisor import (
|
||||
STATUS_REJECTED,
|
||||
TOOL_EGRESS_ALLOW,
|
||||
TOOL_GITLEAKS_ALLOW,
|
||||
archive_proposal,
|
||||
list_pending_proposals,
|
||||
read_audit_entries,
|
||||
read_proposal,
|
||||
read_response,
|
||||
render_diff,
|
||||
sha256_hex,
|
||||
wait_for_response,
|
||||
write_audit_entry,
|
||||
write_proposal,
|
||||
write_response,
|
||||
@@ -173,48 +169,6 @@ class TestQueueIO(unittest.TestCase):
|
||||
write_response(self.slug, r)
|
||||
self.assertEqual(r, read_response(self.slug, "xyz"))
|
||||
|
||||
def test_wait_for_response_returns_when_file_appears(self):
|
||||
p = _proposal()
|
||||
write_proposal(p)
|
||||
|
||||
def write_after_delay():
|
||||
time.sleep(0.05)
|
||||
write_response(self.slug, Response(
|
||||
proposal_id=p.id, status=STATUS_APPROVED, notes="ok",
|
||||
))
|
||||
|
||||
t = threading.Thread(target=write_after_delay)
|
||||
t.start()
|
||||
try:
|
||||
r = wait_for_response(self.slug, p.id, poll_interval=0.01)
|
||||
finally:
|
||||
t.join()
|
||||
self.assertEqual(STATUS_APPROVED, r.status)
|
||||
self.assertEqual("ok", r.notes)
|
||||
|
||||
def test_wait_for_response_times_out(self):
|
||||
deadline = time.monotonic() + 0.05
|
||||
with self.assertRaises(TimeoutError):
|
||||
wait_for_response(
|
||||
self.slug, "never",
|
||||
poll_interval=0.01, deadline=deadline,
|
||||
)
|
||||
|
||||
def test_archive_proposal_hides_rows(self):
|
||||
p = _proposal()
|
||||
write_proposal(p)
|
||||
write_response(self.slug, Response(
|
||||
proposal_id=p.id, status=STATUS_APPROVED, notes="",
|
||||
))
|
||||
archive_proposal(self.slug, p.id)
|
||||
self.assertEqual([], list_pending_proposals(self.slug))
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
read_response(self.slug, p.id)
|
||||
|
||||
def test_archive_is_idempotent_on_missing_files(self):
|
||||
# Should not raise.
|
||||
archive_proposal(self.slug, "nope")
|
||||
|
||||
|
||||
class TestAuditLog(unittest.TestCase):
|
||||
def setUp(self):
|
||||
|
||||
@@ -5,7 +5,6 @@ fallback paths."""
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
@@ -23,7 +22,6 @@ from bot_bottle.orchestrator.supervisor import (
|
||||
read_audit_entries,
|
||||
read_proposal,
|
||||
read_response,
|
||||
wait_for_response,
|
||||
write_audit_entry,
|
||||
)
|
||||
|
||||
@@ -81,22 +79,6 @@ class TestReadMalformed(unittest.TestCase):
|
||||
self.assertEqual([], list_pending_proposals("slug"))
|
||||
|
||||
|
||||
class TestWaitForResponse(unittest.TestCase):
|
||||
def test_missing_response_times_out(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
with patch.dict("os.environ", {"HOME": d}):
|
||||
QueueStore("slug").migrate()
|
||||
with self.assertRaises(TimeoutError):
|
||||
wait_for_response("slug", "p", deadline=time.monotonic())
|
||||
|
||||
def test_empty_db_response_does_not_count(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
with patch.dict("os.environ", {"HOME": d}):
|
||||
QueueStore("slug").migrate()
|
||||
with self.assertRaises(TimeoutError):
|
||||
wait_for_response("slug", "p", deadline=time.monotonic())
|
||||
|
||||
|
||||
class TestReadAuditEntries(unittest.TestCase):
|
||||
def test_missing_log_returns_empty(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as home, \
|
||||
@@ -129,19 +111,6 @@ class TestReadAuditEntries(unittest.TestCase):
|
||||
self.assertEqual(1, len(entries))
|
||||
self.assertEqual("approve", entries[0].operator_action)
|
||||
|
||||
def test_legacy_audit_log_file_does_not_count(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as home, \
|
||||
patch.dict("os.environ", {"HOME": home}):
|
||||
path = supervise.audit_log_path("egress", "slug")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(
|
||||
'{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
|
||||
' "operator_action": "approve", "operator_notes": "",'
|
||||
' "justification": "", "diff": ""}\n'
|
||||
)
|
||||
entries = read_audit_entries("egress", "slug")
|
||||
self.assertEqual([], entries)
|
||||
|
||||
|
||||
class TestStoreGuardBranches(unittest.TestCase):
|
||||
"""Direct QueueStore / AuditStore construction and early-return guard branches."""
|
||||
@@ -170,14 +139,6 @@ class TestStoreGuardBranches(unittest.TestCase):
|
||||
db.unlink()
|
||||
self.assertEqual([], store.list_all_pending_proposals())
|
||||
|
||||
def test_queue_store_missing_db_archive_is_noop(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
db = Path(d) / "q.db"
|
||||
store = QueueStore("key", db_path=db)
|
||||
store.migrate()
|
||||
db.unlink()
|
||||
store.archive_proposal("anything") # must not raise
|
||||
|
||||
def test_queue_store_chmod_oserror_is_swallowed(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
db = Path(d) / "q.db"
|
||||
|
||||
Reference in New Issue
Block a user