Compare commits

..

1 Commits

Author SHA1 Message Date
didericis 2cdedbb7ca docs(prd): add PRD for egress control plane
lint / lint (push) Successful in 2m13s
Out-of-band egress enforcement & cost-control plane: meter token usage
at the egress proxy, evaluate budgets with agent→bottle→parent→global
precedence, and force cutoff/freeze/kill without the agent in the loop.
Introduces a host-level SQLite ledger behind a thin repository API and a
host-only TUI dashboard. Closes the design discussion on #251.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NkwFXLFff9PYPy4wgVBJp9
2026-06-29 11:01:47 -04:00
42 changed files with 743 additions and 1373 deletions
+2 -4
View File
@@ -18,7 +18,7 @@
# /git-gate-entrypoint.sh docker-cp'd at start time # /git-gate-entrypoint.sh docker-cp'd at start time
# /git-gate/creds/* docker-cp'd at start time # /git-gate/creds/* docker-cp'd at start time
# /git/* bare repos, populated at runtime # /git/* bare repos, populated at runtime
# /run/supervise/bot-bottle.db bind-mounted at run time # /run/supervise/queue/ bind-mounted at run time
# /home/mitmproxy/.mitmproxy/ mitmproxy CA dir # /home/mitmproxy/.mitmproxy/ mitmproxy CA dir
# #
# Exposed ports inside the container: # Exposed ports inside the container:
@@ -66,8 +66,6 @@ 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/egress_addon.py /app/egress_addon.py
COPY bot_bottle/dlp_detectors.py /app/dlp_detectors.py COPY bot_bottle/dlp_detectors.py /app/dlp_detectors.py
COPY bot_bottle/yaml_subset.py /app/yaml_subset.py COPY bot_bottle/yaml_subset.py /app/yaml_subset.py
COPY bot_bottle/queue_store.py /app/queue_store.py
COPY bot_bottle/audit_store.py /app/audit_store.py
COPY bot_bottle/supervise.py /app/supervise.py COPY bot_bottle/supervise.py /app/supervise.py
COPY bot_bottle/supervise_server.py /app/supervise_server.py COPY bot_bottle/supervise_server.py /app/supervise_server.py
COPY bot_bottle/sidecar_init.py /app/sidecar_init.py COPY bot_bottle/sidecar_init.py /app/sidecar_init.py
@@ -83,7 +81,7 @@ RUN mkdir -p \
/etc/git-gate \ /etc/git-gate \
/git-gate/creds \ /git-gate/creds \
/git \ /git \
/run/supervise \ /run/supervise/queue \
/home/mitmproxy/.mitmproxy /home/mitmproxy/.mitmproxy
# Documentation only — the compose renderer publishes whichever # Documentation only — the compose renderer publishes whichever
+2 -2
View File
@@ -5,8 +5,8 @@
# bot-bottle # bot-bottle
[![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml) [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/) [![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
[![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md) [![core coverage](https://img.shields.io/badge/core%20coverage-96%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data. **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
-9
View File
@@ -209,15 +209,6 @@ class AgentProvider(ABC):
the supervise sidecar is reachable. No-op when the supervise sidecar is reachable. No-op when
`plan.supervise_plan is None`.""" `plan.supervise_plan is None`."""
@abstractmethod
def headless_prompt(self, prompt: str) -> list[str]:
"""Return the agent CLI args that deliver `prompt` as the
initial task in a non-interactive (headless) session.
Called only when ``--prompt`` is passed to
``./cli.py start --headless``; the returned args are appended
after the provider's ``bypass_args`` and ``startup_args``."""
def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None: def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None:
"""Install the egress MITM CA into the agent's trust store. """Install the egress MITM CA into the agent's trust store.
-120
View File
@@ -1,120 +0,0 @@
"""SQLite-backed audit store for supervise (PRD 0013)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .supervise import AuditEntry
def _sv() -> object:
"""Lazy import of supervise to avoid a circular-import at module init time.
Mirrors our own module identity so patches on supervise.bot_bottle_root
propagate correctly in both flat (sidecar / sys.path-injection tests) and
package contexts."""
import sys
sv_name = "supervise" if __name__ == "audit_store" else "bot_bottle.supervise"
if sv_name in sys.modules:
return sys.modules[sv_name]
try:
import bot_bottle.supervise as _m
except ImportError:
import supervise as _m # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
return _m
def _audit_entry_from_row(row: sqlite3.Row) -> AuditEntry:
m = _sv()
return m.AuditEntry( # type: ignore[attr-defined]
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"],
)
def _host_db_path() -> Path:
return _sv().host_db_path() # type: ignore[attr-defined,no-any-return]
class AuditStore:
"""SQLite-backed persistent store for supervise audit entries."""
def __init__(self, db_path: Path | None = None) -> None:
self.db_path = db_path or _host_db_path()
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._init()
def write_audit_entry(self, entry: AuditEntry) -> Path:
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 [_audit_entry_from_row(row) for row in rows]
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _init(self) -> None:
with self._connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS 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
)
"""
)
self._chmod()
def _chmod(self) -> None:
try:
self.db_path.chmod(0o600)
except OSError:
pass
__all__ = ["AuditStore"]
+5 -4
View File
@@ -34,7 +34,7 @@ from ...egress import (
from ...git_gate import GIT_GATE_HOSTNAME from ...git_gate import GIT_GATE_HOSTNAME
from ...log import die, warn from ...log import die, warn
from ...supervise import ( from ...supervise import (
DB_PATH_IN_CONTAINER, QUEUE_DIR_IN_CONTAINER,
SUPERVISE_HOSTNAME, SUPERVISE_HOSTNAME,
SUPERVISE_PORT, SUPERVISE_PORT,
) )
@@ -163,15 +163,16 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
if sp is not None: if sp is not None:
env += [ env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}", f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}", f"SUPERVISE_PORT={SUPERVISE_PORT}",
] ]
volumes.append({ volumes.append({
"type": "bind", "type": "bind",
"source": str(sp.db_path), "source": str(sp.queue_dir),
"target": DB_PATH_IN_CONTAINER, "target": QUEUE_DIR_IN_CONTAINER,
"read_only": False, "read_only": False,
}) })
internal_aliases = [EGRESS_HOSTNAME] internal_aliases = [EGRESS_HOSTNAME]
if gp.upstreams: if gp.upstreams:
internal_aliases.append(GIT_GATE_HOSTNAME) internal_aliases.append(GIT_GATE_HOSTNAME)
+1 -9
View File
@@ -37,10 +37,7 @@ from pathlib import Path
from typing import Callable, Generator from typing import Callable, Generator
from ...egress import egress_resolve_token_values from ...egress import egress_resolve_token_values
from ...git_gate import ( from ...git_gate import revoke_git_gate_provisioned_keys
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import info, warn from ...log import info, warn
from . import network as network_mod from . import network as network_mod
from . import util as docker_mod from . import util as docker_mod
@@ -121,11 +118,6 @@ def launch(
git_gate_plan = plan.git_gate_plan git_gate_plan = plan.git_gate_plan
if git_gate_plan.upstreams: if git_gate_plan.upstreams:
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
git_gate_plan,
git_gate_state_dir(plan.slug),
)
git_gate_plan = dataclasses.replace( git_gate_plan = dataclasses.replace(
git_gate_plan, git_gate_plan,
internal_network=internal_network, internal_network=internal_network,
+4 -22
View File
@@ -28,12 +28,9 @@ from ...egress import (
egress_resolve_token_values, egress_resolve_token_values,
egress_sidecar_env_entries, egress_sidecar_env_entries,
) )
from ...git_gate import ( from ...git_gate import revoke_git_gate_provisioned_keys
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import die, info, warn from ...log import die, info, warn
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde from ...util import expand_tilde
from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
from ..docker.git_gate import ( from ..docker.git_gate import (
@@ -101,8 +98,6 @@ def launch(
egress_network = egress_network_name(plan.slug) egress_network = egress_network_name(plan.slug)
_create_networks(internal_network, egress_network, stack) _create_networks(internal_network, egress_network, stack)
plan = _provision_git_gate_keys(plan)
sidecar_name = sidecar_container_name(plan.slug) sidecar_name = sidecar_container_name(plan.slug)
container_mod.force_remove_container(sidecar_name) container_mod.force_remove_container(sidecar_name)
_start_sidecar_bundle(plan, sidecar_name, internal_network, egress_network) _start_sidecar_bundle(plan, sidecar_name, internal_network, egress_network)
@@ -246,19 +241,6 @@ def _stamp_agent_urls(
) )
def _provision_git_gate_keys(
plan: MacosContainerBottlePlan,
) -> MacosContainerBottlePlan:
if not plan.git_gate_plan.upstreams:
return plan
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
plan.git_gate_plan,
git_gate_state_dir(plan.slug),
)
return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
def _stage_git_gate(plan: MacosContainerBottlePlan, sidecar_name: str) -> None: def _stage_git_gate(plan: MacosContainerBottlePlan, sidecar_name: str) -> None:
gp = plan.git_gate_plan gp = plan.git_gate_plan
if not gp.upstreams: if not gp.upstreams:
@@ -379,7 +361,7 @@ def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
if plan.supervise_plan is not None: if plan.supervise_plan is not None:
env += [ env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}", f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}", f"SUPERVISE_PORT={SUPERVISE_PORT}",
] ]
return tuple(env) return tuple(env)
@@ -405,7 +387,7 @@ def _sidecar_mounts(
sp = plan.supervise_plan sp = plan.supervise_plan
if sp is not None: if sp is not None:
mounts.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False)) mounts.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
return tuple(mounts) return tuple(mounts)
+4 -21
View File
@@ -27,7 +27,7 @@ from ...egress import (
egress_resolve_token_values, egress_resolve_token_values,
egress_sidecar_env_entries, egress_sidecar_env_entries,
) )
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde from ...util import expand_tilde
from ..docker import util as docker_mod from ..docker import util as docker_mod
from ..docker.egress import ( from ..docker.egress import (
@@ -41,10 +41,7 @@ from ..docker.git_gate import (
GIT_GATE_ENTRYPOINT_IN_CONTAINER, GIT_GATE_ENTRYPOINT_IN_CONTAINER,
GIT_GATE_HOOK_IN_CONTAINER, GIT_GATE_HOOK_IN_CONTAINER,
) )
from ...git_gate import ( from ...git_gate import revoke_git_gate_provisioned_keys
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import info, warn from ...log import info, warn
from ...bottle_state import ( from ...bottle_state import (
egress_state_dir, egress_state_dir,
@@ -177,7 +174,6 @@ def _start_bundle(
) -> SmolmachinesBottlePlan: ) -> SmolmachinesBottlePlan:
"""Build the BundleLaunchSpec, resolve token env, start the """Build the BundleLaunchSpec, resolve token env, start the
sidecar bundle container, and register teardown.""" sidecar bundle container, and register teardown."""
plan = _provision_git_gate_keys(plan)
bundle_spec = _bundle_launch_spec(plan, network, loopback_ip) bundle_spec = _bundle_launch_spec(plan, network, loopback_ip)
token_env = _resolve_token_env(plan, dict(os.environ)) token_env = _resolve_token_env(plan, dict(os.environ))
_bundle.ensure_bundle_image(bundle_spec.image) _bundle.ensure_bundle_image(bundle_spec.image)
@@ -186,19 +182,6 @@ def _start_bundle(
return plan return plan
def _provision_git_gate_keys(
plan: SmolmachinesBottlePlan,
) -> SmolmachinesBottlePlan:
if not plan.git_gate_plan.upstreams:
return plan
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
plan.git_gate_plan,
git_gate_state_dir(plan.slug),
)
return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
def _discover_urls( def _discover_urls(
plan: SmolmachinesBottlePlan, plan: SmolmachinesBottlePlan,
loopback_ip: str, loopback_ip: str,
@@ -369,10 +352,10 @@ def _bundle_launch_spec(
daemons.append("supervise") daemons.append("supervise")
env += [ env += [
f"SUPERVISE_BOTTLE_SLUG={plan.slug}", f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
f"SUPERVISE_PORT={SUPERVISE_PORT}", f"SUPERVISE_PORT={SUPERVISE_PORT}",
] ]
volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False)) volumes.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
# Container ports the agent reaches from the smolvm guest — # Container ports the agent reaches from the smolvm guest —
# published on host loopback so the guest can dial via TSI + # published on host loopback so the guest can dial via TSI +
+3 -2
View File
@@ -284,8 +284,9 @@ def git_gate_state_dir(identity: str) -> Path:
def supervise_state_dir(identity: str) -> Path: def supervise_state_dir(identity: str) -> Path:
"""State subdir reserved for supervise sidecar bind-mount sources. """State subdir reserved for supervise sidecar bind-mount sources.
Runtime queue/audit rows live in the host-level bot-bottle SQLite The queue dir is intentionally NOT under here it lives at
database, so they survive state-dir cleanup.""" ~/.bot-bottle/queue/<slug>/ alongside the audit logs, so it
survives state-dir cleanup."""
return bottle_state_dir(identity) / _SUPERVISE_SUBDIR return bottle_state_dir(identity) / _SUPERVISE_SUBDIR
+7 -142
View File
@@ -2,11 +2,6 @@
interactive claude-code session. The container is torn down when the interactive claude-code session. The container is torn down when the
session ends. session ends.
`--headless` selects a non-interactive launch (agent/bottles/label from
flags, no TUI selectors, no y/N prompt) for orchestrators,
CI, and webhook dispatch. The agent still execs on the inherited
stdio/PTY, so an orchestrator that allocates the PTY drives the session.
The launch core is shared with `cli.py resume <identity>` through The launch core is shared with `cli.py resume <identity>` through
the private orchestrator `_launch_bottle`. the private orchestrator `_launch_bottle`.
""" """
@@ -21,7 +16,7 @@ import tempfile
from pathlib import Path from pathlib import Path
from typing import Callable from typing import Callable
from ..agent_provider import get_provider, runtime_for from ..agent_provider import runtime_for
from ..backend import ( from ..backend import (
Bottle, Bottle,
BottleSpec, BottleSpec,
@@ -36,7 +31,7 @@ from ..bottle_state import (
is_preserved, is_preserved,
mark_preserved, mark_preserved,
) )
from ..log import info, die from ..log import info
from ..manifest import Manifest, ManifestIndex from ..manifest import Manifest, ManifestIndex
from ._common import PROG, USER_CWD, read_tty_line from ._common import PROG, USER_CWD, read_tty_line
from . import tui from . import tui
@@ -55,39 +50,6 @@ def cmd_start(argv: list[str]) -> int:
"or host auto-selection). Overrides the env var when set." "or host auto-selection). Overrides the env var when set."
), ),
) )
parser.add_argument(
"--headless",
action="store_true",
help=(
"non-interactive launch: take agent/bottles/label from flags, "
"skip all prompts. For orchestrators, CI, and webhooks."
),
)
parser.add_argument(
"--bottle",
action="append",
default=None,
metavar="NAME",
help=(
"bottle to compose, repeatable (order = merge order). In "
"--headless, defaults to the agent's own bottle when omitted."
),
)
parser.add_argument(
"--label",
default=None,
help="bottle label / terminal title (--headless default: agent name)",
)
parser.add_argument(
"--color",
default=None,
help="bottle color, one of the 16 ANSI color names (--headless default: none)",
)
parser.add_argument(
"--prompt",
default=None,
help="initial task prompt delivered to the agent (required with --headless)",
)
parser.add_argument( parser.add_argument(
"name", "name",
nargs="?", nargs="?",
@@ -99,12 +61,6 @@ def cmd_start(argv: list[str]) -> int:
dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1" dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
manifest = ManifestIndex.resolve(USER_CWD) manifest = ManifestIndex.resolve(USER_CWD)
backend_name: str | None = args.backend
if args.headless:
return _start_headless(
manifest, args, dry_run=dry_run, backend_name=backend_name
)
agent_name: str | None = args.name agent_name: str | None = args.name
if agent_name is None: if agent_name is None:
@@ -115,6 +71,8 @@ def cmd_start(argv: list[str]) -> int:
if agent_name is None: if agent_name is None:
return 0 return 0
backend_name: str | None = args.backend
# Bottle multiselect: always show after agent selection so operators # Bottle multiselect: always show after agent selection so operators
# can compose bottles at launch time without editing agent manifests. # can compose bottles at launch time without editing agent manifests.
available_bottles = manifest.all_bottle_names available_bottles = manifest.all_bottle_names
@@ -151,83 +109,6 @@ def cmd_start(argv: list[str]) -> int:
) )
# --- Headless launch -----------------------------------------------------
def _start_headless(
manifest: ManifestIndex,
args: argparse.Namespace,
*,
dry_run: bool,
backend_name: str | None,
) -> int:
"""Non-interactive launch path for orchestrators / CI / webhooks.
Resolves agent, bottles, label, and color from flags + manifest
defaults instead of the TUI selectors, and auto-confirms the
preflight. Otherwise runs the same launch core as the interactive
path, so the agent still execs on the inherited stdio/PTY an
orchestrator allocates that PTY and relays it to its
desktop/mobile clients."""
agent_name = args.name
if not agent_name:
die("--headless requires an agent name: ./cli.py start <agent> --headless")
manifest.require_agent(agent_name) # raises ManifestError if unknown
prompt = args.prompt
if not prompt:
die(
"--headless requires --prompt: "
"./cli.py start <agent> --headless --prompt 'Do the thing'"
)
if args.bottle:
bottle_names: tuple[str, ...] = tuple(args.bottle)
else:
default_bottle = _peek_agent_bottle(manifest, agent_name)
if not default_bottle:
die(
f"--headless: agent '{agent_name}' has no default bottle; "
f"pass one or more --bottle NAME"
)
bottle_names = (default_bottle,)
label = _uniquify_label_headless(args.label or agent_name)
spec = BottleSpec(
manifest=manifest,
agent_name=agent_name,
copy_cwd=args.cwd,
user_cwd=USER_CWD,
label=label,
color=args.color or "",
bottle_names=bottle_names,
)
return _launch_bottle(
spec,
dry_run=dry_run,
backend_name=backend_name,
assume_yes=True,
headless_prompt_text=prompt,
)
def _uniquify_label_headless(label: str) -> str:
"""Non-interactive analog of `_resolve_unique_label`: if the label's
slug collides with a running bottle, append -2, -3, until free,
logging the chosen label. Orchestrators fire-and-forget many bottles,
so silently picking a free name beats erroring on every collision."""
active_slugs = {a.slug for a in enumerate_active_agents()}
if docker_mod.slugify(label) not in active_slugs:
return label
n = 2
while docker_mod.slugify(f"{label}-{n}") in active_slugs:
n += 1
chosen = f"{label}-{n}"
info(f"label '{label}' already in use; using '{chosen}'")
return chosen
# --- Launch helpers ------------------------------------------------------ # --- Launch helpers ------------------------------------------------------
@@ -495,19 +376,10 @@ def _launch_bottle(
*, *,
dry_run: bool, dry_run: bool,
backend_name: str | None = None, backend_name: str | None = None,
assume_yes: bool = False,
headless_prompt_text: str = "",
) -> int: ) -> int:
"""Shared launch core for `start` and `resume`. Builds the plan, """Shared launch core for `start` and `resume`. Builds the plan,
prints / dry-runs / prompts as appropriate, brings the bottle up, prints / dry-runs / prompts as appropriate, brings the bottle up,
attaches claude, and prints the resume hint on session end. attaches claude, and prints the resume hint on session end."""
`assume_yes` skips the interactive y/N confirmation (headless /
orchestrator launches), where there is no human at the prompt.
`headless_prompt_text` is passed to the provider's `headless_prompt`
method and the resulting args are appended to startup_args so the
agent receives the initial task without interactive input."""
stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage.")) stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
identity = "" identity = ""
try: try:
@@ -515,7 +387,7 @@ def _launch_bottle(
spec, spec,
stage_dir=stage_dir, stage_dir=stage_dir,
render_preflight=_text_render_preflight(), render_preflight=_text_render_preflight(),
prompt_yes=(lambda: True) if assume_yes else _text_prompt_yes, prompt_yes=_text_prompt_yes,
dry_run=dry_run, dry_run=dry_run,
backend_name=backend_name, backend_name=backend_name,
) )
@@ -525,17 +397,10 @@ def _launch_bottle(
backend = get_bottle_backend(backend_name) backend = get_bottle_backend(backend_name)
with backend.launch(plan) as bottle: with backend.launch(plan) as bottle:
agent_provider_template = getattr(plan, "agent_provider_template", "claude") agent_provider_template = getattr(plan, "agent_provider_template", "claude")
extra_args: tuple[str, ...] = ()
if headless_prompt_text:
extra_args = tuple(
get_provider(agent_provider_template).headless_prompt(
headless_prompt_text
)
)
exit_code = attach_agent( exit_code = attach_agent(
bottle, bottle,
agent_provider_template=agent_provider_template, agent_provider_template=agent_provider_template,
startup_args=plan.agent_provision.startup_args + extra_args, startup_args=plan.agent_provision.startup_args,
) )
info( info(
f"session ended (exit {exit_code}); " f"session ended (exit {exit_code}); "
+16 -9
View File
@@ -45,7 +45,7 @@ from ..supervise import (
TOOL_EGRESS_BLOCK, TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW, TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW, TOOL_EGRESS_TOKEN_ALLOW,
list_all_pending_proposals, list_pending_proposals,
render_diff, render_diff,
write_audit_entry, write_audit_entry,
write_response, write_response,
@@ -63,9 +63,10 @@ _REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_AL
@dataclass(frozen=True) @dataclass(frozen=True)
class QueuedProposal: class QueuedProposal:
"""A pending proposal from the supervise queue.""" """A pending proposal plus the queue dir it was found in."""
proposal: Proposal proposal: Proposal
queue_dir: Path
# Errors any remediation engine may raise. Caught by the TUI key # Errors any remediation engine may raise. Caught by the TUI key
@@ -85,11 +86,16 @@ def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
def discover_pending() -> list[QueuedProposal]: def discover_pending() -> list[QueuedProposal]:
"""Collect pending proposals across bottles.""" """Walk ~/.bot-bottle/queue/* and collect pending proposals."""
out = [ queue_root = _supervise.bot_bottle_root() / "queue"
QueuedProposal(proposal=proposal) if not queue_root.is_dir():
for proposal in list_all_pending_proposals() 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))
out.sort(key=lambda q: q.proposal.arrival_timestamp) out.sort(key=lambda q: q.proposal.arrival_timestamp)
return out return out
@@ -112,6 +118,7 @@ def _detail_lines(
(f"tool: {p.tool}", 0), (f"tool: {p.tool}", 0),
(f"id: {p.id}", 0), (f"id: {p.id}", 0),
(f"arrived: {p.arrival_timestamp}", 0), (f"arrived: {p.arrival_timestamp}", 0),
(f"queue: {qp.queue_dir}", 0),
("", 0), ("", 0),
("justification:", 0), ("justification:", 0),
] ]
@@ -158,7 +165,7 @@ def approve(
notes=notes, notes=notes,
final_file=final_file, final_file=final_file,
) )
write_response(qp.proposal.bottle_slug, response) write_response(qp.queue_dir, response)
_write_audit( _write_audit(
qp, action=status, notes=notes, qp, action=status, notes=notes,
diff_before=diff_before, diff_after=diff_after, diff_before=diff_before, diff_after=diff_after,
@@ -172,7 +179,7 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
notes=reason, notes=reason,
final_file=None, final_file=None,
) )
write_response(qp.proposal.bottle_slug, response) write_response(qp.queue_dir, response)
_write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="") _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
+3 -10
View File
@@ -217,7 +217,7 @@ class ClaudeAgentProvider(AgentProvider):
if not agent.skills: if not agent.skills:
return return
skills_dir = _skills_dir(plan.guest_home) skills_dir = _skills_dir(plan.guest_home)
bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root") bottle.exec(f"mkdir -p {skills_dir}", user="root")
for name in agent.skills: for name in agent.skills:
src = host_skill_dir(name) src = host_skill_dir(name)
if not os.path.isdir(src): if not os.path.isdir(src):
@@ -227,13 +227,9 @@ class ClaudeAgentProvider(AgentProvider):
) )
dst = f"{skills_dir}/{name}" dst = f"{skills_dir}/{name}"
info(f"copying skill {name} into {bottle.name}:{dst}") info(f"copying skill {name} into {bottle.name}:{dst}")
# Defense in depth: skill names are validated kebab-case at bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
# manifest load, but quote the path so a future unvalidated
# field can't inject shell metacharacters here either.
dst_q = shlex.quote(dst)
bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
bottle.cp_in(f"{src}/.", f"{dst}/") bottle.cp_in(f"{src}/.", f"{dst}/")
bottle.exec(f"chown -R node:node {dst_q}", user="root") bottle.exec(f"chown -R node:node {dst}", user="root")
def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None: def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
"""Copy the prompt file into the guest, fix ownership/mode. """Copy the prompt file into the guest, fix ownership/mode.
@@ -313,9 +309,6 @@ class ClaudeAgentProvider(AgentProvider):
f"claude mcp add --scope user --transport http supervise {supervise_url}" f"claude mcp add --scope user --transport http supervise {supervise_url}"
) )
def headless_prompt(self, prompt: str) -> list[str]:
return ["-p", prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None: def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root") result = bottle.exec(script, user="root")
+3 -10
View File
@@ -183,7 +183,7 @@ class CodexAgentProvider(AgentProvider):
if not agent.skills: if not agent.skills:
return return
skills_dir = _skills_dir(plan.guest_home) skills_dir = _skills_dir(plan.guest_home)
bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root") bottle.exec(f"mkdir -p {skills_dir}", user="root")
for name in agent.skills: for name in agent.skills:
src = host_skill_dir(name) src = host_skill_dir(name)
if not os.path.isdir(src): if not os.path.isdir(src):
@@ -193,13 +193,9 @@ class CodexAgentProvider(AgentProvider):
) )
dst = f"{skills_dir}/{name}" dst = f"{skills_dir}/{name}"
info(f"copying skill {name} into {bottle.name}:{dst}") info(f"copying skill {name} into {bottle.name}:{dst}")
# Defense in depth: skill names are validated kebab-case at bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
# manifest load, but quote the path so a future unvalidated
# field can't inject shell metacharacters here either.
dst_q = shlex.quote(dst)
bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
bottle.cp_in(f"{src}/.", f"{dst}/") bottle.cp_in(f"{src}/.", f"{dst}/")
bottle.exec(f"chown -R node:node {dst_q}", user="root") bottle.exec(f"chown -R node:node {dst}", user="root")
def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None: def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
"""Copy the prompt file into the guest, fix ownership/mode. """Copy the prompt file into the guest, fix ownership/mode.
@@ -279,9 +275,6 @@ class CodexAgentProvider(AgentProvider):
f"codex mcp add supervise --url {shlex.quote(supervise_url)}" f"codex mcp add supervise --url {shlex.quote(supervise_url)}"
) )
def headless_prompt(self, prompt: str) -> list[str]:
return [prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None: def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root") result = bottle.exec(script, user="root")
+3 -10
View File
@@ -238,7 +238,7 @@ class PiAgentProvider(AgentProvider):
if not agent.skills: if not agent.skills:
return return
skills_dir = _skills_dir(plan.guest_home) skills_dir = _skills_dir(plan.guest_home)
bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root") bottle.exec(f"mkdir -p {skills_dir}", user="root")
for name in agent.skills: for name in agent.skills:
src = host_skill_dir(name) src = host_skill_dir(name)
if not os.path.isdir(src): if not os.path.isdir(src):
@@ -248,13 +248,9 @@ class PiAgentProvider(AgentProvider):
) )
dst = f"{skills_dir}/{name}" dst = f"{skills_dir}/{name}"
info(f"copying skill {name} into {bottle.name}:{dst}") info(f"copying skill {name} into {bottle.name}:{dst}")
# Defense in depth: skill names are validated kebab-case at bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
# manifest load, but quote the path so a future unvalidated
# field can't inject shell metacharacters here either.
dst_q = shlex.quote(dst)
bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
bottle.cp_in(f"{src}/.", f"{dst}/") bottle.cp_in(f"{src}/.", f"{dst}/")
bottle.exec(f"chown -R node:node {dst_q}", user="root") bottle.exec(f"chown -R node:node {dst}", user="root")
def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None: def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
prompt_path = _prompt_path(plan.guest_home) prompt_path = _prompt_path(plan.guest_home)
@@ -315,9 +311,6 @@ class PiAgentProvider(AgentProvider):
) -> None: ) -> None:
del plan, bottle, supervise_url del plan, bottle, supervise_url
def headless_prompt(self, prompt: str) -> list[str]:
return ["-p", prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None: def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root") result = bottle.exec(script, user="root")
+9 -6
View File
@@ -79,13 +79,14 @@ class EgressAddon:
# only — a restart re-prompts. Mutated only from the asyncio loop that # only — a restart re-prompts. Mutated only from the asyncio loop that
# runs the addon hooks, so no lock is needed. # runs the addon hooks, so no lock is needed.
self.safe_tokens: set[str] = set() 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._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
self._token_allow_timeout = _token_allow_timeout_from_env(os.environ) self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
self._reload(initial=True) self._reload(initial=True)
self._install_sighup() self._install_sighup()
def _supervise_available(self) -> bool: def _supervise_available(self) -> bool:
return bool(self._supervise_slug) return bool(self._supervise_queue_dir and self._supervise_slug)
def _reload(self, *, initial: bool = False) -> None: def _reload(self, *, initial: bool = False) -> None:
try: try:
@@ -392,8 +393,9 @@ class EgressAddon:
justification=_TOKEN_ALLOW_JUSTIFICATION, justification=_TOKEN_ALLOW_JUSTIFICATION,
current_file_hash=_sv.sha256_hex(payload), current_file_hash=_sv.sha256_hex(payload),
) )
queue_dir = Path(self._supervise_queue_dir)
try: try:
_sv.write_proposal(proposal) _sv.write_proposal(queue_dir, proposal)
except OSError as e: except OSError as e:
sys.stderr.write( sys.stderr.write(
f"egress: could not queue token-allow proposal: {e}; " f"egress: could not queue token-allow proposal: {e}; "
@@ -409,8 +411,8 @@ class EgressAddon:
**self._req_ctx(flow), **self._req_ctx(flow),
}) + "\n") }) + "\n")
response = await self._await_token_response(proposal.id) response = await self._await_token_response(queue_dir, proposal.id)
_sv.archive_proposal(self._supervise_slug, proposal.id) _sv.archive_proposal(queue_dir, proposal.id)
if response is not None and response.status in ( if response is not None and response.status in (
_sv.STATUS_APPROVED, _sv.STATUS_MODIFIED, _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
@@ -437,15 +439,16 @@ class EgressAddon:
async def _await_token_response( async def _await_token_response(
self, self,
queue_dir: Path,
proposal_id: str, proposal_id: str,
) -> "_sv.Response | None": ) -> "_sv.Response | None":
"""Poll the DB for the operator's response without blocking the """Poll the queue dir for the operator's response without blocking the
proxy event loop. Returns the Response, or None on timeout.""" proxy event loop. Returns the Response, or None on timeout."""
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
deadline = loop.time() + self._token_allow_timeout deadline = loop.time() + self._token_allow_timeout
while True: while True:
try: try:
return _sv.read_response(self._supervise_slug, proposal_id) return _sv.read_response(queue_dir, proposal_id)
except (OSError, ValueError, KeyError): except (OSError, ValueError, KeyError):
# Not written yet, or a partial/malformed write — retry until # Not written yet, or a partial/malformed write — retry until
# the deadline, then fail closed. # the deadline, then fail closed.
+11 -6
View File
@@ -30,6 +30,7 @@ backend-specific and lives on concrete subclasses (see
from __future__ import annotations from __future__ import annotations
import dataclasses
from abc import ABC from abc import ABC
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -52,7 +53,6 @@ from .git_gate_render import (
_gitconfig_validate_value, _gitconfig_validate_value,
) )
from .git_gate_provision import ( from .git_gate_provision import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys, revoke_git_gate_provisioned_keys,
_provision_dynamic_key, _provision_dynamic_key,
_resolve_identity_file, _resolve_identity_file,
@@ -93,14 +93,20 @@ class GitGate(ABC):
entrypoint, pre-receive hook, and access-hook scripts (mode entrypoint, pre-receive hook, and access-hook scripts (mode
600) under `stage_dir`. Pure host-side, no docker subprocess. 600) under `stage_dir`. Pure host-side, no docker subprocess.
For `gitea` key entries, the returned upstream intentionally For `gitea` key entries, also generates and registers
has an empty identity file. Backend launch fills that in after a fresh deploy key via the forge API and writes the private key
the operator confirms the preflight. + key ID to `stage_dir`.
Returned plan is incomplete: the launch step must fill Returned plan is incomplete: the launch step must fill
`internal_network` / `egress_network` via `dataclasses.replace` `internal_network` / `egress_network` via `dataclasses.replace`
before passing the plan to `.start`.""" before passing the plan to `.start`."""
upstreams = git_gate_upstreams_for_bottle(bottle) upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
for i, entry in enumerate(bottle.git):
upstreams_list[i] = dataclasses.replace(
upstreams_list[i],
identity_file=_resolve_identity_file(entry, slug, stage_dir),
)
upstreams = tuple(upstreams_list)
entrypoint = stage_dir / "git_gate_entrypoint.sh" entrypoint = stage_dir / "git_gate_entrypoint.sh"
entrypoint.write_text(git_gate_render_entrypoint(upstreams)) entrypoint.write_text(git_gate_render_entrypoint(upstreams))
entrypoint.chmod(0o600) entrypoint.chmod(0o600)
@@ -156,7 +162,6 @@ __all__ = [
"git_gate_render_entrypoint", "git_gate_render_entrypoint",
"git_gate_render_hook", "git_gate_render_hook",
"git_gate_render_access_hook", "git_gate_render_access_hook",
"provision_git_gate_dynamic_keys",
"revoke_git_gate_provisioned_keys", "revoke_git_gate_provisioned_keys",
"_gitconfig_validate_value", "_gitconfig_validate_value",
"_provision_dynamic_key", "_provision_dynamic_key",
-43
View File
@@ -9,16 +9,10 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.
from __future__ import annotations from __future__ import annotations
import os import os
import dataclasses
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING
from .log import info from .log import info
from .manifest import ManifestBottle, ManifestGitEntry from .manifest import ManifestBottle, ManifestGitEntry
from .git_gate_render import GitGateUpstream
if TYPE_CHECKING:
from .git_gate import GitGatePlan
def _provision_dynamic_key( def _provision_dynamic_key(
entry: ManifestGitEntry, entry: ManifestGitEntry,
@@ -101,45 +95,8 @@ def _resolve_identity_file(entry: ManifestGitEntry, slug: str, stage_dir: Path)
return entry.IdentityFile return entry.IdentityFile
def provision_git_gate_dynamic_keys(
bottle: ManifestBottle,
plan: "GitGatePlan",
stage_dir: Path,
) -> "GitGatePlan":
"""Provision dynamic git-gate keys and return an updated plan.
This runs during backend launch, after the operator confirms the
preflight. Plan preparation intentionally stays side-effect-light:
dry-runs and aborted launches must not create remote deploy keys.
"""
if not plan.upstreams:
return plan
upstreams_by_name: dict[str, GitGateUpstream] = {
upstream.name: upstream for upstream in plan.upstreams
}
updated: list[GitGateUpstream] = []
for entry in bottle.git:
upstream = upstreams_by_name.get(entry.Name)
if upstream is None:
continue
if entry.Key.provider == "gitea":
identity_file = _provision_dynamic_key(entry, plan.slug, stage_dir)
upstream = dataclasses.replace(upstream, identity_file=identity_file)
updated.append(upstream)
if len(updated) != len(plan.upstreams):
updated_names = {u.name for u in updated}
for upstream in plan.upstreams:
if upstream.name not in updated_names:
updated.append(upstream)
return dataclasses.replace(plan, upstreams=tuple(updated))
__all__ = [ __all__ = [
"revoke_git_gate_provisioned_keys", "revoke_git_gate_provisioned_keys",
"provision_git_gate_dynamic_keys",
"_provision_dynamic_key", "_provision_dynamic_key",
"_resolve_identity_file", "_resolve_identity_file",
] ]
+43 -38
View File
@@ -234,13 +234,13 @@ import hashlib
import json import json
import os import os
import sys import sys
import uuid
from pathlib import Path from pathlib import Path
from bot_bottle import supervise as _sv
report_path = Path(sys.argv[1]) report_path = Path(sys.argv[1])
queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "")
slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "") slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
if not slug: if not queue_dir or not slug:
sys.exit(2) sys.exit(2)
try: try:
@@ -277,19 +277,31 @@ for i, finding in enumerate(raw, 1):
]) ])
payload = "\n".join(lines).rstrip() + "\n" payload = "\n".join(lines).rstrip() + "\n"
proposal = _sv.Proposal.new( proposal_id = str(uuid.uuid4())
bottle_slug=slug, proposal = {
tool=_sv.TOOL_GITLEAKS_ALLOW, "id": proposal_id,
proposed_file=payload, "bottle_slug": slug,
justification=( "tool": "gitleaks-allow",
"proposed_file": payload,
"justification": (
"git-gate found gitleaks findings hidden by # gitleaks:allow; " "git-gate found gitleaks findings hidden by # gitleaks:allow; "
"approve only for dummy test fixtures or confirmed false positives" "approve only for dummy test fixtures or confirmed false positives"
), ),
current_file_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(), "arrival_timestamp": datetime.datetime.now(
now=datetime.datetime.now(datetime.timezone.utc), datetime.timezone.utc
) ).isoformat(),
_sv.write_proposal(proposal) "current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
print(proposal.id) }
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)
PY PY
) )
rc=$? rc=$?
@@ -302,7 +314,8 @@ PY
return 1 return 1
fi fi
slug=${SUPERVISE_BOTTLE_SLUG:-} queue_dir=${SUPERVISE_QUEUE_DIR:-}
response_file="$queue_dir/${proposal_id}.response.json"
timeout=${SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS:-300} timeout=${SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS:-300}
case "$timeout" in case "$timeout" in
''|*[!0-9]*) ''|*[!0-9]*)
@@ -314,35 +327,26 @@ PY
echo "git-gate: approve with './cli.py supervise' to continue this push" >&2 echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
waited=0 waited=0
while [ "$waited" -lt "$timeout" ]; do while [ "$waited" -lt "$timeout" ]; do
status=$(python3 - "$slug" "$proposal_id" <<'PY' if [ -f "$response_file" ]; then
status=$(python3 - "$response_file" <<'PY'
import json
import sys import sys
from bot_bottle import supervise as _sv
slug = sys.argv[1]
try: try:
response = _sv.read_response(slug, sys.argv[2]) with open(sys.argv[1], encoding="utf-8") as f:
except FileNotFoundError: raw = json.load(f)
sys.exit(2) except (OSError, json.JSONDecodeError):
print(response.status) sys.exit(1)
status = raw.get("status")
if not isinstance(status, str):
sys.exit(1)
print(status)
PY PY
) ) || status=""
rc=$?
if [ "$rc" -eq 2 ]; then
status=""
elif [ "$rc" -ne 0 ]; then
status="invalid"
fi
if [ -n "$status" ]; then
case "$status" in case "$status" in
approved|modified) approved|modified)
python3 - "$slug" "$proposal_id" <<'PY' || true mkdir -p "$queue_dir/processed"
import sys 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
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 echo "git-gate: supervisor approved # gitleaks:allow for $ref" >&2
return 0 return 0
;; ;;
@@ -495,3 +499,4 @@ if ! git -C "$repo_dir" rev-parse --verify HEAD >/dev/null 2>&1; then
fi fi
exit 0 exit 0
""" """
+2 -7
View File
@@ -16,16 +16,11 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
from urllib.parse import urlsplit from urllib.parse import urlsplit
from .git_gate import GIT_GATE_TIMEOUT_SECS
DEFAULT_PORT = 9420 DEFAULT_PORT = 9420
# Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than
# imported: this module ships as a flat top-level sibling in the sidecar
# bundle image (see Dockerfile.sidecars), not as part of the bot_bottle
# package, so `bot_bottle.git_gate` and its dependency chain aren't
# available at runtime.
GIT_GATE_TIMEOUT_SECS = 15
# Bound memory use while still allowing ordinary git push packfiles. # Bound memory use while still allowing ordinary git push packfiles.
MAX_BODY_BYTES = 100 * 1024 * 1024 MAX_BODY_BYTES = 100 * 1024 * 1024
+1 -11
View File
@@ -8,7 +8,7 @@ from typing import cast
from .agent_provider import PROVIDER_TEMPLATES from .agent_provider import PROVIDER_TEMPLATES
from .manifest_util import ManifestError, as_json_object from .manifest_util import ManifestError, as_json_object
from .manifest_git import ManifestGitUser from .manifest_git import ManifestGitUser
from .manifest_schema import AGENT_MODEL_KEYS, is_valid_entity_name from .manifest_schema import AGENT_MODEL_KEYS
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -161,16 +161,6 @@ class ManifestAgent:
f"agent '{name}' skills[{i}] must be a string " f"agent '{name}' skills[{i}] must be a string "
f"(was {type(skill).__name__})" f"(was {type(skill).__name__})"
) )
# Skill names become host/guest path segments and are
# interpolated into provisioning shell commands, so they
# must fit the same kebab-case convention as bottle/agent
# filenames — rejecting anything that could break out of a
# path segment or inject shell metacharacters.
if not is_valid_entity_name(skill):
raise ManifestError(
f"agent '{name}' skills[{i}] {skill!r} is not a valid "
f"skill name; must match [a-z][a-z0-9-]*"
)
collected.append(skill) collected.append(skill)
skills = tuple(collected) skills = tuple(collected)
+1 -8
View File
@@ -33,20 +33,13 @@ AGENT_KEYS = (
AGENT_MODEL_KEYS = AGENT_KEYS | frozenset({"prompt"}) AGENT_MODEL_KEYS = AGENT_KEYS | frozenset({"prompt"})
def is_valid_entity_name(name: str) -> bool:
"""True if `name` fits the kebab-case `[a-z][a-z0-9-]*` convention
shared by bottle/agent filenames and skill names. Names that satisfy
this are also safe to interpolate into a host/guest path segment."""
return bool(_FILENAME_RX.match(name))
def entity_name_from_path(path: Path) -> str | None: def entity_name_from_path(path: Path) -> str | None:
"""Return the entity name implied by the filename, or None if the """Return the entity name implied by the filename, or None if the
filename does not fit the [a-z][a-z0-9-]* convention.""" filename does not fit the [a-z][a-z0-9-]* convention."""
if path.suffix != ".md": if path.suffix != ".md":
return None return None
stem = path.stem stem = path.stem
if not is_valid_entity_name(stem): if not _FILENAME_RX.match(stem):
return None return None
return stem return stem
-248
View File
@@ -1,248 +0,0 @@
"""SQLite-backed queue store for supervise proposals and responses (PRD 0013)."""
from __future__ import annotations
import os
import sqlite3
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .supervise import Proposal, Response
def _sv() -> object:
"""Lazy import of supervise to avoid a circular-import at module init time.
By the time any QueueStore method is called, both modules are fully loaded.
Mirrors our own module identity: when we are 'queue_store' (sidecar flat
context or tests that inject bot_bottle/ into sys.path) we use the flat
'supervise' module so that patches on supervise.bot_bottle_root propagate
correctly. When we are 'bot_bottle.queue_store' we use 'bot_bottle.supervise'."""
import sys
sv_name = "supervise" if __name__ == "queue_store" else "bot_bottle.supervise"
if sv_name in sys.modules:
return sys.modules[sv_name]
try:
import bot_bottle.supervise as _m
except ImportError:
import supervise as _m # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
return _m
def _proposal_from_row(row: sqlite3.Row) -> Proposal:
m = _sv()
return m.Proposal( # type: ignore[attr-defined]
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"],
)
def _response_from_row(row: sqlite3.Row) -> Response:
m = _sv()
return m.Response( # type: ignore[attr-defined]
proposal_id=row["proposal_id"],
status=row["status"],
notes=row["notes"],
final_file=row["final_file"],
)
def _host_db_path() -> Path:
return _sv().host_db_path() # type: ignore[attr-defined,no-any-return]
class QueueStore:
"""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:
self.db_path = 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()
self.db_path = Path(env_path) if env_path else _host_db_path()
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._init()
def write_proposal(self, proposal: Proposal) -> Path:
with self._connect() as conn:
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 _proposal_from_row(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 [_proposal_from_row(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 [_proposal_from_row(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 _response_from_row(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),
)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _init(self) -> None:
with self._connect() as conn:
conn.execute(
"""
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)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS 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)
)
"""
)
self._chmod()
def _chmod(self) -> None:
try:
self.db_path.chmod(0o600)
except OSError:
pass
__all__ = ["QueueStore"]
+203 -72
View File
@@ -9,14 +9,15 @@ calls when it needs an operator-reviewed egress change:
Each tool call: the agent passes the full proposed file plus a Each tool call: the agent passes the full proposed file plus a
justification text. The sidecar validates the proposal syntactically, justification text. The sidecar validates the proposal syntactically,
writes it to the host SQLite queue table, and holds the tool-call writes it to the host's per-bottle queue dir, and holds the tool-call
connection open. The operator's supervise TUI connection open. The operator's supervise TUI
(bot_bottle.cli.supervise) sees the proposal, accepts (bot_bottle.cli.supervise) sees the proposal, accepts
approve / modify / reject, and writes a response row. The sidecar sees approve / modify / reject, and writes a response file alongside the
the response and returns `{status, notes}` to the agent. proposal. The sidecar sees the response and returns `{status, notes}`
to the agent.
This module defines the host-side library: dataclasses for the queue This module defines the host-side library: dataclasses for the queue
record shapes, queue read/write helpers, the audit log writer, and the file shapes, queue read/write helpers, the audit log writer, and the
diff renderer. The in-container sidecar lives in diff renderer. The in-container sidecar lives in
bot_bottle/supervise_server.py; the supervise daemon's container bot_bottle/supervise_server.py; the supervise daemon's container
lifecycle is owned by the sidecar bundle (PRD 0024). lifecycle is owned by the sidecar bundle (PRD 0024).
@@ -33,6 +34,8 @@ from __future__ import annotations
import dataclasses import dataclasses
import difflib import difflib
import hashlib import hashlib
import json
import os
import time import time
import uuid import uuid
from abc import ABC from abc import ABC
@@ -83,9 +86,8 @@ STATUSES: tuple[str, ...] = (STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED)
# `routes edit <bottle>` verb writes entries with this action. # `routes edit <bottle>` verb writes entries with this action.
ACTION_OPERATOR_EDIT = "operator-edit" ACTION_OPERATOR_EDIT = "operator-edit"
DB_PATH_IN_CONTAINER = "/run/supervise/bot-bottle.db" QUEUE_DIR_IN_CONTAINER = "/run/supervise/queue"
DEFAULT_POLL_INTERVAL_SEC = 0.5 DEFAULT_POLL_INTERVAL_SEC = 0.5
HOST_DB_FILENAME = "bot-bottle.db"
# --- Paths ----------------------------------------------------------------- # --- Paths -----------------------------------------------------------------
@@ -95,6 +97,10 @@ def bot_bottle_root() -> Path:
return Path.home() / ".bot-bottle" return Path.home() / ".bot-bottle"
def queue_dir_for_slug(slug: str) -> Path:
return bot_bottle_root() / "queue" / slug
def audit_dir() -> Path: def audit_dir() -> Path:
return bot_bottle_root() / "audit" return bot_bottle_root() / "audit"
@@ -103,16 +109,14 @@ def audit_log_path(component: str, slug: str) -> Path:
return audit_dir() / f"{component}-{slug}.log" return audit_dir() / f"{component}-{slug}.log"
def host_db_path() -> Path:
return bot_bottle_root() / HOST_DB_FILENAME
# --- Dataclasses ----------------------------------------------------------- # --- Dataclasses -----------------------------------------------------------
@dataclass(frozen=True) @dataclass(frozen=True)
class Proposal: class Proposal:
"""One pending tool-call from the agent.""" """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 id: str
bottle_slug: str bottle_slug: str
@@ -166,7 +170,7 @@ class Proposal:
@dataclass(frozen=True) @dataclass(frozen=True)
class Response: class Response:
"""The operator's decision on a proposal. The TUI writes one of """The operator's decision on a proposal. The TUI writes one of
these to the queue table; the sidecar reads it and returns the these to the queue dir; the sidecar reads it and returns the
`{status, notes}` pair to the agent's tool call. `{status, notes}` pair to the agent's tool call.
`final_file` carries the file content the supervisor will `final_file` carries the file content the supervisor will
@@ -219,50 +223,90 @@ class AuditEntry:
return dataclasses.asdict(self) return dataclasses.asdict(self)
try:
from .queue_store import QueueStore
from .audit_store import AuditStore
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
# --- Queue I/O ------------------------------------------------------------- # --- Queue I/O -------------------------------------------------------------
def write_proposal(proposal: Proposal) -> Path: def _proposal_filename(proposal_id: str) -> str:
"""Persist `proposal` in the queue database, mode 0o600. 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.
Directory is created if missing.""" Directory is created if missing."""
return QueueStore(proposal.bottle_slug).write_proposal(proposal) 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
def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal: def read_proposal(queue_dir: Path, proposal_id: str) -> Proposal:
return QueueStore(bottle_slug).read_proposal(proposal_id) 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 list_pending_proposals(bottle_slug: str) -> list[Proposal]: def list_pending_proposals(queue_dir: Path) -> list[Proposal]:
"""All proposals for `bottle_slug` that do not yet have a matching """All proposals in `queue_dir` that do not yet have a matching
response. Sorted by `arrival_timestamp` so the operator response file. Sorted by `arrival_timestamp` so the operator
sees the queue FIFO.""" sees the queue FIFO."""
return QueueStore(bottle_slug).list_pending_proposals() 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
def list_all_pending_proposals() -> list[Proposal]: def write_response(queue_dir: Path, response: Response) -> Path:
"""All pending proposals across bottles, sorted FIFO.""" queue_dir.mkdir(parents=True, exist_ok=True)
return QueueStore("").list_all_pending_proposals() 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 write_response(bottle_slug: str, response: Response) -> Path: def read_response(queue_dir: Path, proposal_id: str) -> Response:
return QueueStore(bottle_slug).write_response(response) path = queue_dir / _response_filename(proposal_id)
with path.open() as f:
raw = json.load(f)
def read_response(bottle_slug: str, proposal_id: str) -> Response: if not isinstance(raw, dict):
return QueueStore(bottle_slug).read_response(proposal_id) raise ValueError(f"{path}: top-level must be an object")
return Response.from_dict(raw)
def wait_for_response( def wait_for_response(
bottle_slug: str, queue_dir: Path,
proposal_id: str, proposal_id: str,
*, *,
poll_interval: float = DEFAULT_POLL_INTERVAL_SEC, poll_interval: float = DEFAULT_POLL_INTERVAL_SEC,
@@ -273,35 +317,90 @@ def wait_for_response(
which the wait raises TimeoutError. None waits forever the which the wait raises TimeoutError. None waits forever the
natural shape, since the operator's response time is unbounded. natural shape, since the operator's response time is unbounded.
Polls SQLite so the implementation stays portable and stdlib-only.""" Polls the filesystem so the implementation stays portable and
store = QueueStore(bottle_slug) stdlib-only."""
path = queue_dir / _response_filename(proposal_id)
while True: while True:
try: if path.exists():
return store.read_response(proposal_id) try:
except FileNotFoundError: with path.open() as f:
pass raw = json.load(f)
except (OSError, json.JSONDecodeError):
raw = None
if isinstance(raw, dict):
try:
return Response.from_dict(raw)
except (KeyError, ValueError):
pass
if deadline is not None and time.monotonic() >= deadline: if deadline is not None and time.monotonic() >= deadline:
raise TimeoutError(f"no response for proposal {proposal_id!r}") raise TimeoutError(f"no response for proposal {proposal_id!r}")
time.sleep(poll_interval) time.sleep(poll_interval)
def archive_proposal(bottle_slug: str, proposal_id: str) -> None: def archive_proposal(queue_dir: Path, proposal_id: str) -> None:
"""Mark both proposal and response rows processed. """Move both proposal and response files to `<queue_dir>/processed/`.
Idempotent missing rows are silently skipped.""" Idempotent missing files are silently skipped."""
QueueStore(bottle_slug).archive_proposal(proposal_id) 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)
# --- Audit log ------------------------------------------------------------- # --- Audit log -------------------------------------------------------------
def write_audit_entry(entry: AuditEntry) -> Path: def write_audit_entry(entry: AuditEntry) -> Path:
"""Append `entry` to the host supervise audit table.""" """Append `entry` as one JSON-Lines record to the per-bottle
return AuditStore().write_audit_entry(entry) 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
def read_audit_entries(component: str, slug: str) -> list[AuditEntry]: def read_audit_entries(component: str, slug: str) -> list[AuditEntry]:
"""Load all audit entries for the given component+slug.""" """Load all audit entries for the given component+slug. Empty
return AuditStore().read_audit_entries(component, slug) 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
# --- Diff rendering -------------------------------------------------------- # --- Diff rendering --------------------------------------------------------
@@ -334,34 +433,35 @@ def sha256_hex(content: str) -> str:
class SupervisePlan: class SupervisePlan:
"""Output of Supervise.prepare; consumed by .start. """Output of Supervise.prepare; consumed by .start.
`db_path` is the host database bind-mounted into the sidecar at `queue_dir` is the host directory bind-mounted into the sidecar
/run/supervise/bot-bottle.db. `internal_network` is empty at at /run/supervise/queue. `internal_network` is empty at prepare
prepare time; the backend's launch step fills it via time; the backend's launch step fills it via dataclasses.replace
dataclasses.replace before calling .start.""" before calling .start."""
slug: str slug: str
db_path: Path queue_dir: Path
internal_network: str = "" internal_network: str = ""
class Supervise(ABC): class Supervise(ABC):
"""Per-bottle supervise sidecar. Encapsulates host-side database """Per-bottle supervise sidecar. Encapsulates the host-side
staging; the sidecar's start/stop lifecycle is backend-specific.""" prepare (queue dir staging); the sidecar's start/stop lifecycle
is backend-specific."""
def prepare( def prepare(
self, self,
slug: str, slug: str,
stage_dir: Path, stage_dir: Path,
) -> SupervisePlan: ) -> SupervisePlan:
"""Stage the host database. Returns the plan; `internal_network` """Stage the per-bottle queue dir on the host. Returns the
must be set by the launch step before .start runs.""" plan; `internal_network` must be set by the launch step before
.start runs."""
del stage_dir del stage_dir
db_path = host_db_path() queue_dir = queue_dir_for_slug(slug)
QueueStore(slug) queue_dir.mkdir(parents=True, exist_ok=True)
AuditStore(db_path)
return SupervisePlan( return SupervisePlan(
slug=slug, slug=slug,
db_path=db_path, queue_dir=queue_dir,
) )
# --- Helpers --------------------------------------------------------------- # --- Helpers ---------------------------------------------------------------
@@ -374,15 +474,47 @@ def _require_str(raw: dict[str, object], key: str) -> str:
return value 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__ = [ __all__ = [
"ACTION_OPERATOR_EDIT", "ACTION_OPERATOR_EDIT",
"AuditEntry", "AuditEntry",
"AuditStore",
"COMPONENT_FOR_TOOL", "COMPONENT_FOR_TOOL",
"DEFAULT_POLL_INTERVAL_SEC", "DEFAULT_POLL_INTERVAL_SEC",
"DB_PATH_IN_CONTAINER",
"Proposal", "Proposal",
"QueueStore", "QUEUE_DIR_IN_CONTAINER",
"Response", "Response",
"STATUSES", "STATUSES",
"STATUS_APPROVED", "STATUS_APPROVED",
@@ -404,9 +536,8 @@ __all__ = [
"audit_dir", "audit_dir",
"audit_log_path", "audit_log_path",
"bot_bottle_root", "bot_bottle_root",
"host_db_path",
"list_pending_proposals", "list_pending_proposals",
"list_all_pending_proposals", "queue_dir_for_slug",
"read_audit_entries", "read_audit_entries",
"read_proposal", "read_proposal",
"read_response", "read_response",
+17 -9
View File
@@ -7,13 +7,14 @@ config changes when stuck. The tools are `egress-allow`,
Each queued tool call: Each queued tool call:
1. Validates the proposed file syntactically. 1. Validates the proposed file syntactically.
2. Writes a Proposal to the host SQLite database. 2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
3. Blocks polling for a matching Response row. the host's ~/.bot-bottle/queue/<slug>/).
3. Blocks polling for a matching Response file.
4. Returns the operator's `{status, notes}` to the agent. 4. Returns the operator's `{status, notes}` to the agent.
The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
container creation by the backend's start step). SUPERVISE_DB_PATH container creation by the backend's start step). The queue dir comes
points at the bind-mounted host database. from SUPERVISE_QUEUE_DIR (default `/run/supervise/queue`).
Speaks MCP over HTTP+JSON-RPC. Methods handled: Speaks MCP over HTTP+JSON-RPC. Methods handled:
@@ -41,6 +42,7 @@ import typing
import urllib.error import urllib.error
import urllib.request import urllib.request
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path
try: try:
# Same-directory imports inside the bundle container; these files are # Same-directory imports inside the bundle container; these files are
@@ -275,6 +277,7 @@ def validate_proposed_file(tool: str, content: str) -> None:
@dataclass(frozen=True) @dataclass(frozen=True)
class ServerConfig: class ServerConfig:
bottle_slug: str bottle_slug: str
queue_dir: Path
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS
@@ -373,7 +376,7 @@ def handle_tools_call(
current_file_hash=_sv.sha256_hex(proposed_file), current_file_hash=_sv.sha256_hex(proposed_file),
) )
try: try:
_sv.write_proposal(proposal) _sv.write_proposal(config.queue_dir, proposal)
except OSError as e: except OSError as e:
raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e
sys.stderr.write( sys.stderr.write(
@@ -384,7 +387,7 @@ def handle_tools_call(
deadline = time.monotonic() + config.response_timeout_seconds deadline = time.monotonic() + config.response_timeout_seconds
try: try:
response = _sv.wait_for_response( response = _sv.wait_for_response(
config.bottle_slug, config.queue_dir,
proposal.id, proposal.id,
poll_interval=MIN_RESPONSE_POLL_INTERVAL_SECONDS, poll_interval=MIN_RESPONSE_POLL_INTERVAL_SECONDS,
deadline=deadline, deadline=deadline,
@@ -396,7 +399,7 @@ def handle_tools_call(
"isError": False, "isError": False,
} }
try: try:
_sv.archive_proposal(config.bottle_slug, proposal.id) _sv.archive_proposal(config.queue_dir, proposal.id)
except OSError as e: except OSError as e:
raise _RpcInternalError(f"failed to archive proposal: {e}") from e raise _RpcInternalError(f"failed to archive proposal: {e}") from e
@@ -536,7 +539,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = True allow_reuse_address = True
daemon_threads = True daemon_threads = True
config: ServerConfig = ServerConfig(bottle_slug="") config: ServerConfig = ServerConfig(bottle_slug="", queue_dir=Path())
# --- Entry point ----------------------------------------------------------- # --- Entry point -----------------------------------------------------------
@@ -545,18 +548,21 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
def serve( def serve(
*, *,
bottle_slug: str, bottle_slug: str,
queue_dir: Path,
port: int = _sv.SUPERVISE_PORT, port: int = _sv.SUPERVISE_PORT,
bind: str = "0.0.0.0", bind: str = "0.0.0.0",
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS, response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
) -> typing.NoReturn: ) -> typing.NoReturn:
queue_dir.mkdir(parents=True, exist_ok=True)
server = MCPServer((bind, port), MCPHandler) server = MCPServer((bind, port), MCPHandler)
server.config = ServerConfig( server.config = ServerConfig(
bottle_slug=bottle_slug, bottle_slug=bottle_slug,
queue_dir=queue_dir,
response_timeout_seconds=response_timeout_seconds, response_timeout_seconds=response_timeout_seconds,
) )
sys.stderr.write( sys.stderr.write(
f"supervise listening on {bind}:{port}; " f"supervise listening on {bind}:{port}; "
f"slug={bottle_slug!r}; " f"slug={bottle_slug!r}; queue={queue_dir}; "
f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type] f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type]
) )
sys.stderr.flush() sys.stderr.flush()
@@ -575,6 +581,7 @@ def main(argv: list[str]) -> int:
if not bottle_slug: if not bottle_slug:
sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n") sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
return 2 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))) port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
bind = os.environ.get("SUPERVISE_BIND", "0.0.0.0") bind = os.environ.get("SUPERVISE_BIND", "0.0.0.0")
try: try:
@@ -584,6 +591,7 @@ def main(argv: list[str]) -> int:
return 2 return 2
serve( serve(
bottle_slug=bottle_slug, bottle_slug=bottle_slug,
queue_dir=queue_dir,
port=port, port=port,
bind=bind, bind=bind,
response_timeout_seconds=response_timeout_seconds, response_timeout_seconds=response_timeout_seconds,
+247
View File
@@ -0,0 +1,247 @@
# PRD prd-new: Egress control plane — metering, budgets, and forced cutoff
- **Status:** Draft
- **Author:** didericis
- **Created:** 2026-06-25
- **Issue:** #251
## Summary
Add an **out-of-band egress enforcement & observability plane**: meter every
agent's token usage at the egress proxy, decrement budgets without the agent's
cooperation, and forcibly cut a bottle's egress when a budget is exhausted —
either automatically or on command from a host-level dashboard. The trigger
(usage threshold) and the action (route-drop / freeze / kill) both live in the
egress plane and run with no agent in the loop. This is distinct from the
supervise sidecar (PRD 0013), which is agent-initiated and therefore cannot
enforce a cost cutoff on a runaway agent. State (usage ledger, budgets, audit)
moves into a host-level SQLite database behind a thin repository API, the first
SQL store in an otherwise flat-file repo.
## Problem
bot-bottle can't currently do two things the cost-overrun case demands:
1. **Forced egress shutdown on limit.** When an agent crosses a token
threshold, kill its egress automatically — no human in the loop.
2. **Remote (host-level) management.** Drive agents from a single surface:
see usage, cut egress, stop bottles, to prevent cost overruns.
The existing supervise sidecar (PRD 0013) is **entirely agent-initiated**: every
action begins with the agent voluntarily calling an MCP tool and an operator
approving it. A runaway or expensive agent — exactly the cost-overrun case —
will never call `egress-block` on itself. Supervision is therefore a
**collaborative recovery** mechanism, not an **enforcement** mechanism; making
it mandatory (#249) would not deliver forced cost-cutoff.
The requirement forces a distinction the current design blurs:
- **Plane A — enforcement / observability (this PRD).** System → infrastructure.
Meter usage, cut egress on threshold or command, account for cost.
Out-of-band; independent of the agent. **Unconditional** — an enforcement
plane you can opt out of isn't enforcement.
- **Plane B — agent-facing recovery (the existing supervise sidecar).**
Agent → operator, approval-gated. Useful interactively; meaningless for a
headless agent with no operator watching its queue. Remains optional.
This PRD builds Plane A. It reframes the "always-on control" invariant of #249
as "the egress control plane is always present" — a more defensible property
than "every agent runs the agent-facing supervisor." Unsupervised
(headless/CI/ephemeral) agents stay first-class: still subject to the mandatory
meter + kill switch, they simply lack the agent-facing proposal tools they
couldn't use anyway.
## Goals / Success Criteria
- The egress proxy meters every request to a metered API host (e.g.
`api.anthropic.com`) and records authoritative token usage per bottle and per
agent provider, with no agent cooperation.
- A budget can be set at four scopes with deterministic precedence
(**agent → bottle → parent bottle → global host budget**); the
most-specific applicable budget governs.
- When usage crosses a budget, the bottle's configured **cutoff policy**
(`cutoff` | `freeze` | `kill`) fires automatically, executed host-side on the
egress plane — never via the supervise queue.
- An operator can, from a single **host-level TUI dashboard**, see live per-bottle
usage against budget and command a cutoff/stop on demand.
- Host budgets, default cutoff policy, and per-provider limits are declared in a
new host-level `~/.bot-bottle/settings.yml`, parseable by `yaml_subset.py`.
- All usage, budget state, and enforcement actions persist in a host-level
SQLite DB behind a thin repository API, so the store can later be swapped for
a cross-host cloud service.
## Non-goals
- **Remote control / cross-host control plane.** Web + mobile remote control,
cross-host budgets, and the authn/transport they require are explicitly
deferred. v1 is a **host-only TUI** with no remote surface.
- **Dollar-denominated budgets.** Budgets are token counts keyed by agent
provider, not currency. Price tables are out of scope.
- **Migrating existing flat-file state into SQLite.** Resume `metadata.json`,
transcripts, Dockerfile overrides, the supervise queue, and audit logs stay on
the filesystem. Only the *new* metering/budget/enforcement ledger is SQL.
- **Making the supervise sidecar (Plane B) mandatory.** Out of scope here; this
PRD is the answer to "what should be unconditional" (Plane A), leaving #249's
Plane-B question open.
- **Per-request hard pre-send blocking as the primary mechanism.** The gate is
budget-crossing detected at/after metering; a pre-flight estimator (below) is a
refinement, not the core enforcement path.
## Design
### Two measurements: gate vs. account
There are two distinct needs, and they want different signals:
- **Account (authoritative).** Decrement the real budget from the API
**response**, which already carries authoritative usage (Anthropic
`input_tokens` / `output_tokens`, OpenAI `usage`). The egress addon already
has a `response(flow)` hook (`bot_bottle/egress_addon.py:460`), so the real
number is available with no extra network call. **Caveat:** agent traffic is
mostly streaming SSE, so the response path must tail the stream for the final
usage event rather than parse a single JSON body — scoped explicitly as work.
- **Gate (estimate).** To block *before* sending, only the request is available,
so an estimator / provider `count_tokens` endpoint is the only option.
Calling `count_tokens` for accounting would be both less accurate *and* an extra
metered egress call per request, so accounting uses response `usage` and the
estimator is reserved for the optional pre-flight gate.
### `count_tokens` on agent providers
Add an abstract `count_tokens(request) -> int` to the `AgentProvider`
abstraction (`bot_bottle/agent_provider.py`):
- **Default** is a good-enough stdlib estimator. Prefer stdlib only; a small
pip dependency *for the sidecar* is acceptable for the fallback if stdlib
proves too inaccurate (this does not relax the package's stdlib-first stance —
it would be a sidecar-only dep, like the bundle already carries).
- **Built-in `claude`** uses Anthropic's token-counting endpoint;
**built-in `codex`** uses OpenAI's. These are exact for the gate but cost a
metered call, so they are gate-only; accounting still comes from the response.
### Budgets and precedence
Budgets are token counts keyed by **agent provider name** (the same names
bottles already use). Four scopes, most-specific wins:
```
agent → bottle → parent bottle → global (host)
```
The global host budget is the highest-priority feature to ship (the cross-host
control plane will eventually consume it); per-agent and per-bottle budgets
override it for finer control. A budget can also be supplied **at bottle
launch** (`--budget` or equivalent), overriding the settings.yml defaults for
that run. Enforcement evaluates the effective budget as the
nearest-defined scope at decrement time.
### `~/.bot-bottle/settings.yml`
New **host-level** settings file (the `~/.bot-bottle/` root, *not* the per-repo
`.bot-bottle/` — host budgets must not be committed per-repo). Parsed by
`yaml_subset.py`, so it must stay within that bounded subset (flat mappings,
scalars; no anchors, no multi-line block scalars). Shape:
```yaml
budget:
claude: 5000000 # token budget keyed by agent provider
codex: 2000000
shutdown: cutoff # default cutoff policy: cutoff | freeze | kill
```
### Forced cutoff and cutoff policy
On budget exhaustion (or an operator command), the configured per-bottle cutoff
policy fires. The three policies map onto primitives that already exist:
- **`cutoff`** (default) — drop the bottle's `routes.yaml` to empty and reload
(or isolate the bottle from the egress network); the agent/bottle keeps
running but can no longer reach metered hosts. This is the route-drop already
available on the egress plane (`bot_bottle/backend/egress_apply.py`).
- **`freeze`** — commit/snapshot state, then kill the agent/bottle; resumable
later via `bot_bottle/backend/freeze.py`.
- **`kill`** — tear the bottle down without saving state (backend teardown).
The trigger lives in the metering path and the action in the egress/backend
plane; **neither touches the supervise proposal queue** (design constraint from
#251).
### Host-level SQLite store
**Decision: introduce SQLite now, narrowly.**
- **The dependency objection doesn't apply.** `sqlite3` is in the Python stdlib,
so it does not break the AGENTS.md stdlib-first / no-runtime-pip stance — same
category as the hand-rolled `yaml_subset.py`, except the stdlib already ships
the whole engine.
- **It fits the problem.** A *global* token budget decremented concurrently by N
egress sidecars (today `~/.bot-bottle/` already has `state/`, `audit/`,
`queue/` written by parallel bottles) is a read-modify-write race. Over JSON
that means hand-rolled file locking; SQLite gives atomic transactions + WAL for
free. The per-agent/per-bottle precedence rollup plus "sum across all bottles"
is a `GROUP BY`, not an N-directory rescan.
- **It rehearses the cloud swap.** "Wrap operations in an API so we can swap to a
cloud service" maps directly onto a thin repository/DAO over SQLite → Postgres
later. A JSON-file store is a worse rehearsal than SQL.
**Costs (real but bounded):** a new paradigm in a flat-file repo needs a
`schema_version` table + idempotent startup migrations; SQLite serializes
writers, so WAL mode + `busy_timeout` are required (a non-issue at a handful of
bottles); test fixtures need temp DBs.
**Scope of the store:** one DB at `~/.bot-bottle/bot-bottle.db` behind a thin
repository API. Only the **new** metering/budget/enforcement-audit ledger lives
there. Existing per-bottle blobs (resume `metadata.json`, transcripts,
Dockerfile overrides, supervise queue) stay on the filesystem — migrating them
now is churn for no benefit and they lack the concurrency/aggregation problem.
### Host-level controller + dashboard
A single **host-level controller** owns the meter, budget evaluation, and the
cutoff actions across all bottles (cf. `bot_bottle/cli/supervise.py`'s
cross-bottle view), rather than a per-bottle daemon. v1 ships one host-level
**TUI dashboard** that reads live usage-vs-budget from the SQLite store and
offers on-demand cutoff/stop. The existing supervisor UI should eventually fold
into this same dashboard; this PRD lays the host-level surface it will move to.
## Implementation chunks
Ordered, individually mergeable:
1. **SQLite repository foundation.** `~/.bot-bottle/bot-bottle.db`, schema +
`schema_version` migrations, WAL + `busy_timeout`, thin repository API,
temp-DB test fixtures. No behavior wired yet.
2. **Metering at the egress proxy.** Parse authoritative response `usage`
(including SSE final-usage tailing) in the egress addon `response` hook;
write per-bottle / per-provider usage rows to the ledger.
3. **`settings.yml` + budget model.** Host-level `~/.bot-bottle/settings.yml`
parsed by `yaml_subset.py`; budget precedence (agent → bottle → parent →
global) and the `--budget` launch flag.
4. **Forced cutoff + cutoff policy.** Wire the threshold trigger to the
`cutoff` / `freeze` / `kill` primitives on the egress/backend plane; record
enforcement actions to the audit ledger.
5. **Host-level TUI dashboard.** Live usage-vs-budget view + on-demand
cutoff/stop, reading the store.
6. **`count_tokens` pre-flight gate (optional refinement).** Abstract method +
stdlib estimator default; Anthropic/OpenAI endpoints for built-in
claude/codex; optional pre-send block.
## Open questions
- **SSE usage tailing robustness.** Buffering streamed responses to extract the
final usage event without breaking the agent's own stream consumption — how
much of the body must the addon hold, and what's the failure mode if the
stream is interrupted mid-flight?
- **Crossing mid-request.** A single response can push usage past budget only
*after* it's already been delivered. Is post-hoc cutoff (next request blocked)
sufficient, or is a pre-flight estimator gate (chunk 6) required for v1?
- **Provider name ↔ metered host mapping.** How does the proxy attribute a
flow to an agent-provider budget key — by destination host, by bottle
identity, or both?
- **Parent-bottle budget semantics.** For `bottle extends` (PRD 0025 / 0065)
chains, does "parent bottle" mean the manifest parent, the launching bottle,
or the full ancestry summed?
- **Dashboard ↔ controller transport (even host-only).** In-process, a local
socket, or polling the SQLite store directly? Picks the seam the future remote
control plane will extend.
-134
View File
@@ -1,134 +0,0 @@
# 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. 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.
- Introducing a third-party ORM or migration framework.
## 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.
-188
View File
@@ -1,188 +0,0 @@
"""Unit: `cli.py start --headless` non-interactive launch path.
Headless is the keystone for orchestrators, CI, and webhook
dispatch: agent/bottles/label come from flags + manifest defaults, no
TUI selectors fire, and the preflight y/N is auto-confirmed
(`assume_yes=True`). All actual launch work is stubbed so no container
is created.
"""
from __future__ import annotations
import os
import unittest
from unittest.mock import MagicMock, patch
import bot_bottle.cli.start as start_mod
import bot_bottle.cli.tui as tui_mod
from bot_bottle.backend import ActiveAgent
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestError
def _make_manifest(
agent_names: list[str],
bottle_names: list[str] | None = None,
agent_bottle: str = "",
):
manifest = MagicMock()
manifest.agents = {name: MagicMock(bottle=agent_bottle) for name in agent_names}
manifest.all_agent_names = sorted(agent_names)
manifest.all_bottle_names = sorted(bottle_names or [])
manifest.home_md = None # eager mode so _peek_agent_bottle uses agents dict
manifest.require_agent = MagicMock(return_value=None)
return manifest
def _active_agent(slug: str) -> ActiveAgent:
return ActiveAgent(
backend_name="docker",
slug=slug,
agent_name="demo",
started_at="2026-01-01T00:00:00+00:00",
services=(),
)
class TestCmdStartHeadless(unittest.TestCase):
"""Drive `cmd_start --headless` with launch + TUI stubbed out."""
def setUp(self):
self._manifest = _make_manifest(
["researcher", "implementer"], ["claude", "dev"], agent_bottle="claude"
)
patch(
"bot_bottle.cli.start.ManifestIndex.resolve",
return_value=self._manifest,
).start()
self._launch_mock = patch(
"bot_bottle.cli.start._launch_bottle", return_value=0
).start()
# No bottles running by default → no label collision.
patch(
"bot_bottle.cli.start.enumerate_active_agents", return_value=[]
).start()
# If any TUI picker fires in headless mode, that's a bug.
self._agent_picker = patch.object(tui_mod, "filter_select").start()
self._bottle_picker = patch.object(tui_mod, "filter_multiselect").start()
self._modal = patch.object(tui_mod, "name_color_modal").start()
patch.dict(os.environ, {}, clear=False).start()
os.environ.pop("BOT_BOTTLE_BACKEND", None)
self.addCleanup(patch.stopall)
def _spec(self):
self._launch_mock.assert_called_once()
return self._launch_mock.call_args[0][0]
# -- no TUI in headless --------------------------------------------
def test_headless_fires_no_pickers(self):
rc = start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual(0, rc)
self._agent_picker.assert_not_called()
self._bottle_picker.assert_not_called()
self._modal.assert_not_called()
def test_headless_assume_yes_forwarded(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertTrue(self._launch_mock.call_args[1]["assume_yes"])
# -- prompt --------------------------------------------------------
def test_headless_without_prompt_dies(self):
with self.assertRaises(Die):
start_mod.cmd_start(["--headless", "researcher", "--bottle", "claude"])
self._launch_mock.assert_not_called()
def test_headless_prompt_forwarded_to_launch(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude",
"--prompt", "Implement issue #42"]
)
self.assertEqual(
"Implement issue #42",
self._launch_mock.call_args[1]["headless_prompt_text"],
)
# -- bottle resolution ---------------------------------------------
def test_explicit_bottles_forwarded_in_order(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "dev", "--bottle", "claude",
"--prompt", "Do it"]
)
self.assertEqual(("dev", "claude"), self._spec().bottle_names)
def test_omitted_bottle_falls_back_to_agent_default(self):
start_mod.cmd_start(["--headless", "implementer", "--prompt", "Do it"])
self.assertEqual(("claude",), self._spec().bottle_names)
def test_no_bottle_and_no_default_dies(self):
manifest = _make_manifest(["researcher"], ["claude"], agent_bottle="")
with patch(
"bot_bottle.cli.start.ManifestIndex.resolve", return_value=manifest
):
with self.assertRaises(Die):
start_mod.cmd_start(
["--headless", "researcher", "--prompt", "Do it"]
)
self._launch_mock.assert_not_called()
# -- agent resolution ----------------------------------------------
def test_missing_agent_name_dies(self):
with self.assertRaises(Die):
start_mod.cmd_start(["--headless"])
self._launch_mock.assert_not_called()
def test_unknown_agent_raises_manifest_error(self):
self._manifest.require_agent.side_effect = ManifestError("agent 'x' not defined")
with self.assertRaises(ManifestError):
start_mod.cmd_start(
["--headless", "x", "--bottle", "claude", "--prompt", "Do it"]
)
self._launch_mock.assert_not_called()
# -- label / color -------------------------------------------------
def test_label_defaults_to_agent_name(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual("researcher", self._spec().label)
def test_explicit_label_and_color_forwarded(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude",
"--label", "nightly", "--color", "green", "--prompt", "Do it"]
)
spec = self._spec()
self.assertEqual("nightly", spec.label)
self.assertEqual("green", spec.color)
def test_label_collision_uniquifies(self):
with patch(
"bot_bottle.cli.start.enumerate_active_agents",
return_value=[_active_agent("researcher")],
):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual("researcher-2", self._spec().label)
# -- backend wiring ------------------------------------------------
def test_backend_flag_forwarded(self):
start_mod.cmd_start(
["--headless", "--backend=docker", "researcher", "--bottle", "claude",
"--prompt", "Do it"]
)
self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"])
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -107,7 +107,7 @@ def _egress_plan(
def _supervise_plan() -> SupervisePlan: def _supervise_plan() -> SupervisePlan:
return SupervisePlan( return SupervisePlan(
slug=SLUG, slug=SLUG,
db_path=STATE / "bot-bottle.db", queue_dir=STATE / "supervise" / "queue",
internal_network=f"bot-bottle-net-{SLUG}", internal_network=f"bot-bottle-net-{SLUG}",
) )
@@ -392,7 +392,7 @@ class TestSidecarBundleShape(unittest.TestCase):
sc = self._render(supervise=True)["services"]["sidecars"] sc = self._render(supervise=True)["services"]["sidecars"]
env_strings = sc["environment"] env_strings = sc["environment"]
self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings) self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings)
self.assertIn("SUPERVISE_DB_PATH=/run/supervise/bot-bottle.db", env_strings) self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings)) self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))
def test_volumes_always_includes_egress_ca(self): def test_volumes_always_includes_egress_ca(self):
@@ -408,7 +408,8 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertIn("/etc/egress", targets) self.assertIn("/etc/egress", targets)
self.assertIn("/git-gate-entrypoint.sh", targets) self.assertIn("/git-gate-entrypoint.sh", targets)
self.assertIn("/git-gate/creds/upstream-known_hosts", targets) self.assertIn("/git-gate/creds/upstream-known_hosts", targets)
self.assertIn("/run/supervise/bot-bottle.db", targets) self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
for t in targets))
def test_extra_hosts_omitted_for_git_upstreams(self): def test_extra_hosts_omitted_for_git_upstreams(self):
sc = self._render(with_git=True)["services"]["sidecars"] sc = self._render(with_git=True)["services"]["sidecars"]
+1 -10
View File
@@ -74,7 +74,7 @@ def _plan(
if supervise: if supervise:
supervise_plan = SupervisePlan( supervise_plan = SupervisePlan(
slug="demo-abc12", slug="demo-abc12",
db_path=Path("/tmp/bot-bottle.db"), queue_dir=Path("/tmp/queue"),
) )
return DockerBottlePlan( return DockerBottlePlan(
spec=spec, spec=spec,
@@ -343,14 +343,5 @@ class TestClaudeSuperviseMcp(unittest.TestCase):
) )
class TestClaudeHeadlessPrompt(unittest.TestCase):
def test_returns_p_flag_and_prompt(self):
self.assertEqual(["-p", "Do the task"], ClaudeAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual(["-p", text], ClaudeAgentProvider().headless_prompt(text))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+1 -10
View File
@@ -77,7 +77,7 @@ def _plan(
if supervise: if supervise:
supervise_plan = SupervisePlan( supervise_plan = SupervisePlan(
slug="demo-abc12", slug="demo-abc12",
db_path=Path("/tmp/bot-bottle.db"), queue_dir=Path("/tmp/queue"),
) )
return DockerBottlePlan( return DockerBottlePlan(
spec=spec, spec=spec,
@@ -314,14 +314,5 @@ class TestCodexSuperviseMcp(unittest.TestCase):
) )
class TestCodexHeadlessPrompt(unittest.TestCase):
def test_returns_prompt_as_positional_arg(self):
self.assertEqual(["Do the task"], CodexAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual([text], CodexAgentProvider().headless_prompt(text))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
-9
View File
@@ -223,14 +223,5 @@ class TestPiDockerfile(unittest.TestCase):
self.assertIn("chmod 1777 /tmp /var/tmp", dockerfile) self.assertIn("chmod 1777 /tmp /var/tmp", dockerfile)
class TestPiHeadlessPrompt(unittest.TestCase):
def test_returns_p_flag_and_prompt(self):
self.assertEqual(["-p", "Do the task"], PiAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual(["-p", text], PiAgentProvider().headless_prompt(text))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
@@ -38,7 +38,6 @@ class _Provider(AgentProvider):
def provision_prompt(self, plan, bottle): ... # type: ignore[override] def provision_prompt(self, plan, bottle): ... # type: ignore[override]
def provision(self, plan, bottle): ... # type: ignore[override] def provision(self, plan, bottle): ... # type: ignore[override]
def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override] def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override]
def headless_prompt(self, prompt): return [] # type: ignore[override]
_PROVIDER = _Provider() _PROVIDER = _Provider()
@@ -47,6 +47,7 @@ def _addon() -> EgressAddon:
a: EgressAddon = EgressAddon.__new__(EgressAddon) a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = Config(routes=(), log=LOG_FULL) a.config = Config(routes=(), log=LOG_FULL)
a.safe_tokens = set() a.safe_tokens = set()
a._supervise_queue_dir = ""
a._supervise_slug = "" a._supervise_slug = ""
a._token_allow_timeout = 300.0 a._token_allow_timeout = 300.0
return a return a
+6 -3
View File
@@ -212,6 +212,7 @@ def _addon(config: Config) -> EgressAddon:
a: EgressAddon = EgressAddon.__new__(EgressAddon) a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = config a.config = config
a.safe_tokens = set() a.safe_tokens = set()
a._supervise_queue_dir = ""
a._supervise_slug = "" a._supervise_slug = ""
a._token_allow_timeout = 300.0 a._token_allow_timeout = 300.0
a.routes_path = "/nonexistent/routes.yaml" a.routes_path = "/nonexistent/routes.yaml"
@@ -385,10 +386,10 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
def _sha256_hex(_payload: Any) -> str: def _sha256_hex(_payload: Any) -> str:
return "hash" return "hash"
def _noop(*_args: Any) -> None: def _noop(_a: Any, _b: Any) -> None:
return None return None
def _read_response(_slug: Any, _pid: Any) -> Any: def _read_response(_qd: Any, _pid: Any) -> Any:
if response_status is None: if response_status is None:
raise OSError("not written yet") # forces poll -> timeout raise OSError("not written yet") # forces poll -> timeout
return types.SimpleNamespace(status=response_status) return types.SimpleNamespace(status=response_status)
@@ -408,6 +409,7 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
class TestSuperviseBranch(unittest.TestCase): class TestSuperviseBranch(unittest.TestCase):
def _supervised_addon(self) -> EgressAddon: def _supervised_addon(self) -> EgressAddon:
addon = _addon(Config(routes=(Route(host="api.example.com"),))) addon = _addon(Config(routes=(Route(host="api.example.com"),)))
addon._supervise_queue_dir = "/tmp/egress-queue"
addon._supervise_slug = "test-bottle" addon._supervise_slug = "test-bottle"
addon._token_allow_timeout = 0.05 addon._token_allow_timeout = 0.05
return addon return addon
@@ -630,13 +632,14 @@ class TestRedactSurfaces(unittest.TestCase):
class TestSuperviseWriteFailure(unittest.TestCase): class TestSuperviseWriteFailure(unittest.TestCase):
def test_write_proposal_oserror_blocks(self) -> None: def test_write_proposal_oserror_blocks(self) -> None:
addon = _addon(Config(routes=(Route(host="api.example.com"),))) addon = _addon(Config(routes=(Route(host="api.example.com"),)))
addon._supervise_queue_dir = "/tmp/egress-queue"
addon._supervise_slug = "test-bottle" addon._supervise_slug = "test-bottle"
addon._token_allow_timeout = 0.05 addon._token_allow_timeout = 0.05
flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")) flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}"))
fake = _fake_sv("approved") fake = _fake_sv("approved")
def _raise(_p: Any) -> None: def _raise(_qd: Any, _p: Any) -> None:
raise OSError("disk full") raise OSError("disk full")
fake.write_proposal = _raise fake.write_proposal = _raise
+2 -25
View File
@@ -14,7 +14,6 @@ from bot_bottle.git_gate import (
git_gate_render_access_hook, git_gate_render_access_hook,
git_gate_render_entrypoint, git_gate_render_entrypoint,
git_gate_render_hook, git_gate_render_hook,
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys, revoke_git_gate_provisioned_keys,
_resolve_identity_file, _resolve_identity_file,
git_gate_upstreams_for_bottle, git_gate_upstreams_for_bottle,
@@ -210,9 +209,8 @@ class TestHookRender(unittest.TestCase):
# the suppressed findings for human approval. # the suppressed findings for human approval.
self.assertIn("--ignore-gitleaks-allow", hook) self.assertIn("--ignore-gitleaks-allow", hook)
self.assertIn("--report-format=json", hook) self.assertIn("--report-format=json", hook)
self.assertIn("tool=_sv.TOOL_GITLEAKS_ALLOW", hook) self.assertIn('"tool": "gitleaks-allow"', hook)
self.assertIn("_sv.write_proposal", hook) self.assertIn("SUPERVISE_QUEUE_DIR", hook)
self.assertIn("_sv.read_response", hook)
self.assertIn("SUPERVISE_BOTTLE_SLUG", hook) self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
self.assertIn("supervisor approved # gitleaks:allow", hook) self.assertIn("supervisor approved # gitleaks:allow", hook)
self.assertIn("supervisor rejected # gitleaks:allow", hook) self.assertIn("supervisor rejected # gitleaks:allow", hook)
@@ -373,27 +371,6 @@ class TestDynamicKeyProvisioning(unittest.TestCase):
self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage)) self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
mock_provision.assert_called_once() mock_provision.assert_called_once()
def test_prepare_defers_gitea_key_provisioning(self):
bottle = self._gitea_manifest().bottles["dev"]
with patch("bot_bottle.git_gate_provision._provision_dynamic_key") as mock_provision:
plan = _StubGate().prepare(bottle, "demo", self.stage)
mock_provision.assert_not_called()
self.assertEqual("", plan.upstreams[0].identity_file)
def test_launch_time_helper_provisions_gitea_keys(self):
bottle = self._gitea_manifest().bottles["dev"]
plan = _StubGate().prepare(bottle, "demo", self.stage)
with patch(
"bot_bottle.git_gate_provision._provision_dynamic_key",
return_value="/tmp/provisioned-key",
) as mock_provision:
updated = provision_git_gate_dynamic_keys(bottle, plan, self.stage)
mock_provision.assert_called_once_with(bottle.git[0], "demo", self.stage)
self.assertEqual("/tmp/provisioned-key", updated.upstreams[0].identity_file)
def test_revoke_skips_non_gitea_and_missing_id_file(self): def test_revoke_skips_non_gitea_and_missing_id_file(self):
revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage) revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage)
+2 -4
View File
@@ -71,9 +71,7 @@ def _plan(
else: else:
git_gate_plan = SimpleNamespace(upstreams=()) git_gate_plan = SimpleNamespace(upstreams=())
supervise_plan = ( supervise_plan = (
SimpleNamespace( SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
db_path=Path("/state/bot-bottle.db"),
)
if supervise else None if supervise else None
) )
agent_provision = SimpleNamespace( agent_provision = SimpleNamespace(
@@ -139,7 +137,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
argv, argv,
) )
self.assertIn( self.assertIn(
"type=bind,source=/state/bot-bottle.db,target=/run/supervise/bot-bottle.db", "type=bind,source=/state/supervise/queue,target=/run/supervise/queue",
argv, argv,
) )
-16
View File
@@ -165,22 +165,6 @@ class TestAgentValidation(unittest.TestCase):
with self.assertRaises(ManifestError): with self.assertRaises(ManifestError):
ManifestAgent.from_dict("a", {"skills": [5]}, set()) ManifestAgent.from_dict("a", {"skills": [5]}, set())
def test_skill_name_rejects_shell_metacharacters(self) -> None:
# Skill names become host/guest path segments interpolated into
# provisioning shell commands; anything outside kebab-case is
# rejected at load so it can never reach a `bottle.exec` string.
for bad in ("foo; rm -rf /", "../escape", "foo bar", "Foo", "-leading"):
with self.assertRaises(ManifestError):
ManifestAgent.from_dict("a", {"skills": [bad]}, set())
def test_skill_name_accepts_kebab_case(self) -> None:
agent = ManifestAgent.from_dict(
"a", {"skills": ["init-entry", "quality-eval", "skill0"]}, set()
)
self.assertEqual(
agent.skills, ("init-entry", "quality-eval", "skill0")
)
def test_prompt_not_string(self) -> None: def test_prompt_not_string(self) -> None:
with self.assertRaises(ManifestError): with self.assertRaises(ManifestError):
ManifestAgent.from_dict("a", {"prompt": 5}, set()) ManifestAgent.from_dict("a", {"prompt": 5}, set())
+1 -2
View File
@@ -49,7 +49,6 @@ class _Provider(AgentProvider):
def provision_prompt(self, plan, bottle): ... # type: ignore[override] def provision_prompt(self, plan, bottle): ... # type: ignore[override]
def provision(self, plan, bottle): ... # type: ignore[override] def provision(self, plan, bottle): ... # type: ignore[override]
def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override] def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override]
def headless_prompt(self, prompt): return [] # type: ignore[override]
_PROVIDER = _Provider() _PROVIDER = _Provider()
@@ -130,7 +129,7 @@ def _plan(
if supervise: if supervise:
supervise_plan = SupervisePlan( supervise_plan = SupervisePlan(
slug="demo-abc12", slug="demo-abc12",
db_path=Path("/tmp/bot-bottle.db"), queue_dir=Path("/tmp/queue"),
) )
return SmolmachinesBottlePlan( return SmolmachinesBottlePlan(
spec=spec, spec=spec,
+36 -48
View File
@@ -1,5 +1,6 @@
"""Unit: supervise queue + audit log + diff helpers (PRD 0013).""" """Unit: supervise queue + audit log + diff helpers (PRD 0013)."""
import json
import tempfile import tempfile
import threading import threading
import time import time
@@ -18,7 +19,7 @@ from bot_bottle.supervise import (
TOOL_EGRESS_ALLOW, TOOL_EGRESS_ALLOW,
TOOL_GITLEAKS_ALLOW, TOOL_GITLEAKS_ALLOW,
archive_proposal, archive_proposal,
host_db_path, audit_log_path,
list_pending_proposals, list_pending_proposals,
read_audit_entries, read_audit_entries,
read_proposal, read_proposal,
@@ -111,44 +112,32 @@ class TestResponseRoundtrip(unittest.TestCase):
class TestQueueIO(unittest.TestCase): class TestQueueIO(unittest.TestCase):
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-test.") self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-test.")
self._home_patch = self._patch_home(Path(self._tmp.name)) self.queue_dir = Path(self._tmp.name)
self.slug = "dev"
def tearDown(self): def tearDown(self):
self._home_patch()
self._tmp.cleanup() self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original = supervise.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)
def test_write_and_read_proposal(self): def test_write_and_read_proposal(self):
p = _proposal() p = _proposal()
path = write_proposal(p) path = write_proposal(self.queue_dir, p)
self.assertTrue(path.exists()) self.assertTrue(path.exists())
self.assertEqual(host_db_path(), path)
self.assertEqual(0o600, path.stat().st_mode & 0o777) self.assertEqual(0o600, path.stat().st_mode & 0o777)
loaded = read_proposal(self.slug, p.id) loaded = read_proposal(self.queue_dir, p.id)
self.assertEqual(p, loaded) self.assertEqual(p, loaded)
def test_list_pending_excludes_responded(self): def test_list_pending_excludes_responded(self):
a = _proposal(justification="first") a = _proposal(justification="first")
b = _proposal(justification="second") b = _proposal(justification="second")
write_proposal(a) write_proposal(self.queue_dir, a)
write_proposal(b) write_proposal(self.queue_dir, b)
write_response(self.slug, Response( write_response(self.queue_dir, Response(
proposal_id=a.id, status=STATUS_APPROVED, notes="", proposal_id=a.id, status=STATUS_APPROVED, notes="",
)) ))
pending = list_pending_proposals(self.slug) pending = list_pending_proposals(self.queue_dir)
self.assertEqual([b.id], [p.id for p in pending]) self.assertEqual([b.id], [p.id for p in pending])
def test_list_pending_returns_empty_for_missing_slug(self): def test_list_pending_returns_empty_for_missing_dir(self):
self.assertEqual([], list_pending_proposals("nope")) self.assertEqual([], list_pending_proposals(self.queue_dir / "nope"))
def test_list_pending_sorted_by_arrival(self): def test_list_pending_sorted_by_arrival(self):
# Fabricate two with explicit timestamps. # Fabricate two with explicit timestamps.
@@ -165,30 +154,30 @@ class TestQueueIO(unittest.TestCase):
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
) )
# Write in reverse order. # Write in reverse order.
write_proposal(b) write_proposal(self.queue_dir, b)
write_proposal(a) write_proposal(self.queue_dir, a)
ordered = list_pending_proposals(self.slug) ordered = list_pending_proposals(self.queue_dir)
self.assertEqual([a.id, b.id], [p.id for p in ordered]) self.assertEqual([a.id, b.id], [p.id for p in ordered])
def test_write_and_read_response(self): def test_write_and_read_response(self):
r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no") r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no")
write_response(self.slug, r) write_response(self.queue_dir, r)
self.assertEqual(r, read_response(self.slug, "xyz")) self.assertEqual(r, read_response(self.queue_dir, "xyz"))
def test_wait_for_response_returns_when_file_appears(self): def test_wait_for_response_returns_when_file_appears(self):
p = _proposal() p = _proposal()
write_proposal(p) write_proposal(self.queue_dir, p)
def write_after_delay(): def write_after_delay():
time.sleep(0.05) time.sleep(0.05)
write_response(self.slug, Response( write_response(self.queue_dir, Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="ok", proposal_id=p.id, status=STATUS_APPROVED, notes="ok",
)) ))
t = threading.Thread(target=write_after_delay) t = threading.Thread(target=write_after_delay)
t.start() t.start()
try: try:
r = wait_for_response(self.slug, p.id, poll_interval=0.01) r = wait_for_response(self.queue_dir, p.id, poll_interval=0.01)
finally: finally:
t.join() t.join()
self.assertEqual(STATUS_APPROVED, r.status) self.assertEqual(STATUS_APPROVED, r.status)
@@ -198,24 +187,25 @@ class TestQueueIO(unittest.TestCase):
deadline = time.monotonic() + 0.05 deadline = time.monotonic() + 0.05
with self.assertRaises(TimeoutError): with self.assertRaises(TimeoutError):
wait_for_response( wait_for_response(
self.slug, "never", self.queue_dir, "never",
poll_interval=0.01, deadline=deadline, poll_interval=0.01, deadline=deadline,
) )
def test_archive_proposal_hides_rows(self): def test_archive_proposal_moves_both_files(self):
p = _proposal() p = _proposal()
write_proposal(p) write_proposal(self.queue_dir, p)
write_response(self.slug, Response( write_response(self.queue_dir, Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="", proposal_id=p.id, status=STATUS_APPROVED, notes="",
)) ))
archive_proposal(self.slug, p.id) archive_proposal(self.queue_dir, p.id)
self.assertEqual([], list_pending_proposals(self.slug)) self.assertFalse((self.queue_dir / f"{p.id}.proposal.json").exists())
with self.assertRaises(FileNotFoundError): self.assertFalse((self.queue_dir / f"{p.id}.response.json").exists())
read_response(self.slug, p.id) self.assertTrue((self.queue_dir / "processed" / f"{p.id}.proposal.json").exists())
self.assertTrue((self.queue_dir / "processed" / f"{p.id}.response.json").exists())
def test_archive_is_idempotent_on_missing_files(self): def test_archive_is_idempotent_on_missing_files(self):
# Should not raise. # Should not raise.
archive_proposal(self.slug, "nope") archive_proposal(self.queue_dir, "nope")
class TestAuditLog(unittest.TestCase): class TestAuditLog(unittest.TestCase):
@@ -247,7 +237,6 @@ class TestAuditLog(unittest.TestCase):
diff="--- before\n+++ after\n", diff="--- before\n+++ after\n",
) )
path = write_audit_entry(e) path = write_audit_entry(e)
self.assertEqual(host_db_path(), path)
self.assertEqual(0o600, path.stat().st_mode & 0o777) self.assertEqual(0o600, path.stat().st_mode & 0o777)
loaded = read_audit_entries("cred-proxy", "dev") loaded = read_audit_entries("cred-proxy", "dev")
self.assertEqual([e], loaded) self.assertEqual([e], loaded)
@@ -263,13 +252,12 @@ class TestAuditLog(unittest.TestCase):
justification="", justification="",
diff="", diff="",
)) ))
entries = read_audit_entries("egress", "dev") path = audit_log_path("egress", "dev")
self.assertEqual(3, len(entries)) with path.open() as f:
self.assertEqual( lines = [line for line in f if line.strip()]
["2026-05-25T12:00:00+00:00", "2026-05-25T12:00:01+00:00", self.assertEqual(3, len(lines))
"2026-05-25T12:00:02+00:00"], for line in lines:
[entry.timestamp for entry in entries], self.assertTrue(json.loads(line)) # each line is valid JSON
)
def test_separate_logs_per_component_slug(self): def test_separate_logs_per_component_slug(self):
write_audit_entry(AuditEntry( write_audit_entry(AuditEntry(
@@ -391,7 +379,7 @@ class TestSupervisePrepare(unittest.TestCase):
def test_prepare_creates_queue(self): def test_prepare_creates_queue(self):
plan = _StubSupervise().prepare("dev", self.stage_dir) plan = _StubSupervise().prepare("dev", self.stage_dir)
self.assertTrue(plan.db_path.is_file()) self.assertTrue(plan.queue_dir.is_dir())
self.assertEqual("dev", plan.slug) self.assertEqual("dev", plan.slug)
self.assertEqual("", plan.internal_network) self.assertEqual("", plan.internal_network)
+27 -15
View File
@@ -77,7 +77,9 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
def test_walks_all_slug_subdirs(self): def test_walks_all_slug_subdirs(self):
for slug in ("dev", "api"): for slug in ("dev", "api"):
supervise.write_proposal(_proposal(slug=slug)) qdir = supervise.queue_dir_for_slug(slug)
qdir.mkdir(parents=True)
supervise.write_proposal(qdir, _proposal(slug=slug))
pending = supervise_cli.discover_pending() pending = supervise_cli.discover_pending()
self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending}) self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})
@@ -95,14 +97,18 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
) )
for p in (late, early): for p in (late, early):
supervise.write_proposal(p) qdir = supervise.queue_dir_for_slug(p.bottle_slug)
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
pending = supervise_cli.discover_pending() pending = supervise_cli.discover_pending()
self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending]) self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending])
def test_excludes_already_responded(self): def test_excludes_already_responded(self):
p = _proposal() p = _proposal()
supervise.write_proposal(p) qdir = supervise.queue_dir_for_slug("dev")
supervise.write_response("dev", supervise.Response( qdir.mkdir(parents=True)
supervise.write_proposal(qdir, p)
supervise.write_response(qdir, supervise.Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="", proposal_id=p.id, status=STATUS_APPROVED, notes="",
)) ))
self.assertEqual([], supervise_cli.discover_pending()) self.assertEqual([], supervise_cli.discover_pending())
@@ -117,8 +123,10 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW): def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW):
p = _proposal(tool=tool) p = _proposal(tool=tool)
supervise.write_proposal(p) qdir = supervise.queue_dir_for_slug("dev")
return supervise_cli.QueuedProposal(proposal=p) qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_approve_writes_response(self): def test_approve_writes_response(self):
qp = self._enqueue() qp = self._enqueue()
@@ -127,7 +135,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
return_value=("routes: []\n", "routes:\n - host: example.com\n"), return_value=("routes: []\n", "routes:\n - host: example.com\n"),
): ):
supervise_cli.approve(qp) supervise_cli.approve(qp)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status) self.assertEqual(STATUS_APPROVED, resp.status)
self.assertIsNone(resp.final_file) self.assertIsNone(resp.final_file)
@@ -142,7 +150,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
final_file="routes:\n - host: edited.example.com\n", final_file="routes:\n - host: edited.example.com\n",
notes="tweaked", notes="tweaked",
) )
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_MODIFIED, resp.status) self.assertEqual(STATUS_MODIFIED, resp.status)
self.assertEqual("routes:\n - host: edited.example.com\n", resp.final_file) self.assertEqual("routes:\n - host: edited.example.com\n", resp.final_file)
self.assertEqual("tweaked", resp.notes) self.assertEqual("tweaked", resp.notes)
@@ -150,7 +158,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_reject_writes_rejection(self): def test_reject_writes_rejection(self):
qp = self._enqueue() qp = self._enqueue()
supervise_cli.reject(qp, reason="nope") supervise_cli.reject(qp, reason="nope")
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status) self.assertEqual(STATUS_REJECTED, resp.status)
self.assertEqual("nope", resp.notes) self.assertEqual("nope", resp.notes)
@@ -173,33 +181,36 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_approve_gitleaks_allow_leaves_response_for_gate(self): def test_approve_gitleaks_allow_leaves_response_for_gate(self):
qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
supervise_cli.approve(qp, notes="dummy fixture") supervise_cli.approve(qp, notes="dummy fixture")
# Gate polls the DB for the response; TUI must not archive it. # Gate polls the queue dir for the response; TUI must not archive it.
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status) self.assertEqual(STATUS_APPROVED, resp.status)
self.assertEqual("dummy fixture", resp.notes) self.assertEqual("dummy fixture", resp.notes)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_tui_gitleaks_allow_requires_reason(self): def test_tui_gitleaks_allow_requires_reason(self):
qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
with patch.object(supervise_cli, "_prompt", return_value=""): with patch.object(supervise_cli, "_prompt", return_value=""):
status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type]
self.assertEqual("approve aborted (empty reason)", status) self.assertEqual("approve aborted (empty reason)", status)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_tui_gitleaks_allow_writes_reason(self): def test_tui_gitleaks_allow_writes_reason(self):
qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW) qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
with patch.object(supervise_cli, "_prompt", return_value="test fixture"): with patch.object(supervise_cli, "_prompt", return_value="test fixture"):
status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type]
self.assertIn("approved gitleaks-allow", status) self.assertIn("approved gitleaks-allow", status)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual("test fixture", resp.notes) self.assertEqual("test fixture", resp.notes)
def test_approve_token_allow_leaves_response_for_egress(self): def test_approve_token_allow_leaves_response_for_egress(self):
qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
supervise_cli.approve(qp, notes="false positive") supervise_cli.approve(qp, notes="false positive")
# The egress addon polls the DB for the response; the TUI must # The egress addon polls the queue dir for the response; the TUI must
# not archive it (the addon archives after reading). # not archive it (the addon archives after reading).
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status) self.assertEqual(STATUS_APPROVED, resp.status)
self.assertEqual("false positive", resp.notes) self.assertEqual("false positive", resp.notes)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_token_allow_writes_no_audit_log(self): def test_token_allow_writes_no_audit_log(self):
qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
@@ -211,13 +222,14 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
with patch.object(supervise_cli, "_prompt", return_value=""): with patch.object(supervise_cli, "_prompt", return_value=""):
status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type]
self.assertEqual("approve aborted (empty reason)", status) self.assertEqual("approve aborted (empty reason)", status)
self.assertFalse((qp.queue_dir / "processed").exists())
def test_tui_token_allow_writes_reason(self): def test_tui_token_allow_writes_reason(self):
qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW) qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
with patch.object(supervise_cli, "_prompt", return_value="legit"): with patch.object(supervise_cli, "_prompt", return_value="legit"):
status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type] status = supervise_cli._approve_from_tui(None, qp) # type: ignore[arg-type]
self.assertIn("approved egress-token-allow", status) self.assertIn("approved egress-token-allow", status)
resp = read_response(qp.proposal.bottle_slug, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual("legit", resp.notes) self.assertEqual("legit", resp.notes)
def test_suffix_for_token_allow_is_txt(self): def test_suffix_for_token_allow_is_txt(self):
+57 -61
View File
@@ -4,23 +4,22 @@ fallback paths."""
from __future__ import annotations from __future__ import annotations
import os
import tempfile import tempfile
import time import time
import unittest import unittest
from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
from bot_bottle import supervise from bot_bottle import supervise
from bot_bottle.supervise import ( from bot_bottle.supervise import (
AuditEntry,
Proposal, Proposal,
STATUS_APPROVED,
TOOL_EGRESS_ALLOW, TOOL_EGRESS_ALLOW,
list_pending_proposals, list_pending_proposals,
read_audit_entries, read_audit_entries,
read_proposal, read_proposal,
read_response, read_response,
wait_for_response, wait_for_response,
write_audit_entry,
) )
@@ -38,53 +37,58 @@ class TestPathHelpers(unittest.TestCase):
def test_bot_bottle_root(self) -> None: def test_bot_bottle_root(self) -> None:
self.assertTrue(str(supervise.bot_bottle_root()).endswith(".bot-bottle")) 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): class TestReadMalformed(unittest.TestCase):
def test_read_proposal_missing_row(self) -> None: def test_read_proposal_non_dict(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}), \ (Path(d) / "p.proposal.json").write_text("[]")
self.assertRaises(FileNotFoundError): with self.assertRaises(ValueError):
read_proposal("slug", "p") read_proposal(Path(d), "p")
def test_read_response_missing_row(self) -> None: def test_read_response_non_dict(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}), \ (Path(d) / "p.response.json").write_text("[]")
self.assertRaises(FileNotFoundError): with self.assertRaises(ValueError):
read_response("slug", "p") read_response(Path(d), "p")
def test_list_pending_reads_db_only(self) -> None: def test_list_pending_skips_malformed(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}): qd = Path(d)
supervise.write_proposal(_proposal()) (qd / "bad.proposal.json").write_text("{ not json")
pending = list_pending_proposals("slug") (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)
self.assertEqual(1, len(pending)) self.assertEqual(1, len(pending))
self.assertEqual("slug", pending[0].bottle_slug) self.assertEqual("slug", pending[0].bottle_slug)
def test_list_pending_skips_when_response_present(self) -> None: def test_list_pending_skips_when_response_present(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}): qd = Path(d)
p = _proposal() p = _proposal()
supervise.write_proposal(p) supervise.write_proposal(qd, p)
supervise.write_response("slug", supervise.Response( (qd / f"{p.id}.response.json").write_text("{}") # response exists -> skipped
proposal_id=p.id, self.assertEqual([], list_pending_proposals(qd))
status=STATUS_APPROVED,
notes="",
))
self.assertEqual([], list_pending_proposals("slug"))
class TestWaitForResponse(unittest.TestCase): class TestWaitForResponse(unittest.TestCase):
def test_missing_response_times_out(self) -> None: def test_malformed_response_then_timeout(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}), \ (Path(d) / "p.response.json").write_text("{ not json")
self.assertRaises(TimeoutError): with self.assertRaises(TimeoutError):
wait_for_response("slug", "p", deadline=time.monotonic()) wait_for_response(Path(d), "p", deadline=time.monotonic())
def test_empty_db_response_does_not_count(self) -> None: def test_incomplete_response_then_timeout(self) -> None:
with tempfile.TemporaryDirectory() as d: with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {"HOME": d}), \ (Path(d) / "p.response.json").write_text("{}") # dict but from_dict raises
self.assertRaises(TimeoutError): with self.assertRaises(TimeoutError):
wait_for_response("slug", "p", deadline=time.monotonic()) wait_for_response(Path(d), "p", deadline=time.monotonic())
class TestReadAuditEntries(unittest.TestCase): class TestReadAuditEntries(unittest.TestCase):
@@ -93,43 +97,35 @@ class TestReadAuditEntries(unittest.TestCase):
patch.dict("os.environ", {"HOME": home}): patch.dict("os.environ", {"HOME": home}):
self.assertEqual([], read_audit_entries("egress", "nope")) self.assertEqual([], read_audit_entries("egress", "nope"))
def test_reads_entries_from_db(self) -> None: def test_skips_malformed_lines(self) -> None:
with tempfile.TemporaryDirectory() as home, \
patch.dict("os.environ", {"HOME": home}):
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, \ with tempfile.TemporaryDirectory() as home, \
patch.dict("os.environ", {"HOME": home}): patch.dict("os.environ", {"HOME": home}):
path = supervise.audit_log_path("egress", "slug") path = supervise.audit_log_path("egress", "slug")
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text( valid = (
'{"timestamp": "t", "bottle_slug": "slug", "component": "egress",' '{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
' "operator_action": "approve", "operator_notes": "",' ' "operator_action": "approve", "operator_notes": "",'
' "justification": "", "diff": ""}\n' ' "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"
) )
entries = read_audit_entries("egress", "slug") entries = read_audit_entries("egress", "slug")
self.assertEqual([], entries) self.assertEqual(1, len(entries))
self.assertEqual("approve", entries[0].operator_action)
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)
if __name__ == "__main__": if __name__ == "__main__":
+18 -22
View File
@@ -112,7 +112,7 @@ class TestRpcErrorTaxonomy(unittest.TestCase):
validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n") validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
def test_unknown_tool_in_tools_call_is_client_error(self): def test_unknown_tool_in_tools_call_is_client_error(self):
config = ServerConfig(bottle_slug="dev") config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
with self.assertRaises(_RpcClientError) as cm: with self.assertRaises(_RpcClientError) as cm:
handle_tools_call({"name": "no-such-tool", "arguments": {}}, config) handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
@@ -122,9 +122,9 @@ class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
def test_write_proposal_os_error_raises_internal(self): def test_write_proposal_os_error_raises_internal(self):
config = ServerConfig( config = ServerConfig(
bottle_slug="dev", bottle_slug="dev",
queue_dir=Path("/dev/null/cannot-exist"),
) )
with patch.object(_sv, "write_proposal", side_effect=OSError("disk full")), \ with self.assertRaises(_RpcInternalError) as cm:
self.assertRaises(_RpcInternalError) as cm:
handle_tools_call( handle_tools_call(
{ {
"name": _sv.TOOL_EGRESS_ALLOW, "name": _sv.TOOL_EGRESS_ALLOW,
@@ -265,31 +265,21 @@ class TestHandleToolsList(unittest.TestCase):
class TestHandleToolsCall(unittest.TestCase): class TestHandleToolsCall(unittest.TestCase):
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-server-test.") self._tmp = tempfile.TemporaryDirectory(prefix="supervise-server-test.")
self._home_patch = self._patch_home(Path(self._tmp.name)) self.queue_dir = Path(self._tmp.name)
self.config = ServerConfig(bottle_slug="dev") self.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
def tearDown(self): def tearDown(self):
self._home_patch()
self._tmp.cleanup() self._tmp.cleanup()
def _patch_home(self, fake_home: Path):
original = _sv.bot_bottle_root
def fake_root() -> Path:
return fake_home / ".bot-bottle"
_sv.bot_bottle_root = fake_root # type: ignore[assignment]
return lambda: setattr(_sv, "bot_bottle_root", original)
def _respond_when_proposal_appears(self, status: str, notes: str = "") -> threading.Thread: def _respond_when_proposal_appears(self, status: str, notes: str = "") -> threading.Thread:
"""Background thread: poll the queue for a fresh proposal, write a """Background thread: poll the queue for a fresh proposal, write a
matching response. Returns the thread so the test can join it.""" matching response. Returns the thread so the test can join it."""
def runner(): def runner():
for _ in range(200): for _ in range(200):
pending = _sv.list_pending_proposals("dev") pending = _sv.list_pending_proposals(self.queue_dir)
if pending: if pending:
p = pending[0] p = pending[0]
_sv.write_response("dev", _sv.Response( _sv.write_response(self.queue_dir, _sv.Response(
proposal_id=p.id, status=status, notes=notes, proposal_id=p.id, status=status, notes=notes,
)) ))
return return
@@ -422,11 +412,15 @@ class TestHandleToolsCall(unittest.TestCase):
finally: finally:
responder.join() responder.join()
# No pending proposals left after archive. # No pending proposals left after archive.
self.assertEqual([], _sv.list_pending_proposals("dev")) 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))
def test_pending_response_times_out_without_archive(self): def test_pending_response_times_out_without_archive(self):
config = ServerConfig( config = ServerConfig(
bottle_slug="dev", bottle_slug="dev",
queue_dir=self.queue_dir,
response_timeout_seconds=0.05, response_timeout_seconds=0.05,
) )
result = handle_tools_call( result = handle_tools_call(
@@ -444,7 +438,8 @@ class TestHandleToolsCall(unittest.TestCase):
text = result["content"][0]["text"] # type: ignore[index] text = result["content"][0]["text"] # type: ignore[index]
self.assertIn("status: pending", text) self.assertIn("status: pending", text)
self.assertIn("proposal remains queued", text) self.assertIn("proposal remains queued", text)
self.assertEqual(1, len(_sv.list_pending_proposals("dev"))) self.assertEqual(1, len(_sv.list_pending_proposals(self.queue_dir)))
self.assertFalse((self.queue_dir / "processed").exists())
class TestHandleListEgressRoutes(unittest.TestCase): class TestHandleListEgressRoutes(unittest.TestCase):
@@ -466,7 +461,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()): with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes( result = handle_list_egress_routes(
{}, {},
ServerConfig(bottle_slug="dev"), ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
) )
self.assertFalse(result["isError"]) # type: ignore[index] self.assertFalse(result["isError"]) # type: ignore[index]
@@ -481,7 +476,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()): with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes( result = handle_list_egress_routes(
{}, {},
ServerConfig(bottle_slug="dev"), ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
) )
self.assertTrue(result["isError"]) # type: ignore[index] self.assertTrue(result["isError"]) # type: ignore[index]
@@ -549,6 +544,7 @@ class TestHttpEndToEnd(unittest.TestCase):
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-http-test.") self._tmp = tempfile.TemporaryDirectory(prefix="supervise-http-test.")
self.queue_dir = Path(self._tmp.name)
# Pick a random port by binding to :0 first. # Pick a random port by binding to :0 first.
import socket import socket
s = socket.socket() s = socket.socket()
@@ -556,7 +552,7 @@ class TestHttpEndToEnd(unittest.TestCase):
self.port = s.getsockname()[1] self.port = s.getsockname()[1]
s.close() s.close()
self.server = MCPServer(("127.0.0.1", self.port), MCPHandler) self.server = MCPServer(("127.0.0.1", self.port), MCPHandler)
self.server.config = ServerConfig(bottle_slug="dev") self.server.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
self.thread = threading.Thread( self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True, target=self.server.serve_forever, daemon=True,
) )