PRD: SQLite local storage #320

Merged
didericis merged 16 commits from sqlite-local-storage into main 2026-07-06 15:27:32 -04:00
31 changed files with 1292 additions and 614 deletions
+8 -2
View File
@@ -18,7 +18,7 @@
# /git-gate-entrypoint.sh docker-cp'd at start time
# /git-gate/creds/* docker-cp'd at start time
# /git/* bare repos, populated at runtime
# /run/supervise/queue/ bind-mounted at run time
# /run/supervise/bot-bottle.db bind-mounted at run time
# /home/mitmproxy/.mitmproxy/ mitmproxy CA dir
#
# Exposed ports inside the container:
@@ -66,6 +66,12 @@ COPY bot_bottle/egress_dlp_config.py /app/egress_dlp_config.py
COPY bot_bottle/egress_addon.py /app/egress_addon.py
COPY bot_bottle/dlp_detectors.py /app/dlp_detectors.py
COPY bot_bottle/yaml_subset.py /app/yaml_subset.py
COPY bot_bottle/migrations.py /app/migrations.py
COPY bot_bottle/db_store.py /app/db_store.py
COPY bot_bottle/supervise_types.py /app/supervise_types.py
COPY bot_bottle/queue_store.py /app/queue_store.py
COPY bot_bottle/audit_store.py /app/audit_store.py
COPY bot_bottle/store_manager.py /app/store_manager.py
COPY bot_bottle/supervise.py /app/supervise.py
COPY bot_bottle/supervise_server.py /app/supervise_server.py
COPY bot_bottle/sidecar_init.py /app/sidecar_init.py
@@ -81,7 +87,7 @@ RUN mkdir -p \
/etc/git-gate \
/git-gate/creds \
/git \
/run/supervise/queue \
/run/supervise \
/home/mitmproxy/.mitmproxy
# Documentation only — the compose renderer publishes whichever
+91
View File
@@ -0,0 +1,91 @@
"""SQLite-backed audit store for supervise (PRD 0013)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
try:
from .supervise_types import AuditEntry, host_db_path
from .db_store import DbStore
from .migrations import TableMigrations
except ImportError:
from supervise_types import AuditEntry, host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
didericis marked this conversation as resolved
Review

call this get_supervise_mod for legibility.

call this `get_supervise_mod` for legibility.
Review

Done — renamed to get_supervise_mod() in both audit_store.py and queue_store.py (e8e4f6f).

Done — renamed to `get_supervise_mod()` in both `audit_store.py` and `queue_store.py` (`e8e4f6f`).
from db_store import DbStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
class AuditStore(DbStore):
"""SQLite-backed persistent store for supervise audit entries."""
didericis marked this conversation as resolved
Review

The need for two names smells a bit here... why exactly do we need to import from two directions? Would rather create some sort of wrapper for the audit_store if needed to properly resolve dependencies.

The need for two names smells a bit here... why exactly do we need to import from two directions? Would rather create some sort of wrapper for the audit_store if needed to properly resolve dependencies.
Review

The two names exist because the store modules run in two different identity contexts at runtime:

  1. Package context (bot_bottle.audit_store): supervise is at sys.modules["bot_bottle.supervise"]. Tests patch bot_bottle.supervise.bot_bottle_root.
  2. Sidecar / injected-path context (audit_store): supervise.py is copied flat to /app/ with no package. test_supervise_server.py also injects bot_bottle/ into sys.path and imports supervise directly, so patches go on sys.modules["supervise"].

If get_supervise_mod() always returned bot_bottle.supervise, sidecar imports and the server tests would see an unpatched module — the wrong object.

The clean fix is to extract Proposal, Response, AuditEntry, and host_db_path into a standalone _supervise_types.py module that neither supervise.py nor the store files import circularly. That eliminates get_supervise_mod() entirely. Happy to do it as a follow-up or inline here if you prefer — it touches supervise.py substantially.

The two names exist because the store modules run in two different identity contexts at runtime: 1. **Package context** (`bot_bottle.audit_store`): `supervise` is at `sys.modules["bot_bottle.supervise"]`. Tests patch `bot_bottle.supervise.bot_bottle_root`. 2. **Sidecar / injected-path context** (`audit_store`): `supervise.py` is copied flat to `/app/` with no package. `test_supervise_server.py` also injects `bot_bottle/` into `sys.path` and imports `supervise` directly, so patches go on `sys.modules["supervise"]`. If `get_supervise_mod()` always returned `bot_bottle.supervise`, sidecar imports and the server tests would see an unpatched module — the wrong object. The clean fix is to extract `Proposal`, `Response`, `AuditEntry`, and `host_db_path` into a standalone `_supervise_types.py` module that neither `supervise.py` nor the store files import circularly. That eliminates `get_supervise_mod()` entirely. Happy to do it as a follow-up or inline here if you prefer — it touches `supervise.py` substantially.
Review

@didericis-claude can do it now

@didericis-claude can do it now
Review

Done — PR #329 extracted supervise types and eliminated get_supervise_mod(), merged at 9fb83ef.

Done — PR #329 extracted supervise types and eliminated `get_supervise_mod()`, merged at 9fb83ef.
def __init__(self, db_path: Path | None = None) -> None:
didericis marked this conversation as resolved
Review

We can just put this into the audit store arguments, don't think this needs to exist outside of the store initializer. Can follow that pattern for the supervisor store as well and treat the table migrations as hidden implementation details that only live in the store class.

We can just put this into the audit store arguments, don't think this needs to exist outside of the store initializer. Can follow that pattern for the supervisor store as well and treat the table migrations as hidden implementation details that only live in the store class.
Review

Agreed. Will define _MIGRATIONS inline inside AuditStore.__init__ before the super().__init__() call, and follow the same pattern in QueueStore.__init__. TableMigrations stays imported at module level since it's still referenced inside the constructors.

Agreed. Will define `_MIGRATIONS` inline inside `AuditStore.__init__` before the `super().__init__()` call, and follow the same pattern in `QueueStore.__init__`. `TableMigrations` stays imported at module level since it's still referenced inside the constructors.
Review

Done — 041d9bb. _MIGRATIONS is now a local variable inside AuditStore.__init__ and QueueStore.__init__.

Done — `041d9bb`. `_MIGRATIONS` is now a local variable inside `AuditStore.__init__` and `QueueStore.__init__`.
# One entry per schema version: migrations[0] brings a fresh DB to
# version 1, [1] to version 2, etc. Add new entries at the end; never
# edit existing ones.
migrations = TableMigrations("audit_store", [
# v1 — initial schema
"""
CREATE TABLE IF NOT EXISTS supervise_audit_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
didericis marked this conversation as resolved
Review

put this on the AuditStore class

put this on the AuditStore class
Review

Done — moved to AuditStore._row_to_entry (static method). _proposal_from_row and _response_from_row similarly moved to QueueStore._row_to_proposal / _row_to_response.

Done — moved to `AuditStore._row_to_entry` (static method). `_proposal_from_row` and `_response_from_row` similarly moved to `QueueStore._row_to_proposal` / `_row_to_response`.
timestamp TEXT NOT NULL,
bottle_slug TEXT NOT NULL,
component TEXT NOT NULL,
operator_action TEXT NOT NULL,
operator_notes TEXT NOT NULL,
justification TEXT NOT NULL,
diff TEXT NOT NULL
)
""",
])
super().__init__(db_path or host_db_path(), migrations)
def write_audit_entry(self, entry: AuditEntry) -> Path:
didericis marked this conversation as resolved
Review

method for this is unnecessary

method for this is unnecessary
Review

Removed — inlined as db_path or get_supervise_mod().host_db_path() directly in __init__.

Removed — inlined as `db_path or get_supervise_mod().host_db_path()` directly in `__init__`.
with self._connect() as conn:
conn.execute(
"""
INSERT INTO supervise_audit_entries (
timestamp, bottle_slug, component, operator_action,
operator_notes, justification, diff
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
entry.timestamp,
entry.bottle_slug,
entry.component,
entry.operator_action,
entry.operator_notes,
entry.justification,
entry.diff,
),
)
self._chmod()
return self.db_path
def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]:
if not self.db_path.is_file():
return []
with self._connect() as conn:
rows = conn.execute(
"""
SELECT * FROM supervise_audit_entries
WHERE component = ? AND bottle_slug = ?
ORDER BY id
""",
(component, slug),
).fetchall()
return [self._row_to_entry(row) for row in rows]
@staticmethod
def _row_to_entry(row: sqlite3.Row) -> AuditEntry:
return AuditEntry(
timestamp=row["timestamp"],
bottle_slug=row["bottle_slug"],
component=row["component"],
operator_action=row["operator_action"],
operator_notes=row["operator_notes"],
justification=row["justification"],
diff=row["diff"],
)
__all__ = ["AuditStore"]
+4 -5
View File
@@ -34,7 +34,7 @@ from ...egress import (
from ...git_gate import GIT_GATE_HOSTNAME
from ...log import die, warn
from ...supervise import (
QUEUE_DIR_IN_CONTAINER,
DB_PATH_IN_CONTAINER,
SUPERVISE_HOSTNAME,
SUPERVISE_PORT,
)
@@ -163,16 +163,15 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
if sp is not None:
env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}",
]
volumes.append({
"type": "bind",
"source": str(sp.queue_dir),
"target": QUEUE_DIR_IN_CONTAINER,
"source": str(sp.db_path),
"target": DB_PATH_IN_CONTAINER,
"read_only": False,
})
internal_aliases = [EGRESS_HOSTNAME]
if gp.upstreams:
internal_aliases.append(GIT_GATE_HOSTNAME)
+3 -3
View File
@@ -33,7 +33,7 @@ from ...git_gate import (
revoke_git_gate_provisioned_keys,
)
from ...log import die, info, warn
from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde
from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
from ..docker.git_gate import (
@@ -379,7 +379,7 @@ def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
if plan.supervise_plan is not None:
env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}",
]
return tuple(env)
@@ -405,7 +405,7 @@ def _sidecar_mounts(
sp = plan.supervise_plan
if sp is not None:
mounts.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
mounts.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
return tuple(mounts)
+3 -3
View File
@@ -27,7 +27,7 @@ from ...egress import (
egress_resolve_token_values,
egress_sidecar_env_entries,
)
from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde
from ..docker import util as docker_mod
from ..docker.egress import (
@@ -369,10 +369,10 @@ def _bundle_launch_spec(
daemons.append("supervise")
env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}",
]
volumes.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
# Container ports the agent reaches from the smolvm guest —
# published on host loopback so the guest can dial via TSI +
+2 -3
View File
@@ -284,9 +284,8 @@ def git_gate_state_dir(identity: str) -> Path:
def supervise_state_dir(identity: str) -> Path:
"""State subdir reserved for supervise sidecar bind-mount sources.
The queue dir is intentionally NOT under here — it lives at
~/.bot-bottle/queue/<slug>/ alongside the audit logs, so it
survives state-dir cleanup."""
Runtime queue/audit rows live in the host-level bot-bottle SQLite
database, so they survive state-dir cleanup."""
return bottle_state_dir(identity) / _SUPERVISE_SUBDIR
+14
View File
@@ -9,6 +9,7 @@ import sys
from ..log import Die, die, error
from ..manifest import ManifestError
from ..store_manager import StoreManager
from ._common import PROG
from . import list as _list_mod
from .cleanup import cmd_cleanup
@@ -74,6 +75,19 @@ def main(argv: list[str] | None = None) -> int:
if handler is None:
usage()
die(f"unknown command: {command}")
mgr = StoreManager.instance()
if not mgr.is_migrated():
didericis marked this conversation as resolved
Review

Should change this method actually... (and the db store methods). Probably something like is_migrated. Choose an appropriate name that reflects the fact it returns a boolean.

Should change this method actually... (and the db store methods). Probably something like `is_migrated`. Choose an appropriate name that reflects the fact it returns a boolean.
Review

Agreed. Will rename check_migrationsis_migrated on DbStore, StoreManager, and the call site here.

Agreed. Will rename `check_migrations` → `is_migrated` on `DbStore`, `StoreManager`, and the call site here.
Review

Done — 142974a. Renamed to is_migrated on DbStore, StoreManager, and the test patch.

Done — `142974a`. Renamed to `is_migrated` on `DbStore`, `StoreManager`, and the test patch.
sys.stderr.write("bot-bottle: database schema is out of date\n")
sys.stderr.write("Migrate now? [y/N] ")
sys.stderr.flush()
try:
answer = sys.stdin.readline().strip().lower()
except EOFError:
answer = ""
if answer != "y":
error("migration required — re-run and confirm to migrate")
return 1
mgr.migrate()
try:
return handler(rest) or 0
except ManifestError as e:
+9 -16
View File
@@ -45,7 +45,7 @@ from ..supervise import (
TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW,
list_pending_proposals,
list_all_pending_proposals,
render_diff,
write_audit_entry,
write_response,
@@ -63,10 +63,9 @@ _REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_AL
@dataclass(frozen=True)
class QueuedProposal:
"""A pending proposal plus the queue dir it was found in."""
"""A pending proposal from the supervise queue."""
proposal: Proposal
queue_dir: Path
# Errors any remediation engine may raise. Caught by the TUI key
@@ -86,16 +85,11 @@ def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
def discover_pending() -> list[QueuedProposal]:
"""Walk ~/.bot-bottle/queue/* and collect pending proposals."""
queue_root = _supervise.bot_bottle_root() / "queue"
if not queue_root.is_dir():
return []
out: list[QueuedProposal] = []
for slug_dir in sorted(queue_root.iterdir()):
if not slug_dir.is_dir():
continue
for proposal in list_pending_proposals(slug_dir):
out.append(QueuedProposal(proposal=proposal, queue_dir=slug_dir))
"""Collect pending proposals across bottles."""
out = [
QueuedProposal(proposal=proposal)
for proposal in list_all_pending_proposals()
]
out.sort(key=lambda q: q.proposal.arrival_timestamp)
return out
@@ -118,7 +112,6 @@ def _detail_lines(
(f"tool: {p.tool}", 0),
(f"id: {p.id}", 0),
(f"arrived: {p.arrival_timestamp}", 0),
(f"queue: {qp.queue_dir}", 0),
("", 0),
("justification:", 0),
]
@@ -165,7 +158,7 @@ def approve(
notes=notes,
final_file=final_file,
)
write_response(qp.queue_dir, response)
write_response(qp.proposal.bottle_slug, response)
_write_audit(
qp, action=status, notes=notes,
diff_before=diff_before, diff_after=diff_after,
@@ -179,7 +172,7 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
notes=reason,
final_file=None,
)
write_response(qp.queue_dir, response)
write_response(qp.proposal.bottle_slug, response)
_write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
+59
View File
@@ -0,0 +1,59 @@
"""Shared SQLite-backed store base class for bot-bottle (PRD 0013)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
try:
from .migrations import TableMigrations
except ImportError:
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
class DbVersionError(Exception):
"""Raised when the on-disk schema is behind the current migration list."""
class DbStore:
"""Base for SQLite-backed stores. Subclasses resolve db_path then call super().__init__."""
def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
self.db_path = db_path
self._migrations = migrations
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def is_migrated(self) -> bool:
didericis marked this conversation as resolved
Review

Migrating when we initialize the db store seems like an anti pattern... we should create the following two methods:

  1. check_migrations simply returns whether the migration for the store has been applied or not
  2. migrate attempt to apply the migration

The first method should be called every time bot bottle starts in any context (cli or future orchestrator). If the migrations are not up to date, we should throw an error indicating we need a migration (like DbVersionError or something). In the cli, we should catch this error and then provide an interactive prompt letting the user know they need to update the db. If the say yes, we should apply the migration.

Was considering whether or not we want to back up the db, but I don't think there's enough of high value in it to justify that.

Migrating when we initialize the db store seems like an anti pattern... we should create the following two methods: 1. `check_migrations` simply returns whether the migration for the store has been applied or not 2. `migrate` attempt to apply the migration The first method should be called every time bot bottle starts in any context (cli or future orchestrator). If the migrations are not up to date, we should throw an error indicating we need a migration (like `DbVersionError` or something). In the cli, we should catch this error and then provide an interactive prompt letting the user know they need to update the db. If the say yes, we should apply the migration. Was considering whether or not we want to back up the db, but I don't think there's enough of high value in it to justify that.
Review

Agreed. Plan:

  • Remove the _init() call from DbStore.__init__ (db_store.py:21).
  • Add check_migrations(self) -> bool to DbStore — opens a read-only connection, reads schema_versions, returns True if the stored version equals len(self._migrations.migrations) (i.e. fully up-to-date), False otherwise. Returns False if schema_versions does not exist yet.
  • Add migrate(self) -> None to DbStore — the body of the current _init() (lines 28–31): applies self._migrations and chmods.
  • Add DbVersionError(Exception) to db_store.py.
  • In cli/__init__.py main(), before dispatching to the command handler, instantiate both stores (without migrating) and call check_migrations() on each. If either returns False, raise DbVersionError. Catch it in main(), print an interactive prompt asking the user to confirm, and call migrate() on confirmation.

One open question: for commands that do not touch the DB at all (e.g. info, list), should we still run the check? Skipping it keeps startup fast and avoids a prompt for unrelated commands, but the reviewer said "every time bot bottle starts" — happy to check unconditionally if that is the intent.

Agreed. Plan: - Remove the `_init()` call from `DbStore.__init__` (db_store.py:21). - Add `check_migrations(self) -> bool` to `DbStore` — opens a read-only connection, reads `schema_versions`, returns `True` if the stored version equals `len(self._migrations.migrations)` (i.e. fully up-to-date), `False` otherwise. Returns `False` if `schema_versions` does not exist yet. - Add `migrate(self) -> None` to `DbStore` — the body of the current `_init()` (lines 28–31): applies `self._migrations` and chmods. - Add `DbVersionError(Exception)` to `db_store.py`. - In `cli/__init__.py` `main()`, before dispatching to the command handler, instantiate both stores (without migrating) and call `check_migrations()` on each. If either returns `False`, raise `DbVersionError`. Catch it in `main()`, print an interactive prompt asking the user to confirm, and call `migrate()` on confirmation. One open question: for commands that do not touch the DB at all (e.g. `info`, `list`), should we still run the check? Skipping it keeps startup fast and avoids a prompt for unrelated commands, but the reviewer said "every time bot bottle starts" — happy to check unconditionally if that is the intent.
Review

Think we should probably run it unconditionally, since we'll probably end up calling the db for those other commands in the future. I'm assuming the version check will be quick. Let me know if that's an unreasonable assumption.

Think we should probably run it unconditionally, since we'll probably end up calling the db for those other commands in the future. I'm assuming the version check will be quick. Let me know if that's an unreasonable assumption.
Review

Yep, the assumption is reasonable — it is a single SELECT on schema_versions, so negligible overhead. Will run the check unconditionally.

Yep, the assumption is reasonable — it is a single `SELECT` on `schema_versions`, so negligible overhead. Will run the check unconditionally.
"""Return True if the DB is fully up-to-date, False if migration is needed."""
if not self.db_path.exists():
return False
try:
with self._connect() as conn:
row = conn.execute(
"SELECT version FROM schema_versions WHERE module = ?",
(self._migrations.schema_key,),
).fetchone()
except sqlite3.OperationalError:
return False
version = row[0] if row else 0
return version == len(self._migrations.migrations)
def migrate(self) -> None:
"""Apply any pending migrations and set permissions on the DB file."""
with self._connect() as conn:
self._migrations.apply(conn)
self._chmod()
def _chmod(self) -> None:
try:
self.db_path.chmod(0o600)
except OSError:
pass
__all__ = ["DbStore", "DbVersionError"]
+6 -9
View File
@@ -79,14 +79,13 @@ class EgressAddon:
# only — a restart re-prompts. Mutated only from the asyncio loop that
# runs the addon hooks, so no lock is needed.
self.safe_tokens: set[str] = set()
self._supervise_queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "").strip()
self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
self._reload(initial=True)
self._install_sighup()
def _supervise_available(self) -> bool:
return bool(self._supervise_queue_dir and self._supervise_slug)
return bool(self._supervise_slug)
def _reload(self, *, initial: bool = False) -> None:
try:
@@ -393,9 +392,8 @@ class EgressAddon:
justification=_TOKEN_ALLOW_JUSTIFICATION,
current_file_hash=_sv.sha256_hex(payload),
)
queue_dir = Path(self._supervise_queue_dir)
try:
_sv.write_proposal(queue_dir, proposal)
_sv.write_proposal(proposal)
except OSError as e:
sys.stderr.write(
f"egress: could not queue token-allow proposal: {e}; "
@@ -411,8 +409,8 @@ class EgressAddon:
**self._req_ctx(flow),
}) + "\n")
response = await self._await_token_response(queue_dir, proposal.id)
_sv.archive_proposal(queue_dir, proposal.id)
response = await self._await_token_response(proposal.id)
_sv.archive_proposal(self._supervise_slug, proposal.id)
if response is not None and response.status in (
_sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
@@ -439,16 +437,15 @@ class EgressAddon:
async def _await_token_response(
self,
queue_dir: Path,
proposal_id: str,
) -> "_sv.Response | None":
"""Poll the queue dir for the operator's response without blocking the
"""Poll the DB for the operator's response without blocking the
proxy event loop. Returns the Response, or None on timeout."""
loop = asyncio.get_running_loop()
deadline = loop.time() + self._token_allow_timeout
while True:
try:
return _sv.read_response(queue_dir, proposal_id)
return _sv.read_response(self._supervise_slug, proposal_id)
except (OSError, ValueError, KeyError):
# Not written yet, or a partial/malformed write — retry until
# the deadline, then fail closed.
+38 -43
View File
@@ -234,13 +234,13 @@ import hashlib
import json
import os
import sys
import uuid
from pathlib import Path
from bot_bottle import supervise as _sv
report_path = Path(sys.argv[1])
queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "")
slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
if not queue_dir or not slug:
if not slug:
sys.exit(2)
try:
@@ -277,31 +277,19 @@ for i, finding in enumerate(raw, 1):
])
payload = "\n".join(lines).rstrip() + "\n"
proposal_id = str(uuid.uuid4())
proposal = {
"id": proposal_id,
"bottle_slug": slug,
"tool": "gitleaks-allow",
"proposed_file": payload,
"justification": (
proposal = _sv.Proposal.new(
bottle_slug=slug,
tool=_sv.TOOL_GITLEAKS_ALLOW,
proposed_file=payload,
justification=(
"git-gate found gitleaks findings hidden by # gitleaks:allow; "
"approve only for dummy test fixtures or confirmed false positives"
),
"arrival_timestamp": datetime.datetime.now(
datetime.timezone.utc
).isoformat(),
"current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
}
queue = Path(queue_dir)
queue.mkdir(parents=True, exist_ok=True)
path = queue / f"{proposal_id}.proposal.json"
tmp = path.with_suffix(path.suffix + ".tmp")
with tmp.open("w", encoding="utf-8") as f:
json.dump(proposal, f, indent=2)
f.write("\n")
os.chmod(tmp, 0o600)
os.replace(tmp, path)
print(proposal_id)
current_file_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
now=datetime.datetime.now(datetime.timezone.utc),
)
_sv.write_proposal(proposal)
print(proposal.id)
didericis marked this conversation as resolved
Review

Why do we still need a queue_dir if the queue is in the db?

Why do we still need a `queue_dir` if the queue is in the db?
PY
)
rc=$?
@@ -314,8 +302,7 @@ PY
return 1
fi
queue_dir=${SUPERVISE_QUEUE_DIR:-}
response_file="$queue_dir/${proposal_id}.response.json"
slug=${SUPERVISE_BOTTLE_SLUG:-}
timeout=${SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS:-300}
case "$timeout" in
''|*[!0-9]*)
@@ -327,26 +314,35 @@ PY
echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
waited=0
while [ "$waited" -lt "$timeout" ]; do
if [ -f "$response_file" ]; then
status=$(python3 - "$response_file" <<'PY'
import json
status=$(python3 - "$slug" "$proposal_id" <<'PY'
import sys
from bot_bottle import supervise as _sv
slug = sys.argv[1]
try:
with open(sys.argv[1], encoding="utf-8") as f:
raw = json.load(f)
except (OSError, json.JSONDecodeError):
sys.exit(1)
status = raw.get("status")
if not isinstance(status, str):
sys.exit(1)
print(status)
response = _sv.read_response(slug, sys.argv[2])
except FileNotFoundError:
sys.exit(2)
print(response.status)
PY
) || status=""
)
rc=$?
if [ "$rc" -eq 2 ]; then
status=""
elif [ "$rc" -ne 0 ]; then
status="invalid"
fi
if [ -n "$status" ]; then
case "$status" in
approved|modified)
mkdir -p "$queue_dir/processed"
mv -f "$queue_dir/${proposal_id}.proposal.json" "$queue_dir/processed/" 2>/dev/null || true
mv -f "$queue_dir/${proposal_id}.response.json" "$queue_dir/processed/" 2>/dev/null || true
python3 - "$slug" "$proposal_id" <<'PY' || true
import sys
from bot_bottle import supervise as _sv
_sv.archive_proposal(sys.argv[1], sys.argv[2])
PY
echo "git-gate: supervisor approved # gitleaks:allow for $ref" >&2
return 0
;;
@@ -499,4 +495,3 @@ if ! git -C "$repo_dir" rev-parse --verify HEAD >/dev/null 2>&1; then
fi
exit 0
"""
1
+37
View File
@@ -0,0 +1,37 @@
"""SQLite migration runner for bot-bottle stores."""
from __future__ import annotations
import sqlite3
class TableMigrations:
"""Runs a sequential list of DDL migrations tracked by schema_key in schema_versions."""
def __init__(self, schema_key: str, migrations: list[str]) -> None:
self.schema_key = schema_key
self.migrations = migrations
def apply(self, conn: sqlite3.Connection) -> None:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS schema_versions (
module TEXT PRIMARY KEY,
version INTEGER NOT NULL DEFAULT 0
)
"""
)
row = conn.execute(
"SELECT version FROM schema_versions WHERE module = ?",
(self.schema_key,),
).fetchone()
version = row[0] if row else 0
for i, sql in enumerate(self.migrations[version:], start=version + 1):
conn.execute(sql)
conn.execute(
"INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
(self.schema_key, i),
)
__all__ = ["TableMigrations"]
+215
View File
@@ -0,0 +1,215 @@
"""SQLite-backed queue store for supervise proposals and responses (PRD 0013)."""
from __future__ import annotations
import os
import sqlite3
from pathlib import Path
try:
from .supervise_types import Proposal, Response, host_db_path
from .db_store import DbStore
from .migrations import TableMigrations
except ImportError:
from supervise_types import Proposal, Response, host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from db_store import DbStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
class QueueStore(DbStore):
"""SQLite-backed persistent store for supervise proposals and responses."""
def __init__(self, queue_key: str, db_path: Path | None = None) -> None:
self.queue_key = queue_key
if db_path is not None:
resolved = db_path
else:
# In the sidecar container SUPERVISE_DB_PATH points at the
# bind-mounted host DB. On the host this env var is never set,
# so we always fall through to host_db_path().
env_path = os.environ.get("SUPERVISE_DB_PATH", "").strip()
resolved = Path(env_path) if env_path else host_db_path()
# One entry per schema version: migrations[0] brings a fresh DB to
# version 1, [1] to version 2, etc. Add new entries at the end; never
# edit existing ones.
migrations = TableMigrations("queue_store", [
# v1 — proposals table
"""
CREATE TABLE IF NOT EXISTS supervise_proposals (
queue_key TEXT NOT NULL,
id TEXT NOT NULL,
bottle_slug TEXT NOT NULL,
tool TEXT NOT NULL,
proposed_file TEXT NOT NULL,
justification TEXT NOT NULL,
arrival_timestamp TEXT NOT NULL,
current_file_hash TEXT NOT NULL,
archived INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (queue_key, id)
)
""",
# v2 — responses table
"""
CREATE TABLE IF NOT EXISTS supervise_responses (
didericis marked this conversation as resolved
Review

Same comments about these method and the lazy import from AuditStore apply here

Same comments about these method and the lazy import from AuditStore apply here
Review

All applied here too: _svget_supervise_mod(), row helpers moved to _row_to_proposal / _row_to_response static methods on QueueStore, _host_db_path() inlined.

All applied here too: `_sv` → `get_supervise_mod()`, row helpers moved to `_row_to_proposal` / `_row_to_response` static methods on `QueueStore`, `_host_db_path()` inlined.
queue_key TEXT NOT NULL,
proposal_id TEXT NOT NULL,
status TEXT NOT NULL,
notes TEXT NOT NULL,
final_file TEXT,
archived INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (queue_key, proposal_id)
)
""",
])
super().__init__(resolved, migrations)
def write_proposal(self, proposal: Proposal) -> Path:
with self._connect() as conn:
didericis marked this conversation as resolved
Review

Should probably be a DbStore base class which gets passed a db_path and a migrations object (see https://gitea.dideric.is/didericis/bot-bottle/pulls/320/files#issuecomment-2887 for shape of migrations object)

Should probably be a `DbStore` base class which gets passed a `db_path` and a `migrations` object (see https://gitea.dideric.is/didericis/bot-bottle/pulls/320/files#issuecomment-2887 for shape of migrations object)
Outdated
Review

Agreed on DbStore(db_path, migrations). Plan: subclass constructors resolve db_path (env-var check etc.) and then call super().__init__(db_path, migrations). DbStore owns _connect, _chmod, and _init (which delegates to migrations.apply(conn)). The SUPERVISE_DB_PATH env-var path-resolution stays in QueueStore.__init__ since only the queue store needs it.

Agreed on `DbStore(db_path, migrations)`. Plan: subclass constructors resolve `db_path` (env-var check etc.) and then call `super().__init__(db_path, migrations)`. `DbStore` owns `_connect`, `_chmod`, and `_init` (which delegates to `migrations.apply(conn)`). The `SUPERVISE_DB_PATH` env-var path-resolution stays in `QueueStore.__init__` since only the queue store needs it.
conn.execute(
"""
INSERT OR REPLACE INTO supervise_proposals (
queue_key, id, bottle_slug, tool, proposed_file, justification,
arrival_timestamp, current_file_hash, archived
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
""",
(
self.queue_key,
proposal.id,
proposal.bottle_slug,
proposal.tool,
proposal.proposed_file,
proposal.justification,
proposal.arrival_timestamp,
proposal.current_file_hash,
),
)
self._chmod()
return self.db_path
def read_proposal(self, proposal_id: str) -> Proposal:
with self._connect() as conn:
row = conn.execute(
"""
SELECT * FROM supervise_proposals
WHERE queue_key = ? AND id = ? AND archived = 0
""",
(self.queue_key, proposal_id),
).fetchone()
if row is None:
raise FileNotFoundError(proposal_id)
return self._row_to_proposal(row)
def list_pending_proposals(self) -> list[Proposal]:
if not self.db_path.is_file():
return []
with self._connect() as conn:
rows = conn.execute(
"""
SELECT p.* FROM supervise_proposals p
WHERE p.archived = 0
AND p.queue_key = ?
AND NOT EXISTS (
SELECT 1 FROM supervise_responses r
WHERE r.queue_key = p.queue_key
AND r.proposal_id = p.id
AND r.archived = 0
)
ORDER BY p.arrival_timestamp, p.id
""",
(self.queue_key,),
).fetchall()
return [self._row_to_proposal(row) for row in rows]
def list_all_pending_proposals(self) -> list[Proposal]:
if not self.db_path.is_file():
return []
with self._connect() as conn:
rows = conn.execute(
"""
SELECT p.* FROM supervise_proposals p
WHERE p.archived = 0
AND NOT EXISTS (
SELECT 1 FROM supervise_responses r
WHERE r.queue_key = p.queue_key
AND r.proposal_id = p.id
AND r.archived = 0
)
ORDER BY p.arrival_timestamp, p.id
"""
).fetchall()
return [self._row_to_proposal(row) for row in rows]
def write_response(self, response: Response) -> Path:
with self._connect() as conn:
conn.execute(
"""
INSERT OR REPLACE INTO supervise_responses (
queue_key, proposal_id, status, notes, final_file, archived
) VALUES (?, ?, ?, ?, ?, 0)
""",
(
self.queue_key,
response.proposal_id,
response.status,
response.notes,
response.final_file,
),
)
self._chmod()
return self.db_path
def read_response(self, proposal_id: str) -> Response:
with self._connect() as conn:
row = conn.execute(
"""
SELECT * FROM supervise_responses
WHERE queue_key = ? AND proposal_id = ? AND archived = 0
""",
(self.queue_key, proposal_id),
).fetchone()
if row is None:
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._connect() 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),
)
@staticmethod
def _row_to_proposal(row: sqlite3.Row) -> Proposal:
return Proposal(
id=row["id"],
bottle_slug=row["bottle_slug"],
tool=row["tool"],
proposed_file=row["proposed_file"],
justification=row["justification"],
arrival_timestamp=row["arrival_timestamp"],
current_file_hash=row["current_file_hash"],
)
@staticmethod
def _row_to_response(row: sqlite3.Row) -> Response:
return Response(
proposal_id=row["proposal_id"],
didericis marked this conversation as resolved
Review

Same issue about migrations also applies here

Same issue about migrations also applies here
Review

Same migration runner applied here — _MIGRATIONS list with proposals (v1) and responses (v2), tracked under queue_store key in schema_versions.

Same migration runner applied here — `_MIGRATIONS` list with proposals (v1) and responses (v2), tracked under `queue_store` key in `schema_versions`.
status=row["status"],
notes=row["notes"],
final_file=row["final_file"],
)
__all__ = ["QueueStore"]
+60
View File
@@ -0,0 +1,60 @@
"""Singleton manager for all bot-bottle SQLite stores (PRD 0013)."""
from __future__ import annotations
from pathlib import Path
try:
from .audit_store import AuditStore
from .queue_store import QueueStore
except ImportError:
from audit_store import AuditStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from queue_store import QueueStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
_instance: StoreManager | None = None
class StoreManager:
"""Owns db_path and delegates migrate/is_migrated across all stores.
Use instance() for normal access. Call reset(db_path) in tests to swap
the singleton to a temp path, then reset() with no args to restore the
default."""
def __init__(self, db_path: Path | None = None) -> None:
if db_path is None:
# Lazy import to avoid a circular dependency: supervise imports
# StoreManager at module level; StoreManager must not import
# supervise at module level in return.
try:
from .supervise import host_db_path
except ImportError:
from supervise import host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
db_path = host_db_path()
self.db_path = db_path
@classmethod
def instance(cls) -> StoreManager:
global _instance
if _instance is None:
_instance = cls()
return _instance
@classmethod
def reset(cls, db_path: Path | None = None) -> None:
"""Replace the singleton. Pass db_path for test isolation; omit to restore default."""
global _instance
_instance = cls(db_path)
def is_migrated(self) -> bool:
return (
QueueStore("", self.db_path).is_migrated()
and AuditStore(self.db_path).is_migrated()
)
def migrate(self) -> None:
QueueStore("", self.db_path).migrate()
AuditStore(self.db_path).migrate()
__all__ = ["StoreManager"]
+111 -351
View File
@@ -9,15 +9,14 @@ calls when it needs an operator-reviewed egress change:
Each tool call: the agent passes the full proposed file plus a
justification text. The sidecar validates the proposal syntactically,
writes it to the host's per-bottle queue dir, and holds the tool-call
writes it to the host SQLite queue table, and holds the tool-call
connection open. The operator's supervise TUI
(bot_bottle.cli.supervise) sees the proposal, accepts
approve / modify / reject, and writes a response file alongside the
proposal. The sidecar sees the response and returns `{status, notes}`
to the agent.
approve / modify / reject, and writes a response row. The sidecar sees
the response and returns `{status, notes}` to the agent.
This module defines the host-side library: dataclasses for the queue
file shapes, queue read/write helpers, the audit log writer, and the
record shapes, queue read/write helpers, the audit log writer, and the
diff renderer. The in-container sidecar lives in
bot_bottle/supervise_server.py; the supervise daemon's container
lifecycle is owned by the sidecar bundle (PRD 0024).
@@ -31,37 +30,56 @@ remediation engines that wire real config changes land in PRDs 0014,
from __future__ import annotations
import dataclasses
import difflib
import hashlib
import json
import os
import time
import uuid
from abc import ABC
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
try:
from .supervise_types import (
ACTION_OPERATOR_EDIT,
AuditEntry,
HOST_DB_FILENAME,
Proposal,
Response,
STATUSES,
STATUS_APPROVED,
STATUS_MODIFIED,
STATUS_REJECTED,
TOOLS,
TOOL_EGRESS_ALLOW,
TOOL_EGRESS_BLOCK,
TOOL_EGRESS_TOKEN_ALLOW,
TOOL_GITLEAKS_ALLOW,
TOOL_LIST_EGRESS_ROUTES,
bot_bottle_root,
)
except ImportError:
from supervise_types import ( # type: ignore[import-not-found,no-redef] # pylint: disable=import-error,no-name-in-module
ACTION_OPERATOR_EDIT,
AuditEntry,
HOST_DB_FILENAME,
Proposal,
Response,
STATUSES,
STATUS_APPROVED,
STATUS_MODIFIED,
STATUS_REJECTED,
TOOLS,
TOOL_EGRESS_ALLOW,
TOOL_EGRESS_BLOCK,
TOOL_EGRESS_TOKEN_ALLOW,
TOOL_GITLEAKS_ALLOW,
TOOL_LIST_EGRESS_ROUTES,
bot_bottle_root,
)
SUPERVISE_HOSTNAME = "supervise"
SUPERVISE_PORT = 9100
TOOL_EGRESS_BLOCK = "egress-block"
TOOL_EGRESS_ALLOW = "egress-allow"
TOOL_GITLEAKS_ALLOW = "gitleaks-allow"
# Written directly by the egress addon (not an agent-facing MCP tool) when an
# outbound DLP token block is routed to the operator for override (PRD 0062).
TOOL_EGRESS_TOKEN_ALLOW = "egress-token-allow"
TOOL_LIST_EGRESS_ROUTES = "list-egress-routes"
TOOLS: tuple[str, ...] = (
TOOL_EGRESS_ALLOW,
TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW,
TOOL_LIST_EGRESS_ROUTES,
)
# The supervise sidecar uses these to query egress's
# introspection endpoint for the `list-egress-routes` MCP
# tool. The hostname + port match egress's docker network
1
@@ -77,236 +95,74 @@ COMPONENT_FOR_TOOL: dict[str, str] = {
TOOL_EGRESS_BLOCK: "egress",
}
STATUS_APPROVED = "approved"
STATUS_MODIFIED = "modified"
STATUS_REJECTED = "rejected"
STATUSES: tuple[str, ...] = (STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED)
# Operator-initiated audit entries (no tool call). PRD 0014's
# `routes edit <bottle>` verb writes entries with this action.
ACTION_OPERATOR_EDIT = "operator-edit"
QUEUE_DIR_IN_CONTAINER = "/run/supervise/queue"
DB_PATH_IN_CONTAINER = "/run/supervise/bot-bottle.db"
DEFAULT_POLL_INTERVAL_SEC = 0.5
# --- Paths -----------------------------------------------------------------
def bot_bottle_root() -> Path:
return Path.home() / ".bot-bottle"
def queue_dir_for_slug(slug: str) -> Path:
return bot_bottle_root() / "queue" / slug
def audit_dir() -> Path:
return bot_bottle_root() / "audit"
def host_db_path() -> Path:
# Calls bot_bottle_root() through this module's globals so that patches
# on supervise.bot_bottle_root propagate to callers going through
# supervise.host_db_path().
didericis marked this conversation as resolved Outdated
Outdated
Review

Don't think this should be a separate db path: we want to indicate that there's a single db on the host that's used for this. I also don't think we want to make it possible to have a separate supervisor db with an env var: it'll likely be important to have supervisor proposals and running bottles accessible to the same query for creating a good dashboard UI.

Don't think this should be a separate db path: we want to indicate that there's a single db on the host that's used for this. I also don't think we want to make it possible to have a separate supervisor db with an env var: it'll likely be important to have supervisor proposals and running bottles accessible to the same query for creating a good dashboard UI.
return bot_bottle_root() / HOST_DB_FILENAME
def audit_log_path(component: str, slug: str) -> Path:
return audit_dir() / f"{component}-{slug}.log"
# --- Dataclasses -----------------------------------------------------------
@dataclass(frozen=True)
class Proposal:
"""One pending tool-call from the agent. The sidecar writes one
of these to the queue dir on a tool call; the operator's TUI
reads them; the sidecar polls for a matching Response."""
id: str
bottle_slug: str
tool: str
proposed_file: str
justification: str
arrival_timestamp: str
current_file_hash: str
@classmethod
def new(
cls,
*,
bottle_slug: str,
tool: str,
proposed_file: str,
justification: str,
current_file_hash: str,
now: datetime | None = None,
) -> "Proposal":
ts = (now or datetime.now(timezone.utc)).isoformat()
return cls(
id=str(uuid.uuid4()),
bottle_slug=bottle_slug,
tool=tool,
proposed_file=proposed_file,
justification=justification,
arrival_timestamp=ts,
current_file_hash=current_file_hash,
)
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
@classmethod
def from_dict(cls, raw: dict[str, object]) -> "Proposal":
tool = _require_str(raw, "tool")
if tool not in TOOLS:
raise ValueError(f"tool must be one of {TOOLS}; got {tool!r}")
return cls(
id=_require_str(raw, "id"),
bottle_slug=_require_str(raw, "bottle_slug"),
tool=tool,
proposed_file=_require_str(raw, "proposed_file"),
justification=_require_str(raw, "justification"),
arrival_timestamp=_require_str(raw, "arrival_timestamp"),
current_file_hash=_require_str(raw, "current_file_hash"),
)
@dataclass(frozen=True)
class Response:
"""The operator's decision on a proposal. The TUI writes one of
these to the queue dir; the sidecar reads it and returns the
`{status, notes}` pair to the agent's tool call.
`final_file` carries the file content the supervisor will
actually apply: for `approved`, equal to the proposal's
`proposed_file`; for `modified`, the operator's edited version
(the audit diff is current → final_file, not current →
proposed_file); for `rejected`, None."""
proposal_id: str
status: str
notes: str
final_file: str | None = None
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
@classmethod
def from_dict(cls, raw: dict[str, object]) -> "Response":
status = _require_str(raw, "status")
if status not in STATUSES:
raise ValueError(
f"response status must be one of {STATUSES}; got {status!r}"
)
final = raw.get("final_file")
if final is not None and not isinstance(final, str):
raise ValueError(
f"final_file must be a string or null; got {type(final).__name__}"
)
return cls(
proposal_id=_require_str(raw, "proposal_id"),
status=status,
notes=_require_str(raw, "notes"),
final_file=final,
)
@dataclass(frozen=True)
class AuditEntry:
"""One row of the per-bottle audit log. JSON-Lines, append-only."""
timestamp: str
bottle_slug: str
component: str
operator_action: str
operator_notes: str
justification: str
diff: str
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
try:
from .queue_store import QueueStore
from .audit_store import AuditStore
from .store_manager import StoreManager
except ImportError:
# Sidecar bundle: files are flat-copied under /app, not a package.
from queue_store import QueueStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from audit_store import AuditStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from store_manager import StoreManager # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
# --- Queue I/O -------------------------------------------------------------
def _proposal_filename(proposal_id: str) -> str:
return f"{proposal_id}.proposal.json"
def _response_filename(proposal_id: str) -> str:
return f"{proposal_id}.response.json"
def _id_from_proposal_filename(path: Path) -> str | None:
name = path.name
if not name.endswith(".proposal.json"):
return None
return name[: -len(".proposal.json")]
def write_proposal(queue_dir: Path, proposal: Proposal) -> Path:
"""Persist `proposal` as JSON in the queue dir, mode 0o600.
def write_proposal(proposal: Proposal) -> Path:
"""Persist `proposal` in the queue database, mode 0o600.
Directory is created if missing."""
queue_dir.mkdir(parents=True, exist_ok=True)
path = queue_dir / _proposal_filename(proposal.id)
payload = json.dumps(proposal.to_dict(), indent=2) + "\n"
_atomic_write(path, payload, mode=0o600)
return path
return QueueStore(proposal.bottle_slug).write_proposal(proposal)
def read_proposal(queue_dir: Path, proposal_id: str) -> Proposal:
path = queue_dir / _proposal_filename(proposal_id)
with path.open() as f:
raw = json.load(f)
if not isinstance(raw, dict):
raise ValueError(f"{path}: top-level must be an object")
return Proposal.from_dict(raw)
def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal:
return QueueStore(bottle_slug).read_proposal(proposal_id)
def list_pending_proposals(queue_dir: Path) -> list[Proposal]:
"""All proposals in `queue_dir` that do not yet have a matching
response file. Sorted by `arrival_timestamp` so the operator
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."""
if not queue_dir.is_dir():
return []
out: list[Proposal] = []
for path in sorted(queue_dir.glob("*.proposal.json")):
proposal_id = _id_from_proposal_filename(path)
if proposal_id is None:
continue
if (queue_dir / _response_filename(proposal_id)).exists():
continue
try:
with path.open() as f:
raw = json.load(f)
except (OSError, json.JSONDecodeError):
continue
if not isinstance(raw, dict):
continue
try:
out.append(Proposal.from_dict(raw))
except (KeyError, ValueError):
continue
out.sort(key=lambda p: p.arrival_timestamp)
return out
return QueueStore(bottle_slug).list_pending_proposals()
def write_response(queue_dir: Path, response: Response) -> Path:
queue_dir.mkdir(parents=True, exist_ok=True)
path = queue_dir / _response_filename(response.proposal_id)
payload = json.dumps(response.to_dict(), indent=2) + "\n"
_atomic_write(path, payload, mode=0o600)
return path
def list_all_pending_proposals() -> list[Proposal]:
"""All pending proposals across bottles, sorted FIFO."""
return QueueStore("").list_all_pending_proposals()
def read_response(queue_dir: Path, proposal_id: str) -> Response:
path = queue_dir / _response_filename(proposal_id)
with path.open() as f:
raw = json.load(f)
if not isinstance(raw, dict):
raise ValueError(f"{path}: top-level must be an object")
return Response.from_dict(raw)
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 wait_for_response(
queue_dir: Path,
bottle_slug: str,
proposal_id: str,
*,
poll_interval: float = DEFAULT_POLL_INTERVAL_SEC,
@@ -317,90 +173,35 @@ def wait_for_response(
which the wait raises TimeoutError. None waits forever — the
natural shape, since the operator's response time is unbounded.
Polls the filesystem so the implementation stays portable and
stdlib-only."""
path = queue_dir / _response_filename(proposal_id)
Polls SQLite so the implementation stays portable and stdlib-only."""
store = QueueStore(bottle_slug)
while True:
if path.exists():
try:
with path.open() as f:
raw = json.load(f)
except (OSError, json.JSONDecodeError):
raw = None
if isinstance(raw, dict):
try:
return Response.from_dict(raw)
except (KeyError, ValueError):
pass
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(queue_dir: Path, proposal_id: str) -> None:
"""Move both proposal and response files to `<queue_dir>/processed/`.
Idempotent — missing files are silently skipped."""
processed = queue_dir / "processed"
processed.mkdir(parents=True, exist_ok=True)
for name in (_proposal_filename(proposal_id), _response_filename(proposal_id)):
src = queue_dir / name
if src.exists():
src.rename(processed / name)
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)
# --- Audit log -------------------------------------------------------------
def write_audit_entry(entry: AuditEntry) -> Path:
"""Append `entry` as one JSON-Lines record to the per-bottle
audit log. Acquires an advisory exclusive lock so concurrent
writers don't interleave bytes."""
path = audit_log_path(entry.component, entry.bottle_slug)
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(entry.to_dict(), sort_keys=False) + "\n"
fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
try:
_try_flock(fd)
try:
os.write(fd, line.encode("utf-8"))
finally:
_try_funlock(fd)
finally:
os.close(fd)
return 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. Empty
list if the log doesn't exist."""
path = audit_log_path(component, slug)
if not path.is_file():
return []
out: list[AuditEntry] = []
with path.open() as f:
for raw_line in f:
raw_line = raw_line.strip()
if not raw_line:
continue
try:
raw = json.loads(raw_line)
except json.JSONDecodeError:
continue
if not isinstance(raw, dict):
continue
try:
out.append(AuditEntry(
timestamp=_require_str(raw, "timestamp"),
bottle_slug=_require_str(raw, "bottle_slug"),
component=_require_str(raw, "component"),
operator_action=_require_str(raw, "operator_action"),
operator_notes=_require_str(raw, "operator_notes"),
justification=_require_str(raw, "justification"),
diff=_require_str(raw, "diff"),
))
except ValueError:
continue
return out
"""Load all audit entries for the given component+slug."""
return AuditStore().read_audit_entries(component, slug)
# --- Diff rendering --------------------------------------------------------
@@ -433,89 +234,47 @@ def sha256_hex(content: str) -> str:
class SupervisePlan:
"""Output of Supervise.prepare; consumed by .start.
`queue_dir` is the host directory bind-mounted into the sidecar
at /run/supervise/queue. `internal_network` is empty at prepare
time; the backend's launch step fills it via dataclasses.replace
before calling .start."""
`db_path` is the host database bind-mounted into the sidecar at
/run/supervise/bot-bottle.db. `internal_network` is empty at
prepare time; the backend's launch step fills it via
dataclasses.replace before calling .start."""
slug: str
queue_dir: Path
db_path: Path
internal_network: str = ""
class Supervise(ABC):
"""Per-bottle supervise sidecar. Encapsulates the host-side
prepare (queue dir staging); the sidecar's start/stop lifecycle
is backend-specific."""
"""Per-bottle supervise sidecar. Encapsulates host-side database
staging; the sidecar's start/stop lifecycle is backend-specific."""
def prepare(
self,
slug: str,
stage_dir: Path,
) -> SupervisePlan:
"""Stage the per-bottle queue dir on the host. Returns the
plan; `internal_network` must be set by the launch step before
.start runs."""
"""Stage the host database. Returns the plan; `internal_network`
must be set by the launch step before .start runs."""
del stage_dir
queue_dir = queue_dir_for_slug(slug)
queue_dir.mkdir(parents=True, exist_ok=True)
mgr = StoreManager.instance()
mgr.migrate()
return SupervisePlan(
slug=slug,
queue_dir=queue_dir,
db_path=mgr.db_path,
)
# --- Helpers ---------------------------------------------------------------
def _require_str(raw: dict[str, object], key: str) -> str:
value = raw.get(key)
if not isinstance(value, str):
raise ValueError(f"missing or non-string field {key!r}")
return value
def _atomic_write(path: Path, content: str, *, mode: int) -> None:
"""Atomic: write to a sibling tmp file, fsync, rename."""
tmp = path.with_suffix(path.suffix + ".tmp")
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, mode)
try:
os.write(fd, content.encode("utf-8"))
os.fsync(fd)
finally:
os.close(fd)
os.replace(tmp, path)
try:
import fcntl as _fcntl
def _try_flock(fd: int) -> None: # type: ignore[reportRedeclaration]
try:
_fcntl.flock(fd, _fcntl.LOCK_EX)
except OSError:
pass
def _try_funlock(fd: int) -> None: # type: ignore[reportRedeclaration]
try:
_fcntl.flock(fd, _fcntl.LOCK_UN)
except OSError:
pass
except ImportError: # pragma: no cover — Windows path
def _try_flock(fd: int) -> None: # noqa: F841 — Windows fallback
return None
def _try_funlock(fd: int) -> None: # noqa: F841 — Windows fallback
return None
__all__ = [
"ACTION_OPERATOR_EDIT",
"AuditEntry",
"AuditStore",
"COMPONENT_FOR_TOOL",
"DEFAULT_POLL_INTERVAL_SEC",
"DB_PATH_IN_CONTAINER",
"Proposal",
"QUEUE_DIR_IN_CONTAINER",
"QueueStore",
"Response",
"StoreManager",
"STATUSES",
"STATUS_APPROVED",
"STATUS_MODIFIED",
@@ -536,8 +295,9 @@ __all__ = [
"audit_dir",
"audit_log_path",
"bot_bottle_root",
"host_db_path",
"list_pending_proposals",
"queue_dir_for_slug",
"list_all_pending_proposals",
"read_audit_entries",
"read_proposal",
"read_response",
+9 -17
View File
@@ -7,14 +7,13 @@ config changes when stuck. The tools are `egress-allow`,
Each queued tool call:
1. Validates the proposed file syntactically.
2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
the host's ~/.bot-bottle/queue/<slug>/).
3. Blocks polling for a matching Response file.
2. Writes a Proposal to the host SQLite database.
3. Blocks polling for a matching Response row.
4. Returns the operator's `{status, notes}` to the agent.
The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
container creation by the backend's start step). The queue dir comes
from SUPERVISE_QUEUE_DIR (default `/run/supervise/queue`).
container creation by the backend's start step). SUPERVISE_DB_PATH
points at the bind-mounted host database.
Speaks MCP over HTTP+JSON-RPC. Methods handled:
@@ -42,7 +41,6 @@ import typing
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
try:
# Same-directory imports inside the bundle container; these files are
@@ -277,7 +275,6 @@ def validate_proposed_file(tool: str, content: str) -> None:
@dataclass(frozen=True)
class ServerConfig:
bottle_slug: str
queue_dir: Path
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS
@@ -376,7 +373,7 @@ def handle_tools_call(
current_file_hash=_sv.sha256_hex(proposed_file),
)
try:
_sv.write_proposal(config.queue_dir, proposal)
_sv.write_proposal(proposal)
except OSError as e:
raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e
sys.stderr.write(
@@ -387,7 +384,7 @@ def handle_tools_call(
deadline = time.monotonic() + config.response_timeout_seconds
try:
response = _sv.wait_for_response(
config.queue_dir,
config.bottle_slug,
proposal.id,
poll_interval=MIN_RESPONSE_POLL_INTERVAL_SECONDS,
deadline=deadline,
@@ -399,7 +396,7 @@ def handle_tools_call(
"isError": False,
}
try:
_sv.archive_proposal(config.queue_dir, proposal.id)
_sv.archive_proposal(config.bottle_slug, proposal.id)
except OSError as e:
raise _RpcInternalError(f"failed to archive proposal: {e}") from e
@@ -539,7 +536,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = True
daemon_threads = True
config: ServerConfig = ServerConfig(bottle_slug="", queue_dir=Path())
config: ServerConfig = ServerConfig(bottle_slug="")
# --- Entry point -----------------------------------------------------------
@@ -548,21 +545,18 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
def serve(
*,
bottle_slug: str,
queue_dir: Path,
port: int = _sv.SUPERVISE_PORT,
bind: str = "0.0.0.0",
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
) -> typing.NoReturn:
queue_dir.mkdir(parents=True, exist_ok=True)
server = MCPServer((bind, port), MCPHandler)
server.config = ServerConfig(
bottle_slug=bottle_slug,
queue_dir=queue_dir,
response_timeout_seconds=response_timeout_seconds,
)
sys.stderr.write(
f"supervise listening on {bind}:{port}; "
f"slug={bottle_slug!r}; queue={queue_dir}; "
f"slug={bottle_slug!r}; "
f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type]
)
sys.stderr.flush()
@@ -581,7 +575,6 @@ def main(argv: list[str]) -> int:
if not bottle_slug:
sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
return 2
queue_dir = Path(os.environ.get("SUPERVISE_QUEUE_DIR", _sv.QUEUE_DIR_IN_CONTAINER))
port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
bind = os.environ.get("SUPERVISE_BIND", "0.0.0.0")
try:
@@ -591,7 +584,6 @@ def main(argv: list[str]) -> int:
return 2
serve(
bottle_slug=bottle_slug,
queue_dir=queue_dir,
port=port,
bind=bind,
response_timeout_seconds=response_timeout_seconds,
+182
View File
@@ -0,0 +1,182 @@
"""Shared types and path helpers for supervise stores (PRD 0013).
Extracted from supervise.py so queue_store and audit_store can import
Proposal, Response, AuditEntry, and host_db_path without creating a
circular import (supervise imports from queue_store/audit_store and
vice-versa).
Patching bot_bottle_root on this module propagates through host_db_path
because host_db_path looks up bot_bottle_root via sys.modules[__name__]
at call time rather than capturing it at import time.
"""
from __future__ import annotations
import dataclasses
import sys
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
HOST_DB_FILENAME = "bot-bottle.db"
TOOL_EGRESS_BLOCK = "egress-block"
TOOL_EGRESS_ALLOW = "egress-allow"
TOOL_GITLEAKS_ALLOW = "gitleaks-allow"
TOOL_EGRESS_TOKEN_ALLOW = "egress-token-allow"
TOOL_LIST_EGRESS_ROUTES = "list-egress-routes"
TOOLS: tuple[str, ...] = (
TOOL_EGRESS_ALLOW,
TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW,
TOOL_LIST_EGRESS_ROUTES,
)
STATUS_APPROVED = "approved"
STATUS_MODIFIED = "modified"
STATUS_REJECTED = "rejected"
STATUSES: tuple[str, ...] = (STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED)
ACTION_OPERATOR_EDIT = "operator-edit"
def bot_bottle_root() -> Path:
return Path.home() / ".bot-bottle"
def host_db_path() -> Path:
# Look up bot_bottle_root through this module's live namespace so that
# monkey-patches on supervise_types.bot_bottle_root take effect.
return sys.modules[__name__].bot_bottle_root() / HOST_DB_FILENAME
def _require_str(raw: dict[str, object], key: str) -> str:
value = raw.get(key)
if not isinstance(value, str):
raise ValueError(f"missing or non-string field {key!r}")
return value
@dataclass(frozen=True)
class Proposal:
"""One pending tool-call from the agent."""
id: str
bottle_slug: str
tool: str
proposed_file: str
justification: str
arrival_timestamp: str
current_file_hash: str
@classmethod
def new(
cls,
*,
bottle_slug: str,
tool: str,
proposed_file: str,
justification: str,
current_file_hash: str,
now: datetime | None = None,
) -> "Proposal":
ts = (now or datetime.now(timezone.utc)).isoformat()
return cls(
id=str(uuid.uuid4()),
bottle_slug=bottle_slug,
tool=tool,
proposed_file=proposed_file,
justification=justification,
arrival_timestamp=ts,
current_file_hash=current_file_hash,
)
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
@classmethod
def from_dict(cls, raw: dict[str, object]) -> "Proposal":
tool = _require_str(raw, "tool")
if tool not in TOOLS:
raise ValueError(f"tool must be one of {TOOLS}; got {tool!r}")
return cls(
id=_require_str(raw, "id"),
bottle_slug=_require_str(raw, "bottle_slug"),
tool=tool,
proposed_file=_require_str(raw, "proposed_file"),
justification=_require_str(raw, "justification"),
arrival_timestamp=_require_str(raw, "arrival_timestamp"),
current_file_hash=_require_str(raw, "current_file_hash"),
)
@dataclass(frozen=True)
class Response:
"""The operator's decision on a proposal."""
proposal_id: str
status: str
notes: str
final_file: str | None = None
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
@classmethod
def from_dict(cls, raw: dict[str, object]) -> "Response":
status = _require_str(raw, "status")
if status not in STATUSES:
raise ValueError(
f"response status must be one of {STATUSES}; got {status!r}"
)
final = raw.get("final_file")
if final is not None and not isinstance(final, str):
raise ValueError(
f"final_file must be a string or null; got {type(final).__name__}"
)
return cls(
proposal_id=_require_str(raw, "proposal_id"),
status=status,
notes=_require_str(raw, "notes"),
final_file=final,
)
@dataclass(frozen=True)
class AuditEntry:
"""One row of the per-bottle audit log."""
timestamp: str
bottle_slug: str
component: str
operator_action: str
operator_notes: str
justification: str
diff: str
def to_dict(self) -> dict[str, object]:
return dataclasses.asdict(self)
__all__ = [
"ACTION_OPERATOR_EDIT",
"AuditEntry",
"HOST_DB_FILENAME",
"Proposal",
"Response",
"STATUSES",
"STATUS_APPROVED",
"STATUS_MODIFIED",
"STATUS_REJECTED",
"TOOLS",
"TOOL_EGRESS_ALLOW",
"TOOL_EGRESS_BLOCK",
"TOOL_EGRESS_TOKEN_ALLOW",
"TOOL_GITLEAKS_ALLOW",
"TOOL_LIST_EGRESS_ROUTES",
"bot_bottle_root",
"host_db_path",
]
+135
View File
@@ -0,0 +1,135 @@
# PRD prd-new: SQLite local storage
- **Status:** Active
- **Author:** codex
- **Created:** 2026-07-01
- **Issue:** #319
## Summary
Add a small stdlib SQLite storage layer for bot-bottle host runtime state,
starting with the supervise queue and audit log. This replaces scattered JSON
queue files and JSONL audit logs with structured tables while preserving the
existing public supervise helper functions and sidecar queue mount contract.
## Problem
Bot-bottle currently stores supervise proposals and responses as individual JSON
files under `~/.bot-bottle/queue/<slug>/`, and audit entries as JSONL files
under `~/.bot-bottle/audit/`. That worked for the original interactive TUI, but
new forge-native orchestration needs durable, queryable local state for queues,
audit trails, watchdogs, and lifecycle records. PR #318 started introducing
SQLite-shaped boilerplate for forge state; the storage foundation should live in
its own PR so forge work can build on the shared runtime store instead of adding
one-off persistence.
## Goals / Success Criteria
1. Supervise proposals and responses are persisted through SQLite.
2. Audit entries are persisted through SQLite.
3. Supervise queue helpers use the bottle slug / queue key instead of a queue
directory path.
4. The sidecar receives the host database mount across docker, smolmachines,
and macOS-container backends.
5. The implementation stays stdlib-only.
6. Schema migrations use a `PRAGMA user_version` runner — no third-party deps.
7. Unit tests cover queue round-trips, pending discovery, response waits,
archive semantics, audit round-trips, and path creation.
## Non-goals
- Migrating old JSON queue files or JSONL audit logs.
- Adding forge orchestration state tables.
- Adding egress metering or budget tables.
- Changing the supervise TUI workflow or remediation behavior.
didericis marked this conversation as resolved
Review

Ideally not third party, but do want a migration framework.

Ideally not third party, but do want a migration framework.
Review

Updated: non-goal now says "third-party ORM or migration library" (not framework); added goal 6 explicitly calling out the stdlib schema_versions migration runner.

Updated: non-goal now says "third-party ORM or migration library" (not framework); added goal 6 explicitly calling out the stdlib `schema_versions` migration runner.
- Introducing a third-party ORM or migration library.
## Design
### Database locations
Queue and audit state use the host-level local database:
```text
~/.bot-bottle/bot-bottle.db
```
The supervise sidecar receives that database as a writable bind mount at
`/run/supervise/bot-bottle.db` and gets the path through `SUPERVISE_DB_PATH`.
No per-slug queue directory is mounted into the sidecar. This creates the shared
host database that later forge/native lifecycle work can extend in separate
PRDs.
### Tables
`supervise_proposals` lives in the host database:
```sql
CREATE TABLE supervise_proposals (
queue_key TEXT NOT NULL,
id TEXT NOT NULL,
bottle_slug TEXT NOT NULL,
tool TEXT NOT NULL,
proposed_file TEXT NOT NULL,
justification TEXT NOT NULL,
arrival_timestamp TEXT NOT NULL,
current_file_hash TEXT NOT NULL,
archived INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (queue_key, id)
);
```
`supervise_responses` lives in the host database:
```sql
CREATE TABLE supervise_responses (
queue_key TEXT NOT NULL,
proposal_id TEXT NOT NULL,
status TEXT NOT NULL,
notes TEXT NOT NULL,
final_file TEXT,
archived INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (queue_key, proposal_id)
);
```
`supervise_audit_entries` lives in the host database:
```sql
CREATE TABLE supervise_audit_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
bottle_slug TEXT NOT NULL,
component TEXT NOT NULL,
operator_action TEXT NOT NULL,
operator_notes TEXT NOT NULL,
justification TEXT NOT NULL,
diff TEXT NOT NULL
);
```
### Compatibility
The queue helpers take a bottle slug / queue key and perform equivalent
operations against `~/.bot-bottle/bot-bottle.db`:
- `list_pending_proposals` returns non-archived proposals without a non-archived
response, sorted by arrival time.
- `archive_proposal` marks matching proposal/response rows archived instead of
moving files into `processed/`.
- `wait_for_response` keeps the current polling behavior but polls SQLite.
The old audit path helpers (`audit_dir`, `audit_log_path`) stay available for
compatibility. `audit_log_path` no longer describes the active storage location;
callers should use `read_audit_entries`.
## Implementation chunks
1. Add SQLite store helpers for supervise queue and audit state.
2. Rewire `bot_bottle.supervise` queue/audit functions to the store.
3. Update supervise CLI discovery tests and queue/audit unit tests.
4. Run unit tests, pyright, and pylint for touched modules.
## Open questions
None.
+8
View File
@@ -12,11 +12,19 @@ from unittest.mock import patch
import bot_bottle.cli as climod
from bot_bottle.cli import main
from bot_bottle.db_store import DbStore
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestError
from bot_bottle.store_manager import StoreManager
class TestMainDispatch(unittest.TestCase):
def setUp(self) -> None:
patcher = patch.object(DbStore, "is_migrated", return_value=True)
self._mock_check = patcher.start()
self.addCleanup(patcher.stop)
self.addCleanup(StoreManager.reset)
def test_no_args_prints_usage_returns_2(self) -> None:
with patch("sys.stderr", io.StringIO()):
self.assertEqual(2, main([]))
+3 -4
View File
@@ -107,7 +107,7 @@ def _egress_plan(
def _supervise_plan() -> SupervisePlan:
return SupervisePlan(
slug=SLUG,
queue_dir=STATE / "supervise" / "queue",
db_path=STATE / "bot-bottle.db",
internal_network=f"bot-bottle-net-{SLUG}",
)
@@ -392,7 +392,7 @@ class TestSidecarBundleShape(unittest.TestCase):
sc = self._render(supervise=True)["services"]["sidecars"]
env_strings = sc["environment"]
self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings)
self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
self.assertIn("SUPERVISE_DB_PATH=/run/supervise/bot-bottle.db", env_strings)
self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))
def test_volumes_always_includes_egress_ca(self):
@@ -408,8 +408,7 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertIn("/etc/egress", targets)
self.assertIn("/git-gate-entrypoint.sh", targets)
self.assertIn("/git-gate/creds/upstream-known_hosts", targets)
self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
for t in targets))
self.assertIn("/run/supervise/bot-bottle.db", targets)
def test_extra_hosts_omitted_for_git_upstreams(self):
sc = self._render(with_git=True)["services"]["sidecars"]
+1 -1
View File
@@ -74,7 +74,7 @@ def _plan(
if supervise:
supervise_plan = SupervisePlan(
slug="demo-abc12",
queue_dir=Path("/tmp/queue"),
db_path=Path("/tmp/bot-bottle.db"),
)
return DockerBottlePlan(
spec=spec,
+1 -1
View File
@@ -77,7 +77,7 @@ def _plan(
if supervise:
supervise_plan = SupervisePlan(
slug="demo-abc12",
queue_dir=Path("/tmp/queue"),
db_path=Path("/tmp/bot-bottle.db"),
)
return DockerBottlePlan(
spec=spec,
@@ -47,7 +47,6 @@ def _addon() -> EgressAddon:
a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = Config(routes=(), log=LOG_FULL)
a.safe_tokens = set()
a._supervise_queue_dir = ""
a._supervise_slug = ""
a._token_allow_timeout = 300.0
return a
+3 -6
View File
@@ -212,7 +212,6 @@ def _addon(config: Config) -> EgressAddon:
a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = config
a.safe_tokens = set()
a._supervise_queue_dir = ""
a._supervise_slug = ""
a._token_allow_timeout = 300.0
a.routes_path = "/nonexistent/routes.yaml"
@@ -386,10 +385,10 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
def _sha256_hex(_payload: Any) -> str:
return "hash"
def _noop(_a: Any, _b: Any) -> None:
def _noop(*_args: Any) -> None:
return None
def _read_response(_qd: Any, _pid: Any) -> Any:
def _read_response(_slug: Any, _pid: Any) -> Any:
if response_status is None:
raise OSError("not written yet") # forces poll -> timeout
return types.SimpleNamespace(status=response_status)
@@ -409,7 +408,6 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
class TestSuperviseBranch(unittest.TestCase):
def _supervised_addon(self) -> EgressAddon:
addon = _addon(Config(routes=(Route(host="api.example.com"),)))
addon._supervise_queue_dir = "/tmp/egress-queue"
addon._supervise_slug = "test-bottle"
addon._token_allow_timeout = 0.05
return addon
@@ -632,14 +630,13 @@ class TestRedactSurfaces(unittest.TestCase):
class TestSuperviseWriteFailure(unittest.TestCase):
def test_write_proposal_oserror_blocks(self) -> None:
addon = _addon(Config(routes=(Route(host="api.example.com"),)))
addon._supervise_queue_dir = "/tmp/egress-queue"
addon._supervise_slug = "test-bottle"
addon._token_allow_timeout = 0.05
flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}"))
fake = _fake_sv("approved")
def _raise(_qd: Any, _p: Any) -> None:
def _raise(_p: Any) -> None:
raise OSError("disk full")
fake.write_proposal = _raise
+3 -2
View File
@@ -210,8 +210,9 @@ class TestHookRender(unittest.TestCase):
# the suppressed findings for human approval.
self.assertIn("--ignore-gitleaks-allow", hook)
self.assertIn("--report-format=json", hook)
self.assertIn('"tool": "gitleaks-allow"', hook)
self.assertIn("SUPERVISE_QUEUE_DIR", hook)
self.assertIn("tool=_sv.TOOL_GITLEAKS_ALLOW", hook)
self.assertIn("_sv.write_proposal", hook)
self.assertIn("_sv.read_response", hook)
self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
self.assertIn("supervisor approved # gitleaks:allow", hook)
self.assertIn("supervisor rejected # gitleaks:allow", hook)
+4 -2
View File
@@ -71,7 +71,9 @@ def _plan(
else:
git_gate_plan = SimpleNamespace(upstreams=())
supervise_plan = (
SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
SimpleNamespace(
db_path=Path("/state/bot-bottle.db"),
)
if supervise else None
)
agent_provision = SimpleNamespace(
@@ -137,7 +139,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
argv,
)
self.assertIn(
"type=bind,source=/state/supervise/queue,target=/run/supervise/queue",
"type=bind,source=/state/bot-bottle.db,target=/run/supervise/bot-bottle.db",
argv,
)
+9 -1
View File
@@ -130,7 +130,7 @@ def _plan(
if supervise:
supervise_plan = SupervisePlan(
slug="demo-abc12",
queue_dir=Path("/tmp/queue"),
db_path=Path("/tmp/bot-bottle.db"),
)
return SmolmachinesBottlePlan(
spec=spec,
@@ -422,6 +422,14 @@ class TestBundleLaunchSpec(unittest.TestCase):
spec.environment,
)
def test_supervise_adds_daemon_volume_and_env(self):
from bot_bottle.supervise import DB_PATH_IN_CONTAINER
plan = _plan(supervise=True)
spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
self.assertIn("supervise", spec.daemons_csv)
self.assertIn(f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", spec.environment)
self.assertIn(("/tmp/bot-bottle.db", DB_PATH_IN_CONTAINER, False), spec.volumes)
def test_canary_env_visible_to_smolvm_guest(self):
plan = _plan(canary=True)
with patch.object(
+78 -40
View File
@@ -1,6 +1,5 @@
"""Unit: supervise queue + audit log + diff helpers (PRD 0013)."""
import json
import tempfile
import threading
import time
@@ -9,6 +8,9 @@ from datetime import datetime, timezone
from pathlib import Path
from bot_bottle import supervise
from bot_bottle import supervise_types as sv_types
from bot_bottle.audit_store import AuditStore
from bot_bottle.queue_store import QueueStore
from bot_bottle.supervise import (
AuditEntry,
Proposal,
@@ -19,7 +21,7 @@ from bot_bottle.supervise import (
TOOL_EGRESS_ALLOW,
TOOL_GITLEAKS_ALLOW,
archive_proposal,
audit_log_path,
host_db_path,
list_pending_proposals,
read_audit_entries,
read_proposal,
@@ -112,32 +114,52 @@ class TestResponseRoundtrip(unittest.TestCase):
class TestQueueIO(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-test.")
self.queue_dir = Path(self._tmp.name)
self._home_patch = self._patch_home(Path(self._tmp.name))
self.slug = "dev"
QueueStore("").migrate()
def tearDown(self):
self._home_patch()
self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original_sv = supervise.bot_bottle_root
original_svt = sv_types.bot_bottle_root
def fake_root() -> Path:
return fake_home / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
sv_types.bot_bottle_root = fake_root # type: ignore[assignment]
def restore() -> None:
supervise.bot_bottle_root = original_sv # type: ignore[assignment]
sv_types.bot_bottle_root = original_svt # type: ignore[assignment]
return restore
def test_write_and_read_proposal(self):
p = _proposal()
path = write_proposal(self.queue_dir, p)
path = 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.queue_dir, p.id)
loaded = 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(self.queue_dir, a)
write_proposal(self.queue_dir, b)
write_response(self.queue_dir, Response(
write_proposal(a)
write_proposal(b)
write_response(self.slug, Response(
proposal_id=a.id, status=STATUS_APPROVED, notes="",
))
pending = list_pending_proposals(self.queue_dir)
pending = list_pending_proposals(self.slug)
self.assertEqual([b.id], [p.id for p in pending])
def test_list_pending_returns_empty_for_missing_dir(self):
self.assertEqual([], list_pending_proposals(self.queue_dir / "nope"))
def test_list_pending_returns_empty_for_missing_slug(self):
self.assertEqual([], list_pending_proposals("nope"))
def test_list_pending_sorted_by_arrival(self):
# Fabricate two with explicit timestamps.
@@ -154,30 +176,30 @@ class TestQueueIO(unittest.TestCase):
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
)
# Write in reverse order.
write_proposal(self.queue_dir, b)
write_proposal(self.queue_dir, a)
ordered = list_pending_proposals(self.queue_dir)
write_proposal(b)
write_proposal(a)
ordered = 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.queue_dir, r)
self.assertEqual(r, read_response(self.queue_dir, "xyz"))
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(self.queue_dir, p)
write_proposal(p)
def write_after_delay():
time.sleep(0.05)
write_response(self.queue_dir, Response(
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.queue_dir, p.id, poll_interval=0.01)
r = wait_for_response(self.slug, p.id, poll_interval=0.01)
finally:
t.join()
self.assertEqual(STATUS_APPROVED, r.status)
@@ -187,44 +209,51 @@ class TestQueueIO(unittest.TestCase):
deadline = time.monotonic() + 0.05
with self.assertRaises(TimeoutError):
wait_for_response(
self.queue_dir, "never",
self.slug, "never",
poll_interval=0.01, deadline=deadline,
)
def test_archive_proposal_moves_both_files(self):
def test_archive_proposal_hides_rows(self):
p = _proposal()
write_proposal(self.queue_dir, p)
write_response(self.queue_dir, Response(
write_proposal(p)
write_response(self.slug, Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="",
))
archive_proposal(self.queue_dir, p.id)
self.assertFalse((self.queue_dir / f"{p.id}.proposal.json").exists())
self.assertFalse((self.queue_dir / f"{p.id}.response.json").exists())
self.assertTrue((self.queue_dir / "processed" / f"{p.id}.proposal.json").exists())
self.assertTrue((self.queue_dir / "processed" / f"{p.id}.response.json").exists())
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.queue_dir, "nope")
archive_proposal(self.slug, "nope")
class TestAuditLog(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-audit.")
self._home_patch = self._patch_home(Path(self._tmp.name))
AuditStore().migrate()
def tearDown(self):
self._home_patch()
self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original = supervise.bot_bottle_root
original_sv = supervise.bot_bottle_root
original_svt = sv_types.bot_bottle_root
def fake_root() -> Path:
return fake_home / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
return lambda: setattr(supervise, "bot_bottle_root", original)
sv_types.bot_bottle_root = fake_root # type: ignore[assignment]
def restore() -> None:
supervise.bot_bottle_root = original_sv # type: ignore[assignment]
sv_types.bot_bottle_root = original_svt # type: ignore[assignment]
return restore
def test_write_then_read_single_entry(self):
e = AuditEntry(
@@ -237,6 +266,7 @@ class TestAuditLog(unittest.TestCase):
diff="--- before\n+++ after\n",
)
path = 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")
self.assertEqual([e], loaded)
@@ -252,12 +282,13 @@ class TestAuditLog(unittest.TestCase):
justification="",
diff="",
))
path = audit_log_path("egress", "dev")
with path.open() as f:
lines = [line for line in f if line.strip()]
self.assertEqual(3, len(lines))
for line in lines:
self.assertTrue(json.loads(line)) # each line is valid JSON
entries = 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",
"2026-05-25T12:00:02+00:00"],
[entry.timestamp for entry in entries],
)
def test_separate_logs_per_component_slug(self):
write_audit_entry(AuditEntry(
@@ -369,17 +400,24 @@ class TestSupervisePrepare(unittest.TestCase):
self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original = supervise.bot_bottle_root
original_sv = supervise.bot_bottle_root
original_svt = sv_types.bot_bottle_root
def fake_root() -> Path:
return fake_home / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
return lambda: setattr(supervise, "bot_bottle_root", original)
sv_types.bot_bottle_root = fake_root # type: ignore[assignment]
def restore() -> None:
supervise.bot_bottle_root = original_sv # type: ignore[assignment]
sv_types.bot_bottle_root = original_svt # type: ignore[assignment]
return restore
def test_prepare_creates_queue(self):
plan = _StubSupervise().prepare("dev", self.stage_dir)
self.assertTrue(plan.queue_dir.is_dir())
self.assertTrue(plan.db_path.is_file())
self.assertEqual("dev", plan.slug)
self.assertEqual("", plan.internal_network)
+29 -29
View File
@@ -12,7 +12,10 @@ from pathlib import Path
from unittest.mock import patch
from bot_bottle import supervise
from bot_bottle import supervise_types as sv_types
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,
@@ -52,13 +55,22 @@ class _FakeHomeMixin:
def _setup_fake_home(self):
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-test.")
original = supervise.bot_bottle_root
original_sv = supervise.bot_bottle_root
original_svt = sv_types.bot_bottle_root
def fake_root() -> Path:
return Path(self._tmp.name) / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
self._restore_home = lambda: setattr(supervise, "bot_bottle_root", original)
sv_types.bot_bottle_root = fake_root # type: ignore[assignment]
def restore() -> None:
supervise.bot_bottle_root = original_sv # type: ignore[assignment]
sv_types.bot_bottle_root = original_svt # type: ignore[assignment]
self._restore_home = restore
QueueStore("").migrate()
AuditStore().migrate()
def _teardown_fake_home(self):
self._restore_home()
@@ -77,9 +89,7 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
def test_walks_all_slug_subdirs(self):
for slug in ("dev", "api"):
qdir = supervise.queue_dir_for_slug(slug)
qdir.mkdir(parents=True)
supervise.write_proposal(qdir, _proposal(slug=slug))
supervise.write_proposal(_proposal(slug=slug))
pending = supervise_cli.discover_pending()
self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})
@@ -97,18 +107,14 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
)
for p in (late, early):
qdir = supervise.queue_dir_for_slug(p.bottle_slug)
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
supervise.write_proposal(p)
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()
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True)
supervise.write_proposal(qdir, p)
supervise.write_response(qdir, supervise.Response(
supervise.write_proposal(p)
supervise.write_response("dev", supervise.Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="",
))
self.assertEqual([], supervise_cli.discover_pending())
@@ -123,10 +129,8 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW):
p = _proposal(tool=tool)
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
supervise.write_proposal(p)
return supervise_cli.QueuedProposal(proposal=p)
def test_approve_writes_response(self):
qp = self._enqueue()
@@ -135,7 +139,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
return_value=("routes: []\n", "routes:\n - host: example.com\n"),
):
supervise_cli.approve(qp)
resp = read_response(qp.queue_dir, qp.proposal.id)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status)
self.assertIsNone(resp.final_file)
@@ -150,7 +154,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
final_file="routes:\n - host: edited.example.com\n",
notes="tweaked",
)
resp = read_response(qp.queue_dir, qp.proposal.id)
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)
@@ -158,7 +162,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_reject_writes_rejection(self):
qp = self._enqueue()
supervise_cli.reject(qp, reason="nope")
resp = read_response(qp.queue_dir, qp.proposal.id)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status)
self.assertEqual("nope", resp.notes)
@@ -181,36 +185,33 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
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 queue dir for the response; TUI must not archive it.
resp = read_response(qp.queue_dir, qp.proposal.id)
# 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)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_tui_gitleaks_allow_requires_reason(self):
qp = self._enqueue(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)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_tui_gitleaks_allow_writes_reason(self):
qp = self._enqueue(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.queue_dir, qp.proposal.id)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
self.assertEqual("test fixture", resp.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 queue dir for the response; the TUI must
# The egress addon polls the DB for the response; the TUI must
# not archive it (the addon archives after reading).
resp = read_response(qp.queue_dir, qp.proposal.id)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status)
self.assertEqual("false positive", resp.notes)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_token_allow_writes_no_audit_log(self):
qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
@@ -222,14 +223,13 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
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)
self.assertFalse((qp.queue_dir / "processed").exists())
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.queue_dir, qp.proposal.id)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
self.assertEqual("legit", resp.notes)
def test_suffix_for_token_allow_is_txt(self):
+129 -57
View File
@@ -4,7 +4,6 @@ fallback paths."""
from __future__ import annotations
import os
import tempfile
import time
import unittest
@@ -12,14 +11,19 @@ from pathlib import Path
from unittest.mock import patch
from bot_bottle import supervise
from bot_bottle.audit_store import AuditStore
from bot_bottle.queue_store import QueueStore
from bot_bottle.supervise import (
AuditEntry,
Proposal,
STATUS_APPROVED,
TOOL_EGRESS_ALLOW,
list_pending_proposals,
read_audit_entries,
read_proposal,
read_response,
wait_for_response,
write_audit_entry,
)
@@ -37,58 +41,59 @@ class TestPathHelpers(unittest.TestCase):
def test_bot_bottle_root(self) -> None:
self.assertTrue(str(supervise.bot_bottle_root()).endswith(".bot-bottle"))
def test_queue_dir_for_slug(self) -> None:
self.assertIn("slug", str(supervise.queue_dir_for_slug("slug")))
def test_id_from_non_proposal_filename(self) -> None:
self.assertIsNone(supervise._id_from_proposal_filename(Path("x.response.json")))
class TestReadMalformed(unittest.TestCase):
def test_read_proposal_non_dict(self) -> None:
def test_read_proposal_missing_row(self) -> None:
with tempfile.TemporaryDirectory() as d:
(Path(d) / "p.proposal.json").write_text("[]")
with self.assertRaises(ValueError):
read_proposal(Path(d), "p")
with patch.dict("os.environ", {"HOME": d}):
QueueStore("slug").migrate()
with self.assertRaises(FileNotFoundError):
read_proposal("slug", "p")
def test_read_response_non_dict(self) -> None:
def test_read_response_missing_row(self) -> None:
with tempfile.TemporaryDirectory() as d:
(Path(d) / "p.response.json").write_text("[]")
with self.assertRaises(ValueError):
read_response(Path(d), "p")
with patch.dict("os.environ", {"HOME": d}):
QueueStore("slug").migrate()
with self.assertRaises(FileNotFoundError):
read_response("slug", "p")
def test_list_pending_skips_malformed(self) -> None:
def test_list_pending_reads_db_only(self) -> None:
with tempfile.TemporaryDirectory() as d:
qd = Path(d)
(qd / "bad.proposal.json").write_text("{ not json")
(qd / "arr.proposal.json").write_text("[]")
(qd / "incomplete.proposal.json").write_text("{}") # from_dict raises
supervise.write_proposal(qd, _proposal()) # one valid
pending = list_pending_proposals(qd)
with patch.dict("os.environ", {"HOME": d}):
QueueStore("slug").migrate()
supervise.write_proposal(_proposal())
pending = list_pending_proposals("slug")
self.assertEqual(1, len(pending))
self.assertEqual("slug", pending[0].bottle_slug)
def test_list_pending_skips_when_response_present(self) -> None:
with tempfile.TemporaryDirectory() as d:
qd = Path(d)
p = _proposal()
supervise.write_proposal(qd, p)
(qd / f"{p.id}.response.json").write_text("{}") # response exists -> skipped
self.assertEqual([], list_pending_proposals(qd))
with patch.dict("os.environ", {"HOME": d}):
QueueStore("slug").migrate()
p = _proposal()
supervise.write_proposal(p)
supervise.write_response("slug", supervise.Response(
proposal_id=p.id,
status=STATUS_APPROVED,
notes="",
))
self.assertEqual([], list_pending_proposals("slug"))
class TestWaitForResponse(unittest.TestCase):
def test_malformed_response_then_timeout(self) -> None:
def test_missing_response_times_out(self) -> None:
with tempfile.TemporaryDirectory() as d:
(Path(d) / "p.response.json").write_text("{ not json")
with self.assertRaises(TimeoutError):
wait_for_response(Path(d), "p", deadline=time.monotonic())
with patch.dict("os.environ", {"HOME": d}):
QueueStore("slug").migrate()
with self.assertRaises(TimeoutError):
wait_for_response("slug", "p", deadline=time.monotonic())
def test_incomplete_response_then_timeout(self) -> None:
def test_empty_db_response_does_not_count(self) -> None:
with tempfile.TemporaryDirectory() as d:
(Path(d) / "p.response.json").write_text("{}") # dict but from_dict raises
with self.assertRaises(TimeoutError):
wait_for_response(Path(d), "p", deadline=time.monotonic())
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):
@@ -97,35 +102,102 @@ class TestReadAuditEntries(unittest.TestCase):
patch.dict("os.environ", {"HOME": home}):
self.assertEqual([], read_audit_entries("egress", "nope"))
def test_skips_malformed_lines(self) -> None:
def test_reads_entries_from_db(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)
valid = (
'{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
' "operator_action": "approve", "operator_notes": "",'
' "justification": "", "diff": ""}'
)
path.write_text(
"\n" # blank line skipped
"{ not json\n" # JSONDecodeError skipped
"[]\n" # not a dict skipped
"{}\n" # missing fields -> ValueError skipped
+ valid + "\n"
)
AuditStore().migrate()
write_audit_entry(AuditEntry(
timestamp="t",
bottle_slug="slug",
component="egress",
operator_action="approve",
operator_notes="",
justification="",
diff="",
))
write_audit_entry(AuditEntry(
timestamp="t",
bottle_slug="other",
component="egress",
operator_action="reject",
operator_notes="",
justification="",
diff="",
))
entries = read_audit_entries("egress", "slug")
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 TestFlockFallback(unittest.TestCase):
def test_flock_on_closed_fd_is_swallowed(self) -> None:
# flock on a closed fd raises OSError(EBADF), which the helpers swallow.
fd = os.open(os.devnull, os.O_RDONLY)
os.close(fd)
supervise._try_flock(fd)
supervise._try_funlock(fd)
class TestStoreGuardBranches(unittest.TestCase):
"""Direct QueueStore / AuditStore construction and early-return guard branches."""
def test_queue_store_explicit_db_path(self):
with tempfile.TemporaryDirectory() as d:
db = Path(d) / "q.db"
store = QueueStore("key", db_path=db)
store.migrate()
self.assertTrue(db.is_file())
self.assertEqual(db, store.db_path)
def test_queue_store_missing_db_list_pending_returns_empty(self):
with tempfile.TemporaryDirectory() as d:
db = Path(d) / "q.db"
store = QueueStore("key", db_path=db)
store.migrate()
db.unlink()
self.assertEqual([], store.list_pending_proposals())
def test_queue_store_missing_db_list_all_returns_empty(self):
with tempfile.TemporaryDirectory() as d:
db = Path(d) / "q.db"
store = QueueStore("key", db_path=db)
store.migrate()
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"
store = QueueStore("key", db_path=db)
with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
store.migrate() # must not raise
def test_audit_store_missing_db_read_returns_empty(self):
with tempfile.TemporaryDirectory() as d:
db = Path(d) / "a.db"
store = AuditStore(db_path=db)
store.migrate()
db.unlink()
self.assertEqual([], store.read_audit_entries("egress", "slug"))
def test_audit_store_chmod_oserror_is_swallowed(self):
with tempfile.TemporaryDirectory() as d:
db = Path(d) / "a.db"
store = AuditStore(db_path=db)
with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
store.migrate() # must not raise
if __name__ == "__main__":
+38 -18
View File
@@ -17,8 +17,12 @@ from unittest.mock import patch
# bare name `supervise`.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "bot_bottle"))
import supervise as _sv # noqa: E402 # type: ignore
import queue_store as _qs # noqa: E402 # type: ignore
import audit_store as _as # noqa: E402 # type: ignore
import supervise_types as svt_flat # noqa: E402 # type: ignore
from bot_bottle import supervise_server # noqa: E402
from bot_bottle import supervise_types as svt_pkg # noqa: E402
from bot_bottle.supervise_server import (
ERR_INTERNAL,
ERR_INVALID_PARAMS,
@@ -112,7 +116,7 @@ class TestRpcErrorTaxonomy(unittest.TestCase):
validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
def test_unknown_tool_in_tools_call_is_client_error(self):
config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
config = ServerConfig(bottle_slug="dev")
with self.assertRaises(_RpcClientError) as cm:
handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
@@ -122,9 +126,9 @@ class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
def test_write_proposal_os_error_raises_internal(self):
config = ServerConfig(
bottle_slug="dev",
queue_dir=Path("/dev/null/cannot-exist"),
)
with self.assertRaises(_RpcInternalError) as cm:
with patch.object(_sv, "write_proposal", side_effect=OSError("disk full")), \
self.assertRaises(_RpcInternalError) as cm:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
@@ -265,21 +269,43 @@ class TestHandleToolsList(unittest.TestCase):
class TestHandleToolsCall(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-server-test.")
self.queue_dir = Path(self._tmp.name)
self.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
self._home_patch = self._patch_home(Path(self._tmp.name))
self.config = ServerConfig(bottle_slug="dev")
_qs.QueueStore("dev").migrate()
_as.AuditStore().migrate()
def tearDown(self):
self._home_patch()
self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original_sv = _sv.bot_bottle_root
original_flat = svt_flat.bot_bottle_root
original_pkg = svt_pkg.bot_bottle_root
def fake_root() -> Path:
return fake_home / ".bot-bottle"
_sv.bot_bottle_root = fake_root # type: ignore[assignment]
svt_flat.bot_bottle_root = fake_root # type: ignore[assignment]
svt_pkg.bot_bottle_root = fake_root # type: ignore[assignment]
def restore() -> None:
_sv.bot_bottle_root = original_sv # type: ignore[assignment]
svt_flat.bot_bottle_root = original_flat # type: ignore[assignment]
svt_pkg.bot_bottle_root = original_pkg # type: ignore[assignment]
return restore
def _respond_when_proposal_appears(self, status: str, notes: str = "") -> threading.Thread:
"""Background thread: poll the queue for a fresh proposal, write a
matching response. Returns the thread so the test can join it."""
def runner():
for _ in range(200):
pending = _sv.list_pending_proposals(self.queue_dir)
pending = _sv.list_pending_proposals("dev")
if pending:
p = pending[0]
_sv.write_response(self.queue_dir, _sv.Response(
_sv.write_response("dev", _sv.Response(
proposal_id=p.id, status=status, notes=notes,
))
return
@@ -412,15 +438,11 @@ class TestHandleToolsCall(unittest.TestCase):
finally:
responder.join()
# No pending proposals left after archive.
self.assertEqual([], _sv.list_pending_proposals(self.queue_dir))
# Both files moved to processed/.
processed = list((self.queue_dir / "processed").glob("*.json"))
self.assertEqual(2, len(processed))
self.assertEqual([], _sv.list_pending_proposals("dev"))
def test_pending_response_times_out_without_archive(self):
config = ServerConfig(
bottle_slug="dev",
queue_dir=self.queue_dir,
response_timeout_seconds=0.05,
)
result = handle_tools_call(
@@ -438,8 +460,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(self.queue_dir)))
self.assertFalse((self.queue_dir / "processed").exists())
self.assertEqual(1, len(_sv.list_pending_proposals("dev")))
class TestHandleListEgressRoutes(unittest.TestCase):
@@ -461,7 +482,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes(
{},
ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
ServerConfig(bottle_slug="dev"),
)
self.assertFalse(result["isError"]) # type: ignore[index]
@@ -476,7 +497,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes(
{},
ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
ServerConfig(bottle_slug="dev"),
)
self.assertTrue(result["isError"]) # type: ignore[index]
@@ -544,7 +565,6 @@ class TestHttpEndToEnd(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-http-test.")
self.queue_dir = Path(self._tmp.name)
# Pick a random port by binding to :0 first.
import socket
s = socket.socket()
@@ -552,7 +572,7 @@ class TestHttpEndToEnd(unittest.TestCase):
self.port = s.getsockname()[1]
s.close()
self.server = MCPServer(("127.0.0.1", self.port), MCPHandler)
self.server.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
self.server.config = ServerConfig(bottle_slug="dev")
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True,
)