refactor: extract TableMigrations and DbStore base class

Adds bot_bottle/migrations.py (TableMigrations) and bot_bottle/db_store.py
(DbStore) per PR review. Both stores now inherit from DbStore and hold a
TableMigrations instance instead of duplicating schema-version logic inline.

Dockerfile.sidecars

@@ -18,7 +18,7 @@
 #   /git-gate-entrypoint.sh              docker-cp'd at start time
 #   /git-gate/creds/*                    docker-cp'd at start time
 #   /git/*                               bare repos, populated at runtime
-#   /run/supervise/queue/                bind-mounted at run time
+#   /run/supervise/bot-bottle.db         bind-mounted at run time
 #   /home/mitmproxy/.mitmproxy/          mitmproxy CA dir
 #
 # Exposed ports inside the container:
@@ -66,6 +66,10 @@ COPY bot_bottle/egress_dlp_config.py /app/egress_dlp_config.py
 COPY bot_bottle/egress_addon.py      /app/egress_addon.py
 COPY bot_bottle/dlp_detectors.py     /app/dlp_detectors.py
 COPY bot_bottle/yaml_subset.py       /app/yaml_subset.py
+COPY bot_bottle/migrations.py        /app/migrations.py
+COPY bot_bottle/db_store.py          /app/db_store.py
+COPY bot_bottle/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_server.py  /app/supervise_server.py
 COPY bot_bottle/sidecar_init.py      /app/sidecar_init.py
@@ -81,7 +85,7 @@ RUN mkdir -p \
       /etc/git-gate \
       /git-gate/creds \
       /git \
-      /run/supervise/queue \
+      /run/supervise \
       /home/mitmproxy/.mitmproxy
 
 # Documentation only — the compose renderer publishes whichever

README.md

@@ -5,8 +5,8 @@
 # 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)
-[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
-[![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)
+[![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)
 
 **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.
 

bot_bottle/agent_provider.py

@@ -209,6 +209,15 @@ class AgentProvider(ABC):
         the supervise sidecar is reachable. No-op when
         `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:
         """Install the egress MITM CA into the agent's trust store.
 

bot_bottle/audit_store.py

@@ -0,0 +1,113 @@
+"""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
+
+try:
+    from .db_store import DbStore
+    from .migrations import TableMigrations
+except ImportError:
+    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+def get_supervise_mod() -> 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
+
+
+# One entry per schema version: _MIGRATIONS.migrations[0] brings a fresh DB
+# to version 1, [1] to version 2, and so on.  Add new migrations at the end;
+# never edit existing ones.
+_MIGRATIONS = TableMigrations("audit_store", [
+    # v1 — initial schema
+    """
+    CREATE TABLE IF NOT EXISTS supervise_audit_entries (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        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
+    )
+    """,
+])
+
+
+class AuditStore(DbStore):
+    """SQLite-backed persistent store for supervise audit entries."""
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        resolved = db_path or get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
+        super().__init__(resolved, _MIGRATIONS)
+
+    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 [self._row_to_entry(row) for row in rows]
+
+    @staticmethod
+    def _row_to_entry(row: sqlite3.Row) -> AuditEntry:
+        m = get_supervise_mod()
+        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"],
+        )
+
+
+__all__ = ["AuditStore"]

bot_bottle/backend/docker/compose.py

@@ -34,7 +34,7 @@ from ...egress import (
 from ...git_gate import GIT_GATE_HOSTNAME
 from ...log import die, warn
 from ...supervise import (
-    QUEUE_DIR_IN_CONTAINER,
+    DB_PATH_IN_CONTAINER,
     SUPERVISE_HOSTNAME,
     SUPERVISE_PORT,
 )
@@ -163,16 +163,15 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
     if sp is not None:
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
+            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
             f"SUPERVISE_PORT={SUPERVISE_PORT}",
         ]
         volumes.append({
             "type": "bind",
-            "source": str(sp.queue_dir),
+            "source": str(sp.db_path),
-            "target": QUEUE_DIR_IN_CONTAINER,
+            "target": DB_PATH_IN_CONTAINER,
             "read_only": False,
         })
-
     internal_aliases = [EGRESS_HOSTNAME]
     if gp.upstreams:
         internal_aliases.append(GIT_GATE_HOSTNAME)

bot_bottle/backend/docker/launch.py

@@ -37,7 +37,10 @@ from pathlib import Path
 from typing import Callable, Generator
 
 from ...egress import egress_resolve_token_values
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
+    provision_git_gate_dynamic_keys,
+    revoke_git_gate_provisioned_keys,
+)
 from ...log import info, warn
 from . import network as network_mod
 from . import util as docker_mod
@@ -118,6 +121,11 @@ def launch(
 
         git_gate_plan = plan.git_gate_plan
         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,
                 internal_network=internal_network,

bot_bottle/backend/macos_container/launch.py

@@ -28,9 +28,12 @@ from ...egress import (
     egress_resolve_token_values,
     egress_sidecar_env_entries,
 )
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
+    provision_git_gate_dynamic_keys,
+    revoke_git_gate_provisioned_keys,
+)
 from ...log import die, info, warn
-from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
+from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
 from ..docker.git_gate import (
@@ -98,6 +101,8 @@ def launch(
         egress_network = egress_network_name(plan.slug)
         _create_networks(internal_network, egress_network, stack)
 
+        plan = _provision_git_gate_keys(plan)
+
         sidecar_name = sidecar_container_name(plan.slug)
         container_mod.force_remove_container(sidecar_name)
         _start_sidecar_bundle(plan, sidecar_name, internal_network, egress_network)
@@ -241,6 +246,19 @@ 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:
     gp = plan.git_gate_plan
     if not gp.upstreams:
@@ -361,7 +379,7 @@ def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
     if plan.supervise_plan is not None:
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
+            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
             f"SUPERVISE_PORT={SUPERVISE_PORT}",
         ]
     return tuple(env)
@@ -387,7 +405,7 @@ def _sidecar_mounts(
 
     sp = plan.supervise_plan
     if sp is not None:
-        mounts.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
+        mounts.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
 
     return tuple(mounts)
 

bot_bottle/backend/smolmachines/launch.py

@@ -27,7 +27,7 @@ from ...egress import (
     egress_resolve_token_values,
     egress_sidecar_env_entries,
 )
-from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
+from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker import util as docker_mod
 from ..docker.egress import (
@@ -41,7 +41,10 @@ from ..docker.git_gate import (
     GIT_GATE_ENTRYPOINT_IN_CONTAINER,
     GIT_GATE_HOOK_IN_CONTAINER,
 )
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
+    provision_git_gate_dynamic_keys,
+    revoke_git_gate_provisioned_keys,
+)
 from ...log import info, warn
 from ...bottle_state import (
     egress_state_dir,
@@ -174,6 +177,7 @@ def _start_bundle(
 ) -> SmolmachinesBottlePlan:
     """Build the BundleLaunchSpec, resolve token env, start the
     sidecar bundle container, and register teardown."""
+    plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, loopback_ip)
     token_env = _resolve_token_env(plan, dict(os.environ))
     _bundle.ensure_bundle_image(bundle_spec.image)
@@ -182,6 +186,19 @@ def _start_bundle(
     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(
     plan: SmolmachinesBottlePlan,
     loopback_ip: str,
@@ -352,10 +369,10 @@ def _bundle_launch_spec(
         daemons.append("supervise")
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
+            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
             f"SUPERVISE_PORT={SUPERVISE_PORT}",
         ]
-        volumes.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
+        volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
 
     # Container ports the agent reaches from the smolvm guest —
     # published on host loopback so the guest can dial via TSI +

bot_bottle/bottle_state.py

@@ -284,9 +284,8 @@ def git_gate_state_dir(identity: str) -> Path:
 
 def supervise_state_dir(identity: str) -> Path:
     """State subdir reserved for supervise sidecar bind-mount sources.
-    The queue dir is intentionally NOT under here — it lives at
+    Runtime queue/audit rows live in the host-level bot-bottle SQLite
-    ~/.bot-bottle/queue/<slug>/ alongside the audit logs, so it
+    database, so they survive state-dir cleanup."""
-    survives state-dir cleanup."""
     return bottle_state_dir(identity) / _SUPERVISE_SUBDIR
 
 

bot_bottle/cli/start.py

@@ -2,6 +2,11 @@
 interactive claude-code session. The container is torn down when the
 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 private orchestrator `_launch_bottle`.
 """
@@ -16,7 +21,7 @@ import tempfile
 from pathlib import Path
 from typing import Callable
 
-from ..agent_provider import runtime_for
+from ..agent_provider import get_provider, runtime_for
 from ..backend import (
     Bottle,
     BottleSpec,
@@ -31,7 +36,7 @@ from ..bottle_state import (
     is_preserved,
     mark_preserved,
 )
-from ..log import info
+from ..log import info, die
 from ..manifest import Manifest, ManifestIndex
 from ._common import PROG, USER_CWD, read_tty_line
 from . import tui
@@ -50,6 +55,39 @@ def cmd_start(argv: list[str]) -> int:
             "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(
         "name",
         nargs="?",
@@ -61,6 +99,12 @@ def cmd_start(argv: list[str]) -> int:
     dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
 
     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
     if agent_name is None:
@@ -71,8 +115,6 @@ def cmd_start(argv: list[str]) -> int:
         if agent_name is None:
             return 0
 
-    backend_name: str | None = args.backend
-
     # Bottle multiselect: always show after agent selection so operators
     # can compose bottles at launch time without editing agent manifests.
     available_bottles = manifest.all_bottle_names
@@ -109,6 +151,83 @@ 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 ------------------------------------------------------
 
 
@@ -376,10 +495,19 @@ def _launch_bottle(
     *,
     dry_run: bool,
     backend_name: str | None = None,
+    assume_yes: bool = False,
+    headless_prompt_text: str = "",
 ) -> int:
     """Shared launch core for `start` and `resume`. Builds the plan,
     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."))
     identity = ""
     try:
@@ -387,7 +515,7 @@ def _launch_bottle(
             spec,
             stage_dir=stage_dir,
             render_preflight=_text_render_preflight(),
-            prompt_yes=_text_prompt_yes,
+            prompt_yes=(lambda: True) if assume_yes else _text_prompt_yes,
             dry_run=dry_run,
             backend_name=backend_name,
         )
@@ -397,10 +525,17 @@ def _launch_bottle(
         backend = get_bottle_backend(backend_name)
         with backend.launch(plan) as bottle:
             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(
                 bottle,
                 agent_provider_template=agent_provider_template,
-                startup_args=plan.agent_provision.startup_args,
+                startup_args=plan.agent_provision.startup_args + extra_args,
             )
             info(
                 f"session ended (exit {exit_code}); "

bot_bottle/cli/supervise.py

@@ -45,7 +45,7 @@ from ..supervise import (
     TOOL_EGRESS_BLOCK,
     TOOL_GITLEAKS_ALLOW,
     TOOL_EGRESS_TOKEN_ALLOW,
-    list_pending_proposals,
+    list_all_pending_proposals,
     render_diff,
     write_audit_entry,
     write_response,
@@ -63,10 +63,9 @@ _REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_AL
 
 @dataclass(frozen=True)
 class QueuedProposal:
-    """A pending proposal plus the queue dir it was found in."""
+    """A pending proposal from the supervise queue."""
 
     proposal: Proposal
-    queue_dir: Path
 
 
 # Errors any remediation engine may raise. Caught by the TUI key
@@ -86,16 +85,11 @@ def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
 
 
 def discover_pending() -> list[QueuedProposal]:
-    """Walk ~/.bot-bottle/queue/* and collect pending proposals."""
+    """Collect pending proposals across bottles."""
-    queue_root = _supervise.bot_bottle_root() / "queue"
+    out = [
-    if not queue_root.is_dir():
+        QueuedProposal(proposal=proposal)
-        return []
+        for proposal in list_all_pending_proposals()
-    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)
     return out
 
@@ -118,7 +112,6 @@ def _detail_lines(
         (f"tool: {p.tool}", 0),
         (f"id: {p.id}", 0),
         (f"arrived: {p.arrival_timestamp}", 0),
-        (f"queue: {qp.queue_dir}", 0),
         ("", 0),
         ("justification:", 0),
     ]
@@ -165,7 +158,7 @@ def approve(
         notes=notes,
         final_file=final_file,
     )
-    write_response(qp.queue_dir, response)
+    write_response(qp.proposal.bottle_slug, response)
     _write_audit(
         qp, action=status, notes=notes,
         diff_before=diff_before, diff_after=diff_after,
@@ -179,7 +172,7 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
         notes=reason,
         final_file=None,
     )
-    write_response(qp.queue_dir, response)
+    write_response(qp.proposal.bottle_slug, response)
     _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
 
 

bot_bottle/contrib/claude/agent_provider.py

@@ -313,6 +313,9 @@ class ClaudeAgentProvider(AgentProvider):
                 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:
     result = bottle.exec(script, user="root")

bot_bottle/contrib/codex/agent_provider.py

@@ -279,6 +279,9 @@ class CodexAgentProvider(AgentProvider):
                 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:
     result = bottle.exec(script, user="root")

bot_bottle/contrib/pi/agent_provider.py

@@ -315,6 +315,9 @@ class PiAgentProvider(AgentProvider):
     ) -> None:
         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:
     result = bottle.exec(script, user="root")

bot_bottle/db_store.py

@@ -0,0 +1,40 @@
+"""Shared SQLite-backed store base class for bot-bottle (PRD 0013)."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+try:
+    from .migrations import TableMigrations
+except ImportError:
+    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+class DbStore:
+    """Base for SQLite-backed stores. Subclasses resolve db_path then call super().__init__."""
+
+    def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
+        self.db_path = db_path
+        self._migrations = migrations
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init()
+
+    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:
+            self._migrations.apply(conn)
+        self._chmod()
+
+    def _chmod(self) -> None:
+        try:
+            self.db_path.chmod(0o600)
+        except OSError:
+            pass
+
+
+__all__ = ["DbStore"]

bot_bottle/egress_addon.py

@@ -79,14 +79,13 @@ class EgressAddon:
         # only — a restart re-prompts. Mutated only from the asyncio loop that
         # runs the addon hooks, so no lock is needed.
         self.safe_tokens: set[str] = set()
-        self._supervise_queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "").strip()
         self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
         self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
         self._reload(initial=True)
         self._install_sighup()
 
     def _supervise_available(self) -> bool:
-        return bool(self._supervise_queue_dir and self._supervise_slug)
+        return bool(self._supervise_slug)
 
     def _reload(self, *, initial: bool = False) -> None:
         try:
@@ -393,9 +392,8 @@ class EgressAddon:
             justification=_TOKEN_ALLOW_JUSTIFICATION,
             current_file_hash=_sv.sha256_hex(payload),
         )
-        queue_dir = Path(self._supervise_queue_dir)
         try:
-            _sv.write_proposal(queue_dir, proposal)
+            _sv.write_proposal(proposal)
         except OSError as e:
             sys.stderr.write(
                 f"egress: could not queue token-allow proposal: {e}; "
@@ -411,8 +409,8 @@ class EgressAddon:
             **self._req_ctx(flow),
         }) + "\n")
 
-        response = await self._await_token_response(queue_dir, proposal.id)
+        response = await self._await_token_response(proposal.id)
-        _sv.archive_proposal(queue_dir, proposal.id)
+        _sv.archive_proposal(self._supervise_slug, proposal.id)
 
         if response is not None and response.status in (
             _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
@@ -439,16 +437,15 @@ class EgressAddon:
 
     async def _await_token_response(
         self,
-        queue_dir: Path,
         proposal_id: str,
     ) -> "_sv.Response | None":
-        """Poll the queue dir for the operator's response without blocking the
+        """Poll the DB for the operator's response without blocking the
         proxy event loop. Returns the Response, or None on timeout."""
         loop = asyncio.get_running_loop()
         deadline = loop.time() + self._token_allow_timeout
         while True:
             try:
-                return _sv.read_response(queue_dir, proposal_id)
+                return _sv.read_response(self._supervise_slug, proposal_id)
             except (OSError, ValueError, KeyError):
                 # Not written yet, or a partial/malformed write — retry until
                 # the deadline, then fail closed.

bot_bottle/git_gate.py

@@ -30,7 +30,6 @@ backend-specific and lives on concrete subclasses (see
 
 from __future__ import annotations
 
-import dataclasses
 from abc import ABC
 from dataclasses import dataclass
 from pathlib import Path
@@ -53,6 +52,7 @@ from .git_gate_render import (
     _gitconfig_validate_value,
 )
 from .git_gate_provision import (
+    provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
     _provision_dynamic_key,
     _resolve_identity_file,
@@ -93,20 +93,14 @@ class GitGate(ABC):
         entrypoint, pre-receive hook, and access-hook scripts (mode
         600) under `stage_dir`. Pure host-side, no docker subprocess.
 
-        For `gitea` key entries, also generates and registers
+        For `gitea` key entries, the returned upstream intentionally
-        a fresh deploy key via the forge API and writes the private key
+        has an empty identity file. Backend launch fills that in after
-        + key ID to `stage_dir`.
+        the operator confirms the preflight.
 
         Returned plan is incomplete: the launch step must fill
         `internal_network` / `egress_network` via `dataclasses.replace`
         before passing the plan to `.start`."""
-        upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
+        upstreams = 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.write_text(git_gate_render_entrypoint(upstreams))
         entrypoint.chmod(0o600)
@@ -162,6 +156,7 @@ __all__ = [
     "git_gate_render_entrypoint",
     "git_gate_render_hook",
     "git_gate_render_access_hook",
+    "provision_git_gate_dynamic_keys",
     "revoke_git_gate_provisioned_keys",
     "_gitconfig_validate_value",
     "_provision_dynamic_key",

bot_bottle/git_gate_provision.py

@@ -9,10 +9,16 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.
 from __future__ import annotations
 
 import os
+import dataclasses
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from .log import info
 from .manifest import ManifestBottle, ManifestGitEntry
+from .git_gate_render import GitGateUpstream
+
+if TYPE_CHECKING:
+    from .git_gate import GitGatePlan
 
 def _provision_dynamic_key(
     entry: ManifestGitEntry,
@@ -95,8 +101,45 @@ def _resolve_identity_file(entry: ManifestGitEntry, slug: str, stage_dir: Path)
     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__ = [
     "revoke_git_gate_provisioned_keys",
+    "provision_git_gate_dynamic_keys",
     "_provision_dynamic_key",
     "_resolve_identity_file",
 ]

bot_bottle/git_gate_render.py

@@ -234,13 +234,13 @@ import hashlib
 import json
 import os
 import sys
-import uuid
 from pathlib import Path
 
+from bot_bottle import supervise as _sv
+
 report_path = Path(sys.argv[1])
-queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "")
 slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
-if not queue_dir or not slug:
+if not slug:
     sys.exit(2)
 
 try:
@@ -277,31 +277,19 @@ for i, finding in enumerate(raw, 1):
     ])
 
 payload = "\n".join(lines).rstrip() + "\n"
-proposal_id = str(uuid.uuid4())
+proposal = _sv.Proposal.new(
-proposal = {
+    bottle_slug=slug,
-    "id": proposal_id,
+    tool=_sv.TOOL_GITLEAKS_ALLOW,
-    "bottle_slug": slug,
+    proposed_file=payload,
-    "tool": "gitleaks-allow",
+    justification=(
-    "proposed_file": payload,
-    "justification": (
         "git-gate found gitleaks findings hidden by # gitleaks:allow; "
         "approve only for dummy test fixtures or confirmed false positives"
     ),
-    "arrival_timestamp": datetime.datetime.now(
+    current_file_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
-        datetime.timezone.utc
+    now=datetime.datetime.now(datetime.timezone.utc),
-    ).isoformat(),
+)
-    "current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+_sv.write_proposal(proposal)
-}
+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
   )
   rc=$?
@@ -314,8 +302,7 @@ PY
     return 1
   fi
 
-  queue_dir=${SUPERVISE_QUEUE_DIR:-}
+  slug=${SUPERVISE_BOTTLE_SLUG:-}
-  response_file="$queue_dir/${proposal_id}.response.json"
   timeout=${SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS:-300}
   case "$timeout" in
     ''|*[!0-9]*)
@@ -327,26 +314,35 @@ PY
   echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
   waited=0
   while [ "$waited" -lt "$timeout" ]; do
-    if [ -f "$response_file" ]; then
+    status=$(python3 - "$slug" "$proposal_id" <<'PY'
-      status=$(python3 - "$response_file" <<'PY'
-import json
 import sys
+
+from bot_bottle import supervise as _sv
+
+slug = sys.argv[1]
 try:
-    with open(sys.argv[1], encoding="utf-8") as f:
+    response = _sv.read_response(slug, sys.argv[2])
-        raw = json.load(f)
+except FileNotFoundError:
-except (OSError, json.JSONDecodeError):
+    sys.exit(2)
-    sys.exit(1)
+print(response.status)
-status = raw.get("status")
-if not isinstance(status, str):
-    sys.exit(1)
-print(status)
 PY
-      ) || status=""
+    )
+    rc=$?
+    if [ "$rc" -eq 2 ]; then
+      status=""
+    elif [ "$rc" -ne 0 ]; then
+      status="invalid"
+    fi
+    if [ -n "$status" ]; then
       case "$status" in
         approved|modified)
-          mkdir -p "$queue_dir/processed"
+          python3 - "$slug" "$proposal_id" <<'PY' || true
-          mv -f "$queue_dir/${proposal_id}.proposal.json" "$queue_dir/processed/" 2>/dev/null || true
+import sys
-          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
           return 0
           ;;
@@ -499,4 +495,3 @@ if ! git -C "$repo_dir" rev-parse --verify HEAD >/dev/null 2>&1; then
 fi
 exit 0
 """
-

bot_bottle/git_http_backend.py

@@ -16,11 +16,16 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import urlsplit
 
-from .git_gate import GIT_GATE_TIMEOUT_SECS
-
 
 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.
 MAX_BODY_BYTES = 100 * 1024 * 1024
 

bot_bottle/migrations.py

@@ -0,0 +1,37 @@
+"""SQLite migration runner for bot-bottle stores."""
+
+from __future__ import annotations
+
+import sqlite3
+
+
+class TableMigrations:
+    """Runs a sequential list of DDL migrations tracked by schema_key in schema_versions."""
+
+    def __init__(self, schema_key: str, migrations: list[str]) -> None:
+        self.schema_key = schema_key
+        self.migrations = migrations
+
+    def apply(self, conn: sqlite3.Connection) -> None:
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS schema_versions (
+                module TEXT PRIMARY KEY,
+                version INTEGER NOT NULL DEFAULT 0
+            )
+            """
+        )
+        row = conn.execute(
+            "SELECT version FROM schema_versions WHERE module = ?",
+            (self.schema_key,),
+        ).fetchone()
+        version = row[0] if row else 0
+        for i, sql in enumerate(self.migrations[version:], start=version + 1):
+            conn.execute(sql)
+            conn.execute(
+                "INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
+                (self.schema_key, i),
+            )
+
+
+__all__ = ["TableMigrations"]

bot_bottle/queue_store.py

@@ -0,0 +1,240 @@
+"""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
+
+try:
+    from .db_store import DbStore
+    from .migrations import TableMigrations
+except ImportError:
+    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+def get_supervise_mod() -> 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
+
+
+# One entry per schema version: _MIGRATIONS.migrations[0] brings a fresh DB
+# to version 1, [1] to version 2, and so on.  Add new migrations at the end;
+# never edit existing ones.
+_MIGRATIONS = TableMigrations("queue_store", [
+    # v1 — proposals table
+    """
+    CREATE TABLE IF NOT EXISTS supervise_proposals (
+        queue_key TEXT NOT NULL,
+        id TEXT NOT NULL,
+        bottle_slug TEXT NOT NULL,
+        tool TEXT NOT NULL,
+        proposed_file TEXT NOT NULL,
+        justification TEXT NOT NULL,
+        arrival_timestamp TEXT NOT NULL,
+        current_file_hash TEXT NOT NULL,
+        archived INTEGER NOT NULL DEFAULT 0,
+        PRIMARY KEY (queue_key, id)
+    )
+    """,
+    # v2 — responses table
+    """
+    CREATE TABLE IF NOT EXISTS supervise_responses (
+        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)
+    )
+    """,
+])
+
+
+class QueueStore(DbStore):
+    """SQLite-backed persistent store for supervise proposals and responses."""
+
+    def __init__(self, queue_key: str, db_path: Path | None = None) -> None:
+        self.queue_key = queue_key
+        if db_path is not None:
+            resolved = db_path
+        else:
+            # In the sidecar container SUPERVISE_DB_PATH points at the
+            # bind-mounted host DB. On the host this env var is never set,
+            # so we always fall through to host_db_path().
+            env_path = os.environ.get("SUPERVISE_DB_PATH", "").strip()
+            resolved = Path(env_path) if env_path else get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
+        super().__init__(resolved, _MIGRATIONS)
+
+    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 self._row_to_proposal(row)
+
+    def list_pending_proposals(self) -> list[Proposal]:
+        if not self.db_path.is_file():
+            return []
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT p.* FROM supervise_proposals p
+                WHERE p.archived = 0
+                  AND p.queue_key = ?
+                  AND NOT EXISTS (
+                    SELECT 1 FROM supervise_responses r
+                    WHERE r.queue_key = p.queue_key
+                      AND r.proposal_id = p.id
+                      AND r.archived = 0
+                  )
+                ORDER BY p.arrival_timestamp, p.id
+                """,
+                (self.queue_key,),
+            ).fetchall()
+        return [self._row_to_proposal(row) for row in rows]
+
+    def list_all_pending_proposals(self) -> list[Proposal]:
+        if not self.db_path.is_file():
+            return []
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT p.* FROM supervise_proposals p
+                WHERE p.archived = 0
+                  AND NOT EXISTS (
+                    SELECT 1 FROM supervise_responses r
+                    WHERE r.queue_key = p.queue_key
+                      AND r.proposal_id = p.id
+                      AND r.archived = 0
+                  )
+                ORDER BY p.arrival_timestamp, p.id
+                """
+            ).fetchall()
+        return [self._row_to_proposal(row) for row in rows]
+
+    def write_response(self, response: Response) -> Path:
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT OR REPLACE INTO supervise_responses (
+                    queue_key, proposal_id, status, notes, final_file, archived
+                ) VALUES (?, ?, ?, ?, ?, 0)
+                """,
+                (
+                    self.queue_key,
+                    response.proposal_id,
+                    response.status,
+                    response.notes,
+                    response.final_file,
+                ),
+            )
+        self._chmod()
+        return self.db_path
+
+    def read_response(self, proposal_id: str) -> Response:
+        with self._connect() as conn:
+            row = conn.execute(
+                """
+                SELECT * FROM supervise_responses
+                WHERE queue_key = ? AND proposal_id = ? AND archived = 0
+                """,
+                (self.queue_key, proposal_id),
+            ).fetchone()
+        if row is None:
+            raise FileNotFoundError(proposal_id)
+        return self._row_to_response(row)
+
+    def archive_proposal(self, proposal_id: str) -> None:
+        if not self.db_path.is_file():
+            return
+        with self._connect() as conn:
+            conn.execute(
+                """
+                UPDATE supervise_proposals SET archived = 1
+                WHERE queue_key = ? AND id = ?
+                """,
+                (self.queue_key, proposal_id),
+            )
+            conn.execute(
+                """
+                UPDATE supervise_responses SET archived = 1
+                WHERE queue_key = ? AND proposal_id = ?
+                """,
+                (self.queue_key, proposal_id),
+            )
+
+    @staticmethod
+    def _row_to_proposal(row: sqlite3.Row) -> Proposal:
+        m = get_supervise_mod()
+        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"],
+        )
+
+    @staticmethod
+    def _row_to_response(row: sqlite3.Row) -> Response:
+        m = get_supervise_mod()
+        return m.Response(  # type: ignore[attr-defined]
+            proposal_id=row["proposal_id"],
+            status=row["status"],
+            notes=row["notes"],
+            final_file=row["final_file"],
+        )
+
+
+__all__ = ["QueueStore"]

bot_bottle/supervise.py

@@ -9,15 +9,14 @@ calls when it needs an operator-reviewed egress change:
 
 Each tool call: the agent passes the full proposed file plus a
 justification text. The sidecar validates the proposal syntactically,
-writes it to the host's per-bottle queue dir, and holds the tool-call
+writes it to the host SQLite queue table, and holds the tool-call
 connection open. The operator's supervise TUI
 (bot_bottle.cli.supervise) sees the proposal, accepts
-approve / modify / reject, and writes a response file alongside the
+approve / modify / reject, and writes a response row. The sidecar sees
-proposal. The sidecar sees the response and returns `{status, notes}`
+the response and returns `{status, notes}` to the agent.
-to the agent.
 
 This module defines the host-side library: dataclasses for the queue
-file shapes, queue read/write helpers, the audit log writer, and the
+record shapes, queue read/write helpers, the audit log writer, and the
 diff renderer. The in-container sidecar lives in
 bot_bottle/supervise_server.py; the supervise daemon's container
 lifecycle is owned by the sidecar bundle (PRD 0024).
@@ -34,8 +33,6 @@ from __future__ import annotations
 import dataclasses
 import difflib
 import hashlib
-import json
-import os
 import time
 import uuid
 from abc import ABC
@@ -86,8 +83,9 @@ STATUSES: tuple[str, ...] = (STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED)
 # `routes edit <bottle>` verb writes entries with this action.
 ACTION_OPERATOR_EDIT = "operator-edit"
 
-QUEUE_DIR_IN_CONTAINER = "/run/supervise/queue"
+DB_PATH_IN_CONTAINER = "/run/supervise/bot-bottle.db"
 DEFAULT_POLL_INTERVAL_SEC = 0.5
+HOST_DB_FILENAME = "bot-bottle.db"
 
 
 # --- Paths -----------------------------------------------------------------
@@ -97,10 +95,6 @@ def bot_bottle_root() -> Path:
     return Path.home() / ".bot-bottle"
 
 
-def queue_dir_for_slug(slug: str) -> Path:
-    return bot_bottle_root() / "queue" / slug
-
-
 def audit_dir() -> Path:
     return bot_bottle_root() / "audit"
 
@@ -109,14 +103,16 @@ def audit_log_path(component: str, slug: str) -> Path:
     return audit_dir() / f"{component}-{slug}.log"
 
 
+def host_db_path() -> Path:
+    return bot_bottle_root() / HOST_DB_FILENAME
+
+
 # --- Dataclasses -----------------------------------------------------------
 
 
 @dataclass(frozen=True)
 class Proposal:
-    """One pending tool-call from the agent. The sidecar writes one
+    """One pending tool-call from the agent."""
-    of these to the queue dir on a tool call; the operator's TUI
-    reads them; the sidecar polls for a matching Response."""
 
     id: str
     bottle_slug: str
@@ -170,7 +166,7 @@ class Proposal:
 @dataclass(frozen=True)
 class Response:
     """The operator's decision on a proposal. The TUI writes one of
-    these to the queue dir; the sidecar reads it and returns the
+    these to the queue table; the sidecar reads it and returns the
     `{status, notes}` pair to the agent's tool call.
 
     `final_file` carries the file content the supervisor will
@@ -223,90 +219,50 @@ class AuditEntry:
         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 -------------------------------------------------------------
 
 
-def _proposal_filename(proposal_id: str) -> str:
+def write_proposal(proposal: Proposal) -> Path:
-    return f"{proposal_id}.proposal.json"
+    """Persist `proposal` in the queue database, mode 0o600.
-
-
-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."""
-    queue_dir.mkdir(parents=True, exist_ok=True)
+    return QueueStore(proposal.bottle_slug).write_proposal(proposal)
-    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(queue_dir: Path, proposal_id: str) -> Proposal:
+def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal:
-    path = queue_dir / _proposal_filename(proposal_id)
+    return QueueStore(bottle_slug).read_proposal(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(queue_dir: Path) -> list[Proposal]:
+def list_pending_proposals(bottle_slug: str) -> list[Proposal]:
-    """All proposals in `queue_dir` that do not yet have a matching
+    """All proposals for `bottle_slug` that do not yet have a matching
-    response file. Sorted by `arrival_timestamp` so the operator
+    response. Sorted by `arrival_timestamp` so the operator
     sees the queue FIFO."""
-    if not queue_dir.is_dir():
+    return QueueStore(bottle_slug).list_pending_proposals()
-        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 write_response(queue_dir: Path, response: Response) -> Path:
+def list_all_pending_proposals() -> list[Proposal]:
-    queue_dir.mkdir(parents=True, exist_ok=True)
+    """All pending proposals across bottles, sorted FIFO."""
-    path = queue_dir / _response_filename(response.proposal_id)
+    return QueueStore("").list_all_pending_proposals()
-    payload = json.dumps(response.to_dict(), indent=2) + "\n"
-    _atomic_write(path, payload, mode=0o600)
-    return path
 
 
-def read_response(queue_dir: Path, proposal_id: str) -> Response:
+def write_response(bottle_slug: str, response: Response) -> Path:
-    path = queue_dir / _response_filename(proposal_id)
+    return QueueStore(bottle_slug).write_response(response)
-    with path.open() as f:
+
-        raw = json.load(f)
+
-    if not isinstance(raw, dict):
+def read_response(bottle_slug: str, proposal_id: str) -> Response:
-        raise ValueError(f"{path}: top-level must be an object")
+    return QueueStore(bottle_slug).read_response(proposal_id)
-    return Response.from_dict(raw)
 
 
 def wait_for_response(
-    queue_dir: Path,
+    bottle_slug: str,
     proposal_id: str,
     *,
     poll_interval: float = DEFAULT_POLL_INTERVAL_SEC,
@@ -317,90 +273,35 @@ def wait_for_response(
     which the wait raises TimeoutError. None waits forever — the
     natural shape, since the operator's response time is unbounded.
 
-    Polls the filesystem so the implementation stays portable and
+    Polls SQLite so the implementation stays portable and stdlib-only."""
-    stdlib-only."""
+    store = QueueStore(bottle_slug)
-    path = queue_dir / _response_filename(proposal_id)
     while True:
-        if path.exists():
         try:
-                with path.open() as f:
+            return store.read_response(proposal_id)
-                    raw = json.load(f)
+        except FileNotFoundError:
-            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:
             raise TimeoutError(f"no response for proposal {proposal_id!r}")
         time.sleep(poll_interval)
 
 
-def archive_proposal(queue_dir: Path, proposal_id: str) -> None:
+def archive_proposal(bottle_slug: str, proposal_id: str) -> None:
-    """Move both proposal and response files to `<queue_dir>/processed/`.
+    """Mark both proposal and response rows processed.
-    Idempotent — missing files are silently skipped."""
+    Idempotent — missing rows are silently skipped."""
-    processed = queue_dir / "processed"
+    QueueStore(bottle_slug).archive_proposal(proposal_id)
-    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 -------------------------------------------------------------
 
 
 def write_audit_entry(entry: AuditEntry) -> Path:
-    """Append `entry` as one JSON-Lines record to the per-bottle
+    """Append `entry` to the host supervise audit table."""
-    audit log. Acquires an advisory exclusive lock so concurrent
+    return AuditStore().write_audit_entry(entry)
-    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]:
-    """Load all audit entries for the given component+slug. Empty
+    """Load all audit entries for the given component+slug."""
-    list if the log doesn't exist."""
+    return AuditStore().read_audit_entries(component, slug)
-    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 --------------------------------------------------------
@@ -433,35 +334,34 @@ def sha256_hex(content: str) -> str:
 class SupervisePlan:
     """Output of Supervise.prepare; consumed by .start.
 
-    `queue_dir` is the host directory bind-mounted into the sidecar
+    `db_path` is the host database bind-mounted into the sidecar at
-    at /run/supervise/queue. `internal_network` is empty at prepare
+    /run/supervise/bot-bottle.db. `internal_network` is empty at
-    time; the backend's launch step fills it via dataclasses.replace
+    prepare time; the backend's launch step fills it via
-    before calling .start."""
+    dataclasses.replace before calling .start."""
 
     slug: str
-    queue_dir: Path
+    db_path: Path
     internal_network: str = ""
 
 
 class Supervise(ABC):
-    """Per-bottle supervise sidecar. Encapsulates the host-side
+    """Per-bottle supervise sidecar. Encapsulates host-side database
-    prepare (queue dir staging); the sidecar's start/stop lifecycle
+    staging; the sidecar's start/stop lifecycle is backend-specific."""
-    is backend-specific."""
 
     def prepare(
         self,
         slug: str,
         stage_dir: Path,
     ) -> SupervisePlan:
-        """Stage the per-bottle queue dir on the host. Returns the
+        """Stage the host database. Returns the plan; `internal_network`
-        plan; `internal_network` must be set by the launch step before
+        must be set by the launch step before .start runs."""
-        .start runs."""
         del stage_dir
-        queue_dir = queue_dir_for_slug(slug)
+        db_path = host_db_path()
-        queue_dir.mkdir(parents=True, exist_ok=True)
+        QueueStore(slug)
+        AuditStore(db_path)
         return SupervisePlan(
             slug=slug,
-            queue_dir=queue_dir,
+            db_path=db_path,
         )
 
 # --- Helpers ---------------------------------------------------------------
@@ -474,47 +374,15 @@ def _require_str(raw: dict[str, object], key: str) -> str:
     return value
 
 
-def _atomic_write(path: Path, content: str, *, mode: int) -> None:
-    """Atomic: write to a sibling tmp file, fsync, rename."""
-    tmp = path.with_suffix(path.suffix + ".tmp")
-    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, mode)
-    try:
-        os.write(fd, content.encode("utf-8"))
-        os.fsync(fd)
-    finally:
-        os.close(fd)
-    os.replace(tmp, path)
-
-
-try:
-    import fcntl as _fcntl
-
-    def _try_flock(fd: int) -> None:  # type: ignore[reportRedeclaration]
-        try:
-            _fcntl.flock(fd, _fcntl.LOCK_EX)
-        except OSError:
-            pass
-
-    def _try_funlock(fd: int) -> None:  # type: ignore[reportRedeclaration]
-        try:
-            _fcntl.flock(fd, _fcntl.LOCK_UN)
-        except OSError:
-            pass
-except ImportError:  # pragma: no cover — Windows path
-    def _try_flock(fd: int) -> None:  # noqa: F841 — Windows fallback
-        return None
-
-    def _try_funlock(fd: int) -> None:  # noqa: F841 — Windows fallback
-        return None
-
-
 __all__ = [
     "ACTION_OPERATOR_EDIT",
     "AuditEntry",
+    "AuditStore",
     "COMPONENT_FOR_TOOL",
     "DEFAULT_POLL_INTERVAL_SEC",
+    "DB_PATH_IN_CONTAINER",
     "Proposal",
-    "QUEUE_DIR_IN_CONTAINER",
+    "QueueStore",
     "Response",
     "STATUSES",
     "STATUS_APPROVED",
@@ -536,8 +404,9 @@ __all__ = [
     "audit_dir",
     "audit_log_path",
     "bot_bottle_root",
+    "host_db_path",
     "list_pending_proposals",
-    "queue_dir_for_slug",
+    "list_all_pending_proposals",
     "read_audit_entries",
     "read_proposal",
     "read_response",

bot_bottle/supervise_server.py

@@ -7,14 +7,13 @@ config changes when stuck. The tools are `egress-allow`,
 Each queued tool call:
 
   1. Validates the proposed file syntactically.
-  2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
+  2. Writes a Proposal to the host SQLite database.
-     the host's ~/.bot-bottle/queue/<slug>/).
+  3. Blocks polling for a matching Response row.
-  3. Blocks polling for a matching Response file.
   4. Returns the operator's `{status, notes}` to the agent.
 
 The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
-container creation by the backend's start step). The queue dir comes
+container creation by the backend's start step). SUPERVISE_DB_PATH
-from SUPERVISE_QUEUE_DIR (default `/run/supervise/queue`).
+points at the bind-mounted host database.
 
 Speaks MCP over HTTP+JSON-RPC. Methods handled:
 
@@ -42,7 +41,6 @@ import typing
 import urllib.error
 import urllib.request
 from dataclasses import dataclass
-from pathlib import Path
 
 try:
     # Same-directory imports inside the bundle container; these files are
@@ -277,7 +275,6 @@ def validate_proposed_file(tool: str, content: str) -> None:
 @dataclass(frozen=True)
 class ServerConfig:
     bottle_slug: str
-    queue_dir: Path
     response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS
 
 
@@ -376,7 +373,7 @@ def handle_tools_call(
         current_file_hash=_sv.sha256_hex(proposed_file),
     )
     try:
-        _sv.write_proposal(config.queue_dir, proposal)
+        _sv.write_proposal(proposal)
     except OSError as e:
         raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e
     sys.stderr.write(
@@ -387,7 +384,7 @@ def handle_tools_call(
     deadline = time.monotonic() + config.response_timeout_seconds
     try:
         response = _sv.wait_for_response(
-            config.queue_dir,
+            config.bottle_slug,
             proposal.id,
             poll_interval=MIN_RESPONSE_POLL_INTERVAL_SECONDS,
             deadline=deadline,
@@ -399,7 +396,7 @@ def handle_tools_call(
             "isError": False,
         }
     try:
-        _sv.archive_proposal(config.queue_dir, proposal.id)
+        _sv.archive_proposal(config.bottle_slug, proposal.id)
     except OSError as e:
         raise _RpcInternalError(f"failed to archive proposal: {e}") from e
 
@@ -539,7 +536,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
 class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
     allow_reuse_address = True
     daemon_threads = True
-    config: ServerConfig = ServerConfig(bottle_slug="", queue_dir=Path())
+    config: ServerConfig = ServerConfig(bottle_slug="")
 
 
 # --- Entry point -----------------------------------------------------------
@@ -548,21 +545,18 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
 def serve(
     *,
     bottle_slug: str,
-    queue_dir: Path,
     port: int = _sv.SUPERVISE_PORT,
     bind: str = "0.0.0.0",
     response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
 ) -> typing.NoReturn:
-    queue_dir.mkdir(parents=True, exist_ok=True)
     server = MCPServer((bind, port), MCPHandler)
     server.config = ServerConfig(
         bottle_slug=bottle_slug,
-        queue_dir=queue_dir,
         response_timeout_seconds=response_timeout_seconds,
     )
     sys.stderr.write(
         f"supervise listening on {bind}:{port}; "
-        f"slug={bottle_slug!r}; queue={queue_dir}; "
+        f"slug={bottle_slug!r}; "
         f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n"  # type: ignore[arg-type]
     )
     sys.stderr.flush()
@@ -581,7 +575,6 @@ def main(argv: list[str]) -> int:
     if not bottle_slug:
         sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
         return 2
-    queue_dir = Path(os.environ.get("SUPERVISE_QUEUE_DIR", _sv.QUEUE_DIR_IN_CONTAINER))
     port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
     bind = os.environ.get("SUPERVISE_BIND", "0.0.0.0")
     try:
@@ -591,7 +584,6 @@ def main(argv: list[str]) -> int:
         return 2
     serve(
         bottle_slug=bottle_slug,
-        queue_dir=queue_dir,
         port=port,
         bind=bind,
         response_timeout_seconds=response_timeout_seconds,

docs/prds/prd-new-sqlite-local-storage.md

@@ -0,0 +1,135 @@
+# PRD prd-new: SQLite local storage
+
+- **Status:** Active
+- **Author:** codex
+- **Created:** 2026-07-01
+- **Issue:** #319
+
+## Summary
+
+Add a small stdlib SQLite storage layer for bot-bottle host runtime state,
+starting with the supervise queue and audit log. This replaces scattered JSON
+queue files and JSONL audit logs with structured tables while preserving the
+existing public supervise helper functions and sidecar queue mount contract.
+
+## Problem
+
+Bot-bottle currently stores supervise proposals and responses as individual JSON
+files under `~/.bot-bottle/queue/<slug>/`, and audit entries as JSONL files
+under `~/.bot-bottle/audit/`. That worked for the original interactive TUI, but
+new forge-native orchestration needs durable, queryable local state for queues,
+audit trails, watchdogs, and lifecycle records. PR #318 started introducing
+SQLite-shaped boilerplate for forge state; the storage foundation should live in
+its own PR so forge work can build on the shared runtime store instead of adding
+one-off persistence.
+
+## Goals / Success Criteria
+
+1. Supervise proposals and responses are persisted through SQLite.
+2. Audit entries are persisted through SQLite.
+3. Supervise queue helpers use the bottle slug / queue key instead of a queue
+   directory path.
+4. The sidecar receives the host database mount across docker, smolmachines,
+   and macOS-container backends.
+5. The implementation stays stdlib-only.
+6. Schema migrations use a `PRAGMA user_version` runner — no third-party deps.
+7. Unit tests cover queue round-trips, pending discovery, response waits,
+   archive semantics, audit round-trips, and path creation.
+
+## Non-goals
+
+- Migrating old JSON queue files or JSONL audit logs.
+- Adding forge orchestration state tables.
+- Adding egress metering or budget tables.
+- Changing the supervise TUI workflow or remediation behavior.
+- Introducing a third-party ORM or migration library.
+
+## Design
+
+### Database locations
+
+Queue and audit state use the host-level local database:
+
+```text
+~/.bot-bottle/bot-bottle.db
+```
+
+The supervise sidecar receives that database as a writable bind mount at
+`/run/supervise/bot-bottle.db` and gets the path through `SUPERVISE_DB_PATH`.
+No per-slug queue directory is mounted into the sidecar. This creates the shared
+host database that later forge/native lifecycle work can extend in separate
+PRDs.
+
+### Tables
+
+`supervise_proposals` lives in the host database:
+
+```sql
+CREATE TABLE supervise_proposals (
+    queue_key TEXT NOT NULL,
+    id TEXT NOT NULL,
+    bottle_slug TEXT NOT NULL,
+    tool TEXT NOT NULL,
+    proposed_file TEXT NOT NULL,
+    justification TEXT NOT NULL,
+    arrival_timestamp TEXT NOT NULL,
+    current_file_hash TEXT NOT NULL,
+    archived INTEGER NOT NULL DEFAULT 0,
+    PRIMARY KEY (queue_key, id)
+);
+```
+
+`supervise_responses` lives in the host database:
+
+```sql
+CREATE TABLE supervise_responses (
+    queue_key TEXT NOT NULL,
+    proposal_id TEXT NOT NULL,
+    status TEXT NOT NULL,
+    notes TEXT NOT NULL,
+    final_file TEXT,
+    archived INTEGER NOT NULL DEFAULT 0,
+    PRIMARY KEY (queue_key, proposal_id)
+);
+```
+
+`supervise_audit_entries` lives in the host database:
+
+```sql
+CREATE TABLE supervise_audit_entries (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp TEXT NOT NULL,
+    bottle_slug TEXT NOT NULL,
+    component TEXT NOT NULL,
+    operator_action TEXT NOT NULL,
+    operator_notes TEXT NOT NULL,
+    justification TEXT NOT NULL,
+    diff TEXT NOT NULL
+);
+```
+
+### Compatibility
+
+The queue helpers take a bottle slug / queue key and perform equivalent
+operations against `~/.bot-bottle/bot-bottle.db`:
+
+- `list_pending_proposals` returns non-archived proposals without a non-archived
+  response, sorted by arrival time.
+- `archive_proposal` marks matching proposal/response rows archived instead of
+  moving files into `processed/`.
+- `wait_for_response` keeps the current polling behavior but polls SQLite.
+
+The old audit path helpers (`audit_dir`, `audit_log_path`) stay available for
+compatibility. `audit_log_path` no longer describes the active storage location;
+callers should use `read_audit_entries`.
+
+## Implementation chunks
+
+1. Add SQLite store helpers for supervise queue and audit state.
+2. Rewire `bot_bottle.supervise` queue/audit functions to the store.
+3. Update supervise CLI discovery tests and queue/audit unit tests.
+4. Run unit tests, pyright, and pylint for touched modules.
+
+## Open questions
+
+None.

docs/research/monetization-positioning.md

@@ -1,490 +0,0 @@
-# Monetization & competitive positioning
-
-Where, if anywhere, bot-bottle has a paid wedge — given a 2026
-competitive field that has largely commoditized "sandbox a coding
-agent." Folds together the agent-provider-agnostic framing, the Fly
-remote-backend idea, the supervisor/egress-audit play, and the
-solo-dev/Linux brand instinct, then asks the only question that
-matters: is there a viable path to revenue that the competition does
-not already foreclose?
-
-Companion to
-[`agent-sandbox-landscape.md`](agent-sandbox-landscape.md) (the
-isolation-tech survey),
-[`built-in-supervisor-design.md`](built-in-supervisor-design.md) (the
-supervise surface this would extend), and
-[`secret-minimization-over-dlp.md`](secret-minimization-over-dlp.md)
-(why custody, not detection, is the real moat).
-
-Market data current as of June 2026.
-
-## Summary
-
-**Verdict: a path exists, but it is narrow, and it is not the path the
-project is currently shaped for.** Every individual property bot-bottle
-leans on — isolation, BYO-image, egress filtering, OSS, self-hosting —
-is matched by some competitor, and several are now *free* from the agent
-vendors themselves. There is exactly one defensible position left: the
-**bundle** that no single competitor occupies —
-
-> uniform egress audit + secret custody + policy, across *heterogeneous
-> coding agents you don't trust*, on your infra or a managed pool.
-
-Monetization is viable **only** if the product is sold as cross-vendor
-**fleet governance + egress audit for teams**, not as solo-dev agent
-safety (which the labs give away free). The solo-dev/Linux/anti-corporate
-energy is real and worth using — but as a *distribution and trust*
-engine that drives bottom-up adoption into teams, never as the revenue
-positioning itself. Get those two wires crossed and the business dies:
-you'd be courting the lowest-willingness-to-pay audience on earth while
-repelling the only buyer who pays.
-
-Net: **viable, conditional, and unforgiving of positioning error.** Do
-Phase 1 (self-hostable egress-audit dashboard) regardless — it's
-low-risk and it's the demo that makes everything else legible. Gate the
-go/no-go on whether 5–10 teams confirm they'd pay for cross-vendor
-egress audit *before* building the hosted tier.
-
-## The two axes of "agnostic"
-
-bot-bottle differentiates on two orthogonal axes, and conflating them
-muddies the pitch:
-
-1. **Agent-provider agnostic** — run Claude Code, Codex, Aider, a local
-   model, behind one control layer. Already real in the code
-   (`agent_provider.py`, Claude/Codex templates, BYO Dockerfile). This
-   is the axis the labs *structurally cannot* match — Anthropic only
-   runs Claude, OpenAI only their models. Durable.
-2. **Compute backend** — local (docker / Apple Container / smolmachines)
-   today; a remote **Fly** backend would add a managed pool. This is the
-   axis that makes "fleet" literal for orgs and opens metered billing.
-   Fly is a strong first remote backend because it also subsumes remote
-   spin-up (Machines API) and the tunnel problem (6PN/WireGuard) — but
-   "provider-agnostic compute" should be *earned* after backend #2, not
-   designed up front (premature generalization trap).
-
-## Competitive field, by capability
-
-The field doesn't have one competitor; it has a different set on each
-capability bot-bottle touches. Five dimensions:
-
-| Capability | Who has it | bot-bottle's standing |
-| :-- | :-- | :-- |
-| **Isolation / sandbox** | Anthropic & OpenAI **native, free**; OSS devcontainer wrappers; E2B/Modal/Daytona/Northflank | Commoditized. Not a wedge. |
-| **Arbitrary BYO Docker image** | Sandbox PaaS (E2B/Modal/Daytona/Northflank) yes; **managed agents: ~none** (Codex = fixed `codex-universal` + setup scripts; Copilot "not supported"; Devin/Jules constrained) | Wedge **vs. managed agents** (structural: it's their infra). Table stakes vs. PaaS. |
-| **Egress audit + alerts** | LLM-observability tools (Braintrust/Langfuse/Phoenix/Helicone/Datadog) — but on *model calls*, wrong layer. Network-egress security (DeepInspect, AI gateways) — right layer, but decoupled from the agent, not cross-vendor. Sandbox PaaS = gateway/filter, not an audit surface. | **~Nobody in bot-bottle's exact shape** (per-agent egress, tied to the sandbox, with DLP context, cross-vendor). This is the wedge. |
-| **OSS / self-hosting** | Managed agents: ~none. Sandbox PaaS: ~half (E2B OSS+self-host; Northflank BYOC; Modal closed; **Daytona leaving OSS**). Devcontainer wrappers: ~all. Observability: several. | Real wedge **vs. managed agents only**. Table stakes vs. PaaS, zero differentiation vs. wrappers. |
-| **Cross-vendor uniformity** | Nobody — the labs won't, PaaS is agent-neutral infra not agent-aware control, wrappers are single-tool | Wedge. The connective tissue of the whole position. |
-
-The pattern: **isolation and OSS/self-host are commodity; BYO-image and
-cross-vendor are wedges only against the managed agents; egress-audit in
-the integrated form is the one thing genuinely unoccupied.**
-
-## Where bot-bottle is alone vs. where it's table stakes
-
-- **Alone (the moat):** egress audit + secret custody + policy, *tied to
-  the agent sandbox*, *with DLP context* (which secret, which host,
-  which agent/task), *uniform across vendors*. No competitor bundles
-  these. An enterprise *could* bolt DeepInspect-style egress monitoring
-  onto a sandbox, so the defensibility is the **integration and
-  per-agent context**, not "we can see egress."
-- **Table stakes (do not lead with these):** "we sandbox agents" (free
-  from the labs), "we're open source" (E2B is; the wrapper crowd all
-  is), "we self-host" (Northflank BYOC, E2B, every wrapper).
-
-## The two existential competitive facts
-
-1. **The agent vendors ship good-enough sandboxing for free.** Claude
-   Code now has Seatbelt/bubblewrap + a network proxy natively; Codex
-   has its own sandbox + approvals. This compresses the *single-vendor,
-   single-dev* market to ~zero willingness-to-pay. It is *why* the
-   product must be cross-vendor fleet governance, not local agent
-   safety.
-2. **Northflank is converging from the infra side.** It already ships
-   dedicated egress gateways + proxy-based secret injection + BYOC.
-   It is the nearest thing to bot-bottle's differentiator as a managed
-   platform — but infra-first and agent-neutral, not agent-aware,
-   cross-vendor, or audit-first. Watch it.
-
-## Monetization path (sequenced)
-
-Open-core: **give away the sandbox, charge for the control plane.**
-
-- **Phase 0 — validate (1–2 wks, parallel).** Ask 5–10 teams running 2+
-  agents: would you pay for one egress-audit + policy plane across
-  Claude *and* Codex? Gate the rest on a yes.
-- **Phase 1 — the wedge (self-hostable, OSS).** Multi-bottle egress
-  dashboard + web approval queue + exportable audit log, built over the
-  existing `supervise_server.py` JSON-RPC and the egress event levels
-  (`LOG_BLOCKS` / `LOG_FULL`). Low risk, half-built, and the 30-second
-  demo that sells everything. The compliance hook (75% of enterprises
-  rank auditability #1) lives here.
-- **Phase 2 — the paywall (hosted team tier).** Multi-tenant supervisor:
-  SSO/RBAC, audit retention, alerting, **centralized policy push**
-  (define egress allowlist + DLP once, enforce across all agents —
-  the moat made concrete). Gate on team/compliance features, *never* on
-  the core security.
-- **Phase 3 — Fly remote backend.** Managed agent pool → "fleet" becomes
-  literal; metered (agent-hours) billing; subsumes remote spin-up +
-  tunnel.
-- **Phase 4 — deepen.** Second agent provider done deeply (lean
-  open-source/open-weight for rug-pull resistance); egress anomaly
-  detection (the DLP stream becomes a product); SOC2/audit-export for
-  larger buyers.
-
-**Do not build first:** the p2p mobile app (least monetizable, 6PN
-gives the tunnel free), a generic multi-cloud abstraction (premature),
-or the hosted SaaS before Phase 0.
-
-## Brand vs. revenue: the solo-dev / Linux instinct
-
-The instinct to court Linux/hacker/solo-dev users and stay "not too
-corporate" is **right for distribution, dangerous as strategy.**
-
-- **Right:** it's how OSS infra gets discovered and trusted (HN, stars,
-  word-of-mouth, security-circle vouching); authenticity is a real moat
-  vs. the corporate players *because the architecture sincerely embodies
-  it* (local-first, `$HOME` trust boundary, no phone-home); and it fits
-  the founder.
-- **Dangerous:** that audience is the lowest-WTP cohort that exists
-  (self-hosts the free thing, forks rather than pays), and "not too
-  corporate" reads to a VP of Eng as "not enterprise-ready." Building an
-  anti-SaaS brand and then shipping a paid tier invites the sell-out /
-  rug-pull backlash — which **Daytona just triggered** going closed.
-
-**Resolution — be Tailscale, not a manifesto.** Use the developer-first,
-respects-you energy as the *funnel*; sell *through* the solo advocate,
-bottom-up, into the team that pays. Two guardrails:
-
-1. "Anti-corporate" must not mean "anti-team-features." SSO/RBAC/audit
-   retention *are* the monetization; build them in a developer-respecting
-   way (Tailscale has SSO and is still beloved). Tone is the brand; team
-   features are the product.
-2. Set the open-core social contract publicly **on day one** — core
-   sandbox open and self-hostable forever; hosted control plane is how
-   the lights stay on. The communities that don't revolt are the ones
-   told the deal upfront.
-
-Concrete: the README frames the Docker/**Linux** backend as "legacy."
-If courting the Linux crowd, make the Linux path (Docker+gVisor,
-libkrun/smolmachines) first-class in the docs, not the fallback.
-
-## Individuals, mobile, and the Pi-ecosystem reality check
-
-"Individual devs won't pay" (above) is too blunt and needs refining.
-The accurate claim: individuals won't pay for **safety-as-insurance**
-(abstract risk reduction the labs give away free), but they *do* pay for
-**capability/convenience felt daily** — Claude Pro, Cursor, Tailscale
-Personal. "Drive my self-hosted agent from my phone" is capability, not
-insurance, so it has a real (low-priced, high-churn) WTP profile. The
-self-hoster/Linux crowd specifically pays for **sovereignty/control**,
-just not for enterprise insurance. So an individual "sovereign remote
-agent access" tier is *not* unreasonable in principle.
-
-**But the market has already run that experiment, in public, for free.**
-The Pi ecosystem (pi.dev) has commoditized every convenience layer an
-individual product would charge for:
-
-| Capability | Already free/OSS | bot-bottle differentiates? |
-| :-- | :-- | :-- |
-| Remote control from mobile | remote-pi, Paseo, TelePi | ❌ commoditized |
-| Multi-agent orchestration from mobile | Paseo, pi-agent-dashboard | ❌ commoditized |
-| **Launch** new agents from mobile | Paseo (`paseo run`) | ❌ commoditized |
-| Launch into a **sandboxed, egress-audited** env | nobody | ✅ the moat |
-
-Paseo (`getpaseo/paseo`, on the App Store) does the full thing an
-individual remote-control tier would charge for — launch *and* attach
-agents on a laptop/VM/dev-server, driven from mobile over an E2E relay —
-free and open source. It *orchestrates* agents; it does **not** sandbox them, run
-an egress chokepoint, DLP-scan, or audit. None of the Pi-ecosystem tools
-do. So the residue, yet again, is **isolation + governance**, not
-remote/launch convenience.
-
-Two takeaways:
-1. **Don't compete on orchestration/launch/remote UX** — it's a solved,
-   free, fast-moving, App-Store-shipping space around Pi. You won't win
-   it and it isn't the moat.
-2. **Be the safe runtime orchestrators launch *into*.** Launch-from-mobile
-   is table stakes; *launch-into-a-sealed-egress-audited-bottle* is the
-   differentiator. bot-bottle is the sandbox an orchestrator like Paseo
-   would target, or that you wrap thin orchestration around — never the
-   orchestrator itself.
-
-Capability layers commoditize fast: every individual/mobile angle
-probed in this analysis collapsed back to the same cross-vendor +
-sandbox + egress-audit + custody bundle. Mobile remote belongs as a
-*funnel delighter* on top of the team product, not a standalone paid
-line.
-
-## Forge-native orchestration as the delivery vehicle
-
-The strongest concrete *product shape* for the moat is not a bespoke
-dashboard and not a Paseo competitor — it is **the git forge as the
-orchestrator, with bot-bottle as the safe runtime it launches into.**
-The forge already provides, for free, everything an orchestrator would
-otherwise have to build: identity (agent/bot users, signed commits),
-state (issues, labels, PRs/MRs, comments), triggers (webhooks, CI,
-comment commands), review (diffs, approvals, status checks), audit
-(commits/comments/reviews), and permissions (repo access, protected
-branches, token scopes). bot-bottle supplies the one thing the forge
-doesn't: **least-privilege, secret-isolated, audited execution of
-untrusted agents.** Same moat (custody + audit + policy), better
-vehicle — and it lands the product where teams already live, so it
-avoids building an agent dashboard before one is needed.
-
-The flow is essentially free to assemble:
-
-```
-issue/PR/MR event → webhook → policy/router → assign agent user +
-branch/worktree → run agent in an isolated bottle (no ambient secrets)
-→ commit as agent identity → open PR/MR → CI + human review + merge
-```
-
-**Crowding (why this is less saturated than it looks):**
-
-| Layer | How crowded |
-| :-- | :-- |
-| Generic multi-agent orchestrators (worktree/TUI/dashboard) | very — 50–100+ |
-| Forge-native issue/PR/MR orchestration | moderate — ~10–30 serious |
-| Self-hostable, least-privilege, audited, forge-portable | **single digits** |
-
-The deeper you go toward *untrusted-agent safety + auditability +
-self-hostable + forge-portable*, the emptier it gets.
-
-**The GitHub/GitLab first-party trap → lead Gitea + sovereignty.**
-GitHub (Agentic Workflows, Copilot coding agent) and GitLab (Duo Agent
-Platform) are the forge *vendors* building native issue-to-PR agent
-orchestration with native identity/permissions/audit. On their turf you
-lose the integration-depth battle the same way single-vendor agent
-safety loses to Anthropic/OpenAI — the same "incumbent ships it free,
-deeper" dynamic, one layer up. So the durable opening is **Gitea +
-self-hosted** (no first-party agent platform exists — the open Gitea
-feature request for an AI code agent confirms the vacuum) plus
-**cross-forge *untrusted-agent* safety**, which no forge vendor will
-build because they want you running *their* agent, not arbitrary ones
-under uniform least-privilege across competitors' forges. Cross-vendor
-neutrality, applied to forges.
-
-**Buyer reconciliation.** The least-crowded opening (self-hosted Gitea)
-overlaps the lowest-WTP crowd (indie self-hosters), while the paying
-teams sit on GitHub/GitLab where first-party competition is fiercest.
-The intersection that resolves it: **orgs running self-hosted forges for
-sovereignty/compliance reasons** (regulated, air-gapped, security-
-conscious, on-prem). They have budget, they run self-hosted GitLab/Gitea,
-*and* shipping code to a cloud agent vendor is a non-starter — so "run
-untrusted agents sandboxed, least-privilege, fully audited, inside our
-forge, on our infra" is a procurement checkbox, not a nicety. That is
-where "least-crowded" finally meets "has money."
-
-**Separate moat-hard-parts from cost-hard-parts.** The orchestration
-"hard parts" are two different things, and conflating them oversells the
-fit:
-
-| Moat (your differentiated strength) | Undifferentiated cost (everyone faces) |
-| :-- | :-- |
-| permission isolation | idempotency / dedupe / run ledger |
-| secret handling under malicious prompts | concurrency, locks, cancellation |
-| run provenance | queueing / scheduling / cleanup |
-| policy language | merge-conflict handling (~27% agent-PR conflict rate) |
-
-The right column is generic distributed-systems plumbing that wins you
-nothing and that merge-conflict resolution especially is a *different
-competency* from sandbox/custody. Keep it thin in the MVP; do not build a
-policy DSL + durable ledger + conflict resolver before one org pays.
-
-**The killer feature: run provenance on every agent PR.** A check/comment
-answering — which agent, which model, which prompt, which base commit,
-which policy, which tools, which network egress, which test results —
-attached at the moment a human reviews. It renders the (invisible)
-custody + egress-audit work as a PR artifact the buyer sees at the exact
-trust-decision point. No forge vendor's first-party agent will show you
-"here is everything the untrusted agent could reach." Build this first.
-
-**MVP** (`@bot-bottle fix this`): create an isolated worktree/bottle →
-check out the issue branch → run the selected harness as a named agent
-user → deny ambient secrets by default → record prompt/model/tools/policy
-→ commit with bot identity → open PR/MR → attach the run-provenance
-footer (log + tests + permission/egress summary) → require human merge.
-The security model *is* the product. This rides the headless launch
-primitive directly: webhook → `start --headless` into an isolated bottle
-→ commit as agent identity → PR with provenance.
-
-Open-core line, refined in the next section: the trigger *convention*
-(label/assignee) stays open so anyone can adopt it, but the
-**orchestrator that receives webhooks and governs lifecycle is the paid
-control plane**; the runtime — and a signed-provenance emission API —
-stay free.
-
-## The open/paid boundary, refined: orchestrator as the paid control plane
-
-The forge-native shape sharpens the open-core line past the rough
-"trigger free, execution paid" cut above. Working it through four
-constraints — value capture, provenance integrity, the sovereignty
-buyer, and what the forge *structurally cannot do* — yields a precise
-boundary.
-
-**The orchestrator is the control plane, and the control plane is the
-paid product.** With the forge supplying identity / state / triggers /
-review, bot-bottle's orchestrator (`bot-bottle-orchestrator`, already
-specced as a separate binary in the forge-native PRD) is where webhooks
-land and bottle lifecycle + governance live. That binary can stay
-**closed/private from day one** without breaking the open-core contract:
-the runtime stays OSS; the control plane is how the lights stay on. This
-is "give away the sandbox, charge for the control plane" made literal —
-the orchestrator *is* the control plane.
-
-**Charge for the moat, not the webhook.** Holding webhooks and managing
-bottle lifecycle is commodity — the forge vendors build it first-party,
-and it's the "undifferentiated cost" column above (idempotency, queueing,
-dispatch). If the pitch is "we catch the webhook," they out-build it
-free. The paid value is the two things the forge *cannot* do:
-
-1. **See inside the run** — which model / prompt / policy / tools / egress
-   produced the diff, whether a secret nearly left. Runtime-level data
-   only the bottle holds.
-2. **Aggregate and enforce across runs** — retain / search / export every
-   run across every repo; push one egress/DLP/capability policy
-   fleet-wide and detect drift.
-
-The explainable heuristic: **anything legible within a single run on a
-single node is free; anything requiring cross-run aggregation, central
-enforcement, or identity/fleet management is paid.** That is also the
-individual-vs-team line — individuals live in single runs, teams need the
-aggregate.
-
-**Provenance: emit free (signed), sell the product.** The forge is the
-wrong system of record for provenance — a markdown footer is mutable by
-any maintainer, unsigned, per-PR, with no aggregation, so a maintainer
-could simply edit it. The authoritative record therefore lives in the
-(paid) control plane. The *runtime* emits **signed** provenance through a
-**free API** — tamper-evident offline (edit it and the signature breaks;
-verify with no server), so on-prem teams can route it into their own
-SIEM. What's paid is the *product* over that stream: retention, search,
-cross-run, export, policy. Whether a copy also lands in the PR footer is
-an optional, off-by-default marketing dial — one consumer of the free
-API, not a free provenance surface, and never the audit record. The
-mutability "bug" becomes a paid feature: the control plane flags *"PR
-footer edited / doesn't match the signed run."* (Prometheus model:
-`/metrics` is free to scrape; managed retention + dashboards are the
-business.)
-
-**On-prem priority: self-hosted runners over self-hosted provenance.**
-The sovereignty buyer's *hard structural constraint* is where the agent
-**executes** against private code, secrets, and network — that's the
-runner, and it cannot leave the perimeter. Audit metadata is softer; many
-regulated orgs ship logs to SaaS while keeping the workload inside. So:
-
-- Self-hosted **runner** = baseline, always, for that buyer.
-- Self-hosted **provenance store** = premium tier of the strictest subset
-  (air-gapped, hard data-residency) — and largely covered by the free
-  emission API → their own SIEM, so it may never need to be a product you
-  build.
-- Precision so you don't trip your own free tier: a single self-hosted
-  runner *is the OSS runtime on their box* — free. What's paid is the
-  **fleet control plane**: enrolling/managing many runners, central
-  policy push, dispatch/identity/quota, health/scaling. You don't sell
-  "a runner," you sell **running a governed fleet**.
-
-**Resulting tiers:**
-
-| Layer | What it is | Open/Paid | Deployment |
-| :-- | :-- | :-- | :-- |
-| **Runtime** | isolation + ephemeral bottles, cred-proxy, supervise, `start --headless`, signed-provenance emission API | Free / OSS | Always self-host |
-| **Single runner** | the OSS runtime on a box | Free / OSS | Self-host |
-| **Control plane** | cross-run audit retention/search/export, central policy push, SSO/RBAC dispatch, fleet management of runners, alerting | **Paid** | Hosted *or* self-host-licensed — same code |
-| **Capacity** | managed Fly runner pool, metered (agent-hours) | **Paid add-on** | Hosted only |
-
-Fly stays a **capacity/convenience line, not the moat** — it monetizes
-even solo hackers (capability, not insurance), but a managed runner pool
-is reselling compute against Fly/E2B/Northflank on price. It's a bundle
-attached to the governance, never the thing defended. Self-host is *not*
-a separate product: on-prem buyers get the same closed control plane,
-licensed, pointed at their own runners.
-
-## Risks to the thesis
-
-- **Lab encroachment.** If Anthropic/OpenAI add cross-agent governance
-  or open their managed egress logs, the wedge narrows. Mitigate by
-  going deep on cross-vendor + custody + audit *now*, while they're
-  single-vendor.
-- **Rug-pull dependency.** You run the labs' agents; they can restrict
-  their agent to their own sandbox via ToS/tech. Hedge toward
-  open-source/open-weight agents for durability.
-- **Northflank (or E2B) ships agent-aware audit.** Plausible from the
-  infra side. Your defense is agent-awareness + the supervise approval
-  loop + cross-vendor, not raw egress visibility.
-- **WTP may simply not be there.** The honest failure mode: teams like
-  the audit but won't pay because "we already sandbox in CI." Phase 0
-  exists to find this out cheaply before building Phase 2/3.
-- **Forge-vendor encroachment (forge-native path).** GitHub Agentic
-  Workflows / Copilot and GitLab Duo are first-party and deepening.
-  Defense: aim at self-hosted Gitea + sovereignty buyers where no
-  first-party agent platform exists, and at cross-forge untrusted-agent
-  neutrality the vendors won't build. Don't fight them GitHub-native.
-- **Orchestration-reliability scope creep.** The forge-native build
-  drags in idempotency, queueing, concurrency, and merge-conflict
-  handling — undifferentiated plumbing that isn't the moat. Keep it thin
-  until a paying org forces it.
-
-## Recommendation
-
-Build Phase 1 now — it's low-risk, half-built, and the proof artifact.
-Run Phase 0 in parallel. Treat a clear yes from 5–10 teams as the
-green light for the hosted tier; treat a soft maybe as a signal to stay
-an excellent OSS tool with a tip-jar/support model rather than a
-venture-shaped SaaS. The technology is not the risk — the codebase is
-exemplary and the architecture already supports the pivot. The risk is
-**positioning discipline**: sell cross-vendor fleet governance to teams,
-use the indie brand as the funnel, and never let the anti-corporate
-aesthetic veto the features that pay.
-
-## Sources
-
-- Anthropic — Claude Code sandboxing:
-  https://www.anthropic.com/engineering/claude-code-sandboxing
-- OpenAI Codex — cloud environments:
-  https://developers.openai.com/codex/cloud/environments ;
-  custom-image feature request:
-  https://community.openai.com/t/feature-request-custom-docker-images/1265333
-- GitHub Copilot — custom container image (not supported), discussion
-  #194105: https://github.com/orgs/community/discussions/194105
-- DeepInspect — AI egress monitoring:
-  https://www.deepinspect.ai/blog/ai-egress-monitoring
-- Braintrust — AI agent observability/alerting:
-  https://www.braintrust.dev/articles/best-ai-agent-observability-tools-2026
-- E2B (OSS, Apache-2.0): https://github.com/e2b-dev/e2b ;
-  infra/self-host: https://github.com/e2b-dev/infra
-- Daytona going closed source:
-  https://www.daytona.io/dotfiles/updates/daytona-is-going-closed-source
-- Northflank — BYOC / egress gateways:
-  https://northflank.com/blog/what-is-byoc-in-cloud-computing ;
-  https://northflank.com/blog/self-hostable-alternatives-to-e2b-for-ai-agents
-- Modal Sandboxes: https://modal.com/products/sandboxes
-- AI agent orchestration / enterprise governance (75% cite
-  auditability):
-  https://viston.tech/ai-agent-orchestration-in-2026-moving-from-pilots-to-enterprise-wide-execution/
-- Pi harness (provider-agnostic CLI): https://pi.dev/packages/remote-pi ;
-  https://github.com/earendil-works/pi
-- Paseo (launch + attach agents from desktop/mobile, OSS):
-  https://github.com/getpaseo/paseo ;
-  https://apps.apple.com/us/app/paseo-remote-coding-agents/id6758887924
-- pi-agent-dashboard (mobile-first remote control via mDNS/zrok):
-  https://github.com/BlackBeltTechnology/pi-agent-dashboard
-- TelePi (Telegram remote control for Pi):
-  https://futurelab.studio/blog/telepi-telegram-remote-control-for-pi/
-- Forge-native landscape (provided via conversation, not independently
-  re-verified):
-  - awesome-agent-orchestrators (50+ generic orchestrators):
-    https://github.com/andyrewlee/awesome-agent-orchestrators
-  - GitHub Agentic Workflows (first-party repo automation):
-    https://github.blog/ai-and-ml/automate-repository-tasks-with-github-agentic-workflows/
-  - GitLab Duo Agent Platform GA:
-    https://ir.gitlab.com/news/news-details/2026/GitLab-Announces-the-General-Availability-of-GitLab-Duo-Agent-Platform/default.aspx
-  - ai-review (cross-forge review incl. Gitea):
-    https://github.com/Nikita-Filonov/ai-review
-  - Gitea feature request — AI code agent (the vacuum):
-    https://github.com/go-gitea/gitea/issues/34527
-  - Phoenix — safe GitHub issue resolution (label-based webhook state
-    machine): https://arxiv.org/abs/2606.20243
-  - AgenticFlict — ~27% merge-conflict rate in agent PRs:
-    https://arxiv.org/abs/2604.03551

tests/unit/test_cli_start_headless.py

@@ -0,0 +1,188 @@
+"""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()

tests/unit/test_compose.py

@@ -107,7 +107,7 @@ def _egress_plan(
 def _supervise_plan() -> SupervisePlan:
     return SupervisePlan(
         slug=SLUG,
-        queue_dir=STATE / "supervise" / "queue",
+        db_path=STATE / "bot-bottle.db",
         internal_network=f"bot-bottle-net-{SLUG}",
     )
 
@@ -392,7 +392,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         sc = self._render(supervise=True)["services"]["sidecars"]
         env_strings = sc["environment"]
         self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings)
-        self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
+        self.assertIn("SUPERVISE_DB_PATH=/run/supervise/bot-bottle.db", env_strings)
         self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))
 
     def test_volumes_always_includes_egress_ca(self):
@@ -408,8 +408,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertIn("/etc/egress", targets)
         self.assertIn("/git-gate-entrypoint.sh", targets)
         self.assertIn("/git-gate/creds/upstream-known_hosts", targets)
-        self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
+        self.assertIn("/run/supervise/bot-bottle.db", targets)
-                            for t in targets))
 
     def test_extra_hosts_omitted_for_git_upstreams(self):
         sc = self._render(with_git=True)["services"]["sidecars"]

tests/unit/test_contrib_claude_provider.py

@@ -74,7 +74,7 @@ def _plan(
     if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
-            queue_dir=Path("/tmp/queue"),
+            db_path=Path("/tmp/bot-bottle.db"),
         )
     return DockerBottlePlan(
         spec=spec,
@@ -343,5 +343,14 @@ 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__":
     unittest.main()

tests/unit/test_contrib_codex_provider.py

@@ -77,7 +77,7 @@ def _plan(
     if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
-            queue_dir=Path("/tmp/queue"),
+            db_path=Path("/tmp/bot-bottle.db"),
         )
     return DockerBottlePlan(
         spec=spec,
@@ -314,5 +314,14 @@ 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__":
     unittest.main()

tests/unit/test_contrib_pi_provider.py

@@ -223,5 +223,14 @@ class TestPiDockerfile(unittest.TestCase):
         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__":
     unittest.main()

tests/unit/test_docker_provision_git_user.py

@@ -38,6 +38,7 @@ class _Provider(AgentProvider):
     def provision_prompt(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 headless_prompt(self, prompt): return []  # type: ignore[override]
 
 
 _PROVIDER = _Provider()

tests/unit/test_egress_addon_log_redaction.py

@@ -47,7 +47,6 @@ def _addon() -> EgressAddon:
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = Config(routes=(), log=LOG_FULL)
     a.safe_tokens = set()
-    a._supervise_queue_dir = ""
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     return a

tests/unit/test_egress_addon_request_flow.py

@@ -212,7 +212,6 @@ def _addon(config: Config) -> EgressAddon:
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = config
     a.safe_tokens = set()
-    a._supervise_queue_dir = ""
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     a.routes_path = "/nonexistent/routes.yaml"
@@ -386,10 +385,10 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
     def _sha256_hex(_payload: Any) -> str:
         return "hash"
 
-    def _noop(_a: Any, _b: Any) -> None:
+    def _noop(*_args: Any) -> None:
         return None
 
-    def _read_response(_qd: Any, _pid: Any) -> Any:
+    def _read_response(_slug: Any, _pid: Any) -> Any:
         if response_status is None:
             raise OSError("not written yet")  # forces poll -> timeout
         return types.SimpleNamespace(status=response_status)
@@ -409,7 +408,6 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
 class TestSuperviseBranch(unittest.TestCase):
     def _supervised_addon(self) -> EgressAddon:
         addon = _addon(Config(routes=(Route(host="api.example.com"),)))
-        addon._supervise_queue_dir = "/tmp/egress-queue"
         addon._supervise_slug = "test-bottle"
         addon._token_allow_timeout = 0.05
         return addon
@@ -632,14 +630,13 @@ class TestRedactSurfaces(unittest.TestCase):
 class TestSuperviseWriteFailure(unittest.TestCase):
     def test_write_proposal_oserror_blocks(self) -> None:
         addon = _addon(Config(routes=(Route(host="api.example.com"),)))
-        addon._supervise_queue_dir = "/tmp/egress-queue"
         addon._supervise_slug = "test-bottle"
         addon._token_allow_timeout = 0.05
         flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}"))
 
         fake = _fake_sv("approved")
 
-        def _raise(_qd: Any, _p: Any) -> None:
+        def _raise(_p: Any) -> None:
             raise OSError("disk full")
 
         fake.write_proposal = _raise

tests/unit/test_git_gate.py

@@ -14,6 +14,7 @@ from bot_bottle.git_gate import (
     git_gate_render_access_hook,
     git_gate_render_entrypoint,
     git_gate_render_hook,
+    provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
     _resolve_identity_file,
     git_gate_upstreams_for_bottle,
@@ -209,8 +210,9 @@ class TestHookRender(unittest.TestCase):
         # the suppressed findings for human approval.
         self.assertIn("--ignore-gitleaks-allow", hook)
         self.assertIn("--report-format=json", hook)
-        self.assertIn('"tool": "gitleaks-allow"', hook)
+        self.assertIn("tool=_sv.TOOL_GITLEAKS_ALLOW", hook)
-        self.assertIn("SUPERVISE_QUEUE_DIR", hook)
+        self.assertIn("_sv.write_proposal", hook)
+        self.assertIn("_sv.read_response", hook)
         self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
         self.assertIn("supervisor approved # gitleaks:allow", hook)
         self.assertIn("supervisor rejected # gitleaks:allow", hook)
@@ -371,6 +373,27 @@ class TestDynamicKeyProvisioning(unittest.TestCase):
             self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
         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):
         revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage)
 

tests/unit/test_macos_container_launch.py

@@ -71,7 +71,9 @@ def _plan(
     else:
         git_gate_plan = SimpleNamespace(upstreams=())
     supervise_plan = (
-        SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
+        SimpleNamespace(
+            db_path=Path("/state/bot-bottle.db"),
+        )
         if supervise else None
     )
     agent_provision = SimpleNamespace(
@@ -137,7 +139,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
             argv,
         )
         self.assertIn(
-            "type=bind,source=/state/supervise/queue,target=/run/supervise/queue",
+            "type=bind,source=/state/bot-bottle.db,target=/run/supervise/bot-bottle.db",
             argv,
         )
 

tests/unit/test_smolmachines_provision.py

@@ -49,6 +49,7 @@ class _Provider(AgentProvider):
     def provision_prompt(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 headless_prompt(self, prompt): return []  # type: ignore[override]
 
 
 _PROVIDER = _Provider()
@@ -129,7 +130,7 @@ def _plan(
     if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
-            queue_dir=Path("/tmp/queue"),
+            db_path=Path("/tmp/bot-bottle.db"),
         )
     return SmolmachinesBottlePlan(
         spec=spec,
@@ -421,6 +422,14 @@ class TestBundleLaunchSpec(unittest.TestCase):
             spec.environment,
         )
 
+    def test_supervise_adds_daemon_volume_and_env(self):
+        from bot_bottle.supervise import DB_PATH_IN_CONTAINER
+        plan = _plan(supervise=True)
+        spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
+        self.assertIn("supervise", spec.daemons_csv)
+        self.assertIn(f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", spec.environment)
+        self.assertIn(("/tmp/bot-bottle.db", DB_PATH_IN_CONTAINER, False), spec.volumes)
+
     def test_canary_env_visible_to_smolvm_guest(self):
         plan = _plan(canary=True)
         with patch.object(

tests/unit/test_supervise.py

@@ -1,6 +1,5 @@
 """Unit: supervise queue + audit log + diff helpers (PRD 0013)."""
 
-import json
 import tempfile
 import threading
 import time
@@ -19,7 +18,7 @@ from bot_bottle.supervise import (
     TOOL_EGRESS_ALLOW,
     TOOL_GITLEAKS_ALLOW,
     archive_proposal,
-    audit_log_path,
+    host_db_path,
     list_pending_proposals,
     read_audit_entries,
     read_proposal,
@@ -112,32 +111,44 @@ class TestResponseRoundtrip(unittest.TestCase):
 class TestQueueIO(unittest.TestCase):
     def setUp(self):
         self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-test.")
-        self.queue_dir = Path(self._tmp.name)
+        self._home_patch = self._patch_home(Path(self._tmp.name))
+        self.slug = "dev"
 
     def tearDown(self):
+        self._home_patch()
         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):
         p = _proposal()
-        path = write_proposal(self.queue_dir, p)
+        path = write_proposal(p)
         self.assertTrue(path.exists())
+        self.assertEqual(host_db_path(), path)
         self.assertEqual(0o600, path.stat().st_mode & 0o777)
-        loaded = read_proposal(self.queue_dir, p.id)
+        loaded = read_proposal(self.slug, p.id)
         self.assertEqual(p, loaded)
 
     def test_list_pending_excludes_responded(self):
         a = _proposal(justification="first")
         b = _proposal(justification="second")
-        write_proposal(self.queue_dir, a)
+        write_proposal(a)
-        write_proposal(self.queue_dir, b)
+        write_proposal(b)
-        write_response(self.queue_dir, Response(
+        write_response(self.slug, Response(
             proposal_id=a.id, status=STATUS_APPROVED, notes="",
         ))
-        pending = list_pending_proposals(self.queue_dir)
+        pending = list_pending_proposals(self.slug)
         self.assertEqual([b.id], [p.id for p in pending])
 
-    def test_list_pending_returns_empty_for_missing_dir(self):
+    def test_list_pending_returns_empty_for_missing_slug(self):
-        self.assertEqual([], list_pending_proposals(self.queue_dir / "nope"))
+        self.assertEqual([], list_pending_proposals("nope"))
 
     def test_list_pending_sorted_by_arrival(self):
         # Fabricate two with explicit timestamps.
@@ -154,30 +165,30 @@ class TestQueueIO(unittest.TestCase):
             now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
         )
         # Write in reverse order.
-        write_proposal(self.queue_dir, b)
+        write_proposal(b)
-        write_proposal(self.queue_dir, a)
+        write_proposal(a)
-        ordered = list_pending_proposals(self.queue_dir)
+        ordered = list_pending_proposals(self.slug)
         self.assertEqual([a.id, b.id], [p.id for p in ordered])
 
     def test_write_and_read_response(self):
         r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no")
-        write_response(self.queue_dir, r)
+        write_response(self.slug, r)
-        self.assertEqual(r, read_response(self.queue_dir, "xyz"))
+        self.assertEqual(r, read_response(self.slug, "xyz"))
 
     def test_wait_for_response_returns_when_file_appears(self):
         p = _proposal()
-        write_proposal(self.queue_dir, p)
+        write_proposal(p)
 
         def write_after_delay():
             time.sleep(0.05)
-            write_response(self.queue_dir, Response(
+            write_response(self.slug, Response(
                 proposal_id=p.id, status=STATUS_APPROVED, notes="ok",
             ))
 
         t = threading.Thread(target=write_after_delay)
         t.start()
         try:
-            r = wait_for_response(self.queue_dir, p.id, poll_interval=0.01)
+            r = wait_for_response(self.slug, p.id, poll_interval=0.01)
         finally:
             t.join()
         self.assertEqual(STATUS_APPROVED, r.status)
@@ -187,25 +198,24 @@ class TestQueueIO(unittest.TestCase):
         deadline = time.monotonic() + 0.05
         with self.assertRaises(TimeoutError):
             wait_for_response(
-                self.queue_dir, "never",
+                self.slug, "never",
                 poll_interval=0.01, deadline=deadline,
             )
 
-    def test_archive_proposal_moves_both_files(self):
+    def test_archive_proposal_hides_rows(self):
         p = _proposal()
-        write_proposal(self.queue_dir, p)
+        write_proposal(p)
-        write_response(self.queue_dir, Response(
+        write_response(self.slug, Response(
             proposal_id=p.id, status=STATUS_APPROVED, notes="",
         ))
-        archive_proposal(self.queue_dir, p.id)
+        archive_proposal(self.slug, p.id)
-        self.assertFalse((self.queue_dir / f"{p.id}.proposal.json").exists())
+        self.assertEqual([], list_pending_proposals(self.slug))
-        self.assertFalse((self.queue_dir / f"{p.id}.response.json").exists())
+        with self.assertRaises(FileNotFoundError):
-        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.proposal.json").exists())
+            read_response(self.slug, p.id)
-        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.response.json").exists())
 
     def test_archive_is_idempotent_on_missing_files(self):
         # Should not raise.
-        archive_proposal(self.queue_dir, "nope")
+        archive_proposal(self.slug, "nope")
 
 
 class TestAuditLog(unittest.TestCase):
@@ -237,6 +247,7 @@ class TestAuditLog(unittest.TestCase):
             diff="--- before\n+++ after\n",
         )
         path = write_audit_entry(e)
+        self.assertEqual(host_db_path(), path)
         self.assertEqual(0o600, path.stat().st_mode & 0o777)
         loaded = read_audit_entries("cred-proxy", "dev")
         self.assertEqual([e], loaded)
@@ -252,12 +263,13 @@ class TestAuditLog(unittest.TestCase):
                 justification="",
                 diff="",
             ))
-        path = audit_log_path("egress", "dev")
+        entries = read_audit_entries("egress", "dev")
-        with path.open() as f:
+        self.assertEqual(3, len(entries))
-            lines = [line for line in f if line.strip()]
+        self.assertEqual(
-        self.assertEqual(3, len(lines))
+            ["2026-05-25T12:00:00+00:00", "2026-05-25T12:00:01+00:00",
-        for line in lines:
+             "2026-05-25T12:00:02+00:00"],
-            self.assertTrue(json.loads(line))  # each line is valid JSON
+            [entry.timestamp for entry in entries],
+        )
 
     def test_separate_logs_per_component_slug(self):
         write_audit_entry(AuditEntry(
@@ -379,7 +391,7 @@ class TestSupervisePrepare(unittest.TestCase):
 
     def test_prepare_creates_queue(self):
         plan = _StubSupervise().prepare("dev", self.stage_dir)
-        self.assertTrue(plan.queue_dir.is_dir())
+        self.assertTrue(plan.db_path.is_file())
         self.assertEqual("dev", plan.slug)
         self.assertEqual("", plan.internal_network)
 

tests/unit/test_supervise_cli.py

@@ -77,9 +77,7 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
 
     def test_walks_all_slug_subdirs(self):
         for slug in ("dev", "api"):
-            qdir = supervise.queue_dir_for_slug(slug)
+            supervise.write_proposal(_proposal(slug=slug))
-            qdir.mkdir(parents=True)
-            supervise.write_proposal(qdir, _proposal(slug=slug))
         pending = supervise_cli.discover_pending()
         self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})
 
@@ -97,18 +95,14 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
             now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
         )
         for p in (late, early):
-            qdir = supervise.queue_dir_for_slug(p.bottle_slug)
+            supervise.write_proposal(p)
-            qdir.mkdir(parents=True, exist_ok=True)
-            supervise.write_proposal(qdir, p)
         pending = supervise_cli.discover_pending()
         self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending])
 
     def test_excludes_already_responded(self):
         p = _proposal()
-        qdir = supervise.queue_dir_for_slug("dev")
+        supervise.write_proposal(p)
-        qdir.mkdir(parents=True)
+        supervise.write_response("dev", supervise.Response(
-        supervise.write_proposal(qdir, p)
-        supervise.write_response(qdir, supervise.Response(
             proposal_id=p.id, status=STATUS_APPROVED, notes="",
         ))
         self.assertEqual([], supervise_cli.discover_pending())
@@ -123,10 +117,8 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
 
     def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW):
         p = _proposal(tool=tool)
-        qdir = supervise.queue_dir_for_slug("dev")
+        supervise.write_proposal(p)
-        qdir.mkdir(parents=True, exist_ok=True)
+        return supervise_cli.QueuedProposal(proposal=p)
-        supervise.write_proposal(qdir, p)
-        return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
 
     def test_approve_writes_response(self):
         qp = self._enqueue()
@@ -135,7 +127,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
             return_value=("routes: []\n", "routes:\n  - host: example.com\n"),
         ):
             supervise_cli.approve(qp)
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual(STATUS_APPROVED, resp.status)
         self.assertIsNone(resp.final_file)
 
@@ -150,7 +142,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
                 final_file="routes:\n  - host: edited.example.com\n",
                 notes="tweaked",
             )
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual(STATUS_MODIFIED, resp.status)
         self.assertEqual("routes:\n  - host: edited.example.com\n", resp.final_file)
         self.assertEqual("tweaked", resp.notes)
@@ -158,7 +150,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
     def test_reject_writes_rejection(self):
         qp = self._enqueue()
         supervise_cli.reject(qp, reason="nope")
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual(STATUS_REJECTED, resp.status)
         self.assertEqual("nope", resp.notes)
 
@@ -181,36 +173,33 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
     def test_approve_gitleaks_allow_leaves_response_for_gate(self):
         qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
         supervise_cli.approve(qp, notes="dummy fixture")
-        # Gate polls the queue dir for the response; TUI must not archive it.
+        # Gate polls the DB for the response; TUI must not archive it.
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual(STATUS_APPROVED, resp.status)
         self.assertEqual("dummy fixture", resp.notes)
-        self.assertFalse((qp.queue_dir / "processed").exists())
 
     def test_tui_gitleaks_allow_requires_reason(self):
         qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
         with patch.object(supervise_cli, "_prompt", return_value=""):
             status = supervise_cli._approve_from_tui(None, qp)  # type: ignore[arg-type]
         self.assertEqual("approve aborted (empty reason)", status)
-        self.assertFalse((qp.queue_dir / "processed").exists())
 
     def test_tui_gitleaks_allow_writes_reason(self):
         qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
         with patch.object(supervise_cli, "_prompt", return_value="test fixture"):
             status = supervise_cli._approve_from_tui(None, qp)  # type: ignore[arg-type]
         self.assertIn("approved gitleaks-allow", status)
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual("test fixture", resp.notes)
 
     def test_approve_token_allow_leaves_response_for_egress(self):
         qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
         supervise_cli.approve(qp, notes="false positive")
-        # The egress addon polls the queue dir for the response; the TUI must
+        # The egress addon polls the DB for the response; the TUI must
         # not archive it (the addon archives after reading).
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual(STATUS_APPROVED, resp.status)
         self.assertEqual("false positive", resp.notes)
-        self.assertFalse((qp.queue_dir / "processed").exists())
 
     def test_token_allow_writes_no_audit_log(self):
         qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
@@ -222,14 +211,13 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
         with patch.object(supervise_cli, "_prompt", return_value=""):
             status = supervise_cli._approve_from_tui(None, qp)  # type: ignore[arg-type]
         self.assertEqual("approve aborted (empty reason)", status)
-        self.assertFalse((qp.queue_dir / "processed").exists())
 
     def test_tui_token_allow_writes_reason(self):
         qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
         with patch.object(supervise_cli, "_prompt", return_value="legit"):
             status = supervise_cli._approve_from_tui(None, qp)  # type: ignore[arg-type]
         self.assertIn("approved egress-token-allow", status)
-        resp = read_response(qp.queue_dir, qp.proposal.id)
+        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
         self.assertEqual("legit", resp.notes)
 
     def test_suffix_for_token_allow_is_txt(self):

tests/unit/test_supervise_edge.py

@@ -4,7 +4,6 @@ fallback paths."""
 
 from __future__ import annotations
 
-import os
 import tempfile
 import time
 import unittest
@@ -12,14 +11,19 @@ from pathlib import Path
 from unittest.mock import patch
 
 from bot_bottle import supervise
+from bot_bottle.audit_store import AuditStore
+from bot_bottle.queue_store import QueueStore
 from bot_bottle.supervise import (
+    AuditEntry,
     Proposal,
+    STATUS_APPROVED,
     TOOL_EGRESS_ALLOW,
     list_pending_proposals,
     read_audit_entries,
     read_proposal,
     read_response,
     wait_for_response,
+    write_audit_entry,
 )
 
 
@@ -37,58 +41,53 @@ class TestPathHelpers(unittest.TestCase):
     def test_bot_bottle_root(self) -> None:
         self.assertTrue(str(supervise.bot_bottle_root()).endswith(".bot-bottle"))
 
-    def test_queue_dir_for_slug(self) -> None:
-        self.assertIn("slug", str(supervise.queue_dir_for_slug("slug")))
-
-    def test_id_from_non_proposal_filename(self) -> None:
-        self.assertIsNone(supervise._id_from_proposal_filename(Path("x.response.json")))
-
 
 class TestReadMalformed(unittest.TestCase):
-    def test_read_proposal_non_dict(self) -> None:
+    def test_read_proposal_missing_row(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            (Path(d) / "p.proposal.json").write_text("[]")
+            with patch.dict("os.environ", {"HOME": d}), \
-            with self.assertRaises(ValueError):
+                    self.assertRaises(FileNotFoundError):
-                read_proposal(Path(d), "p")
+                read_proposal("slug", "p")
 
-    def test_read_response_non_dict(self) -> None:
+    def test_read_response_missing_row(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            (Path(d) / "p.response.json").write_text("[]")
+            with patch.dict("os.environ", {"HOME": d}), \
-            with self.assertRaises(ValueError):
+                    self.assertRaises(FileNotFoundError):
-                read_response(Path(d), "p")
+                read_response("slug", "p")
 
-    def test_list_pending_skips_malformed(self) -> None:
+    def test_list_pending_reads_db_only(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            qd = Path(d)
+            with patch.dict("os.environ", {"HOME": d}):
-            (qd / "bad.proposal.json").write_text("{ not json")
+                supervise.write_proposal(_proposal())
-            (qd / "arr.proposal.json").write_text("[]")
+                pending = list_pending_proposals("slug")
-            (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("slug", pending[0].bottle_slug)
 
     def test_list_pending_skips_when_response_present(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            qd = Path(d)
+            with patch.dict("os.environ", {"HOME": d}):
                 p = _proposal()
-            supervise.write_proposal(qd, p)
+                supervise.write_proposal(p)
-            (qd / f"{p.id}.response.json").write_text("{}")  # response exists -> skipped
+                supervise.write_response("slug", supervise.Response(
-            self.assertEqual([], list_pending_proposals(qd))
+                    proposal_id=p.id,
+                    status=STATUS_APPROVED,
+                    notes="",
+                ))
+                self.assertEqual([], list_pending_proposals("slug"))
 
 
 class TestWaitForResponse(unittest.TestCase):
-    def test_malformed_response_then_timeout(self) -> None:
+    def test_missing_response_times_out(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            (Path(d) / "p.response.json").write_text("{ not json")
+            with patch.dict("os.environ", {"HOME": d}), \
-            with self.assertRaises(TimeoutError):
+                    self.assertRaises(TimeoutError):
-                wait_for_response(Path(d), "p", deadline=time.monotonic())
+                wait_for_response("slug", "p", deadline=time.monotonic())
 
-    def test_incomplete_response_then_timeout(self) -> None:
+    def test_empty_db_response_does_not_count(self) -> None:
         with tempfile.TemporaryDirectory() as d:
-            (Path(d) / "p.response.json").write_text("{}")  # dict but from_dict raises
+            with patch.dict("os.environ", {"HOME": d}), \
-            with self.assertRaises(TimeoutError):
+                    self.assertRaises(TimeoutError):
-                wait_for_response(Path(d), "p", deadline=time.monotonic())
+                wait_for_response("slug", "p", deadline=time.monotonic())
 
 
 class TestReadAuditEntries(unittest.TestCase):
@@ -97,35 +96,94 @@ class TestReadAuditEntries(unittest.TestCase):
                 patch.dict("os.environ", {"HOME": home}):
             self.assertEqual([], read_audit_entries("egress", "nope"))
 
-    def test_skips_malformed_lines(self) -> None:
+    def test_reads_entries_from_db(self) -> None:
         with tempfile.TemporaryDirectory() as home, \
                 patch.dict("os.environ", {"HOME": home}):
-            path = supervise.audit_log_path("egress", "slug")
+            write_audit_entry(AuditEntry(
-            path.parent.mkdir(parents=True, exist_ok=True)
+                timestamp="t",
-            valid = (
+                bottle_slug="slug",
-                '{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
+                component="egress",
-                ' "operator_action": "approve", "operator_notes": "",'
+                operator_action="approve",
-                ' "justification": "", "diff": ""}'
+                operator_notes="",
-            )
+                justification="",
-            path.write_text(
+                diff="",
-                "\n"               # blank line skipped
+            ))
-                "{ not json\n"     # JSONDecodeError skipped
+            write_audit_entry(AuditEntry(
-                "[]\n"             # not a dict skipped
+                timestamp="t",
-                "{}\n"             # missing fields -> ValueError skipped
+                bottle_slug="other",
-                + valid + "\n"
+                component="egress",
-            )
+                operator_action="reject",
+                operator_notes="",
+                justification="",
+                diff="",
+            ))
             entries = read_audit_entries("egress", "slug")
             self.assertEqual(1, len(entries))
             self.assertEqual("approve", entries[0].operator_action)
 
+    def test_legacy_audit_log_file_does_not_count(self) -> None:
+        with tempfile.TemporaryDirectory() as home, \
+                patch.dict("os.environ", {"HOME": home}):
+            path = supervise.audit_log_path("egress", "slug")
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(
+                '{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
+                ' "operator_action": "approve", "operator_notes": "",'
+                ' "justification": "", "diff": ""}\n'
+            )
+            entries = read_audit_entries("egress", "slug")
+            self.assertEqual([], entries)
 
-class TestFlockFallback(unittest.TestCase):
+
-    def test_flock_on_closed_fd_is_swallowed(self) -> None:
+class TestStoreGuardBranches(unittest.TestCase):
-        # flock on a closed fd raises OSError(EBADF), which the helpers swallow.
+    """Direct QueueStore / AuditStore construction and early-return guard branches."""
-        fd = os.open(os.devnull, os.O_RDONLY)
+
-        os.close(fd)
+    def test_queue_store_explicit_db_path(self):
-        supervise._try_flock(fd)
+        with tempfile.TemporaryDirectory() as d:
-        supervise._try_funlock(fd)
+            db = Path(d) / "q.db"
+            store = QueueStore("key", db_path=db)
+            self.assertTrue(db.is_file())
+            self.assertEqual(db, store.db_path)
+
+    def test_queue_store_missing_db_list_pending_returns_empty(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "q.db"
+            store = QueueStore("key", db_path=db)
+            db.unlink()
+            self.assertEqual([], store.list_pending_proposals())
+
+    def test_queue_store_missing_db_list_all_returns_empty(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "q.db"
+            store = QueueStore("key", db_path=db)
+            db.unlink()
+            self.assertEqual([], store.list_all_pending_proposals())
+
+    def test_queue_store_missing_db_archive_is_noop(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "q.db"
+            store = QueueStore("key", db_path=db)
+            db.unlink()
+            store.archive_proposal("anything")  # must not raise
+
+    def test_queue_store_chmod_oserror_is_swallowed(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "q.db"
+            with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
+                QueueStore("key", db_path=db)  # must not raise
+
+    def test_audit_store_missing_db_read_returns_empty(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "a.db"
+            store = AuditStore(db_path=db)
+            db.unlink()
+            self.assertEqual([], store.read_audit_entries("egress", "slug"))
+
+    def test_audit_store_chmod_oserror_is_swallowed(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = Path(d) / "a.db"
+            with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
+                AuditStore(db_path=db)  # must not raise
 
 
 if __name__ == "__main__":

tests/unit/test_supervise_server.py

@@ -112,7 +112,7 @@ class TestRpcErrorTaxonomy(unittest.TestCase):
             validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
 
     def test_unknown_tool_in_tools_call_is_client_error(self):
-        config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
+        config = ServerConfig(bottle_slug="dev")
         with self.assertRaises(_RpcClientError) as cm:
             handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
         self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
@@ -122,9 +122,9 @@ class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
     def test_write_proposal_os_error_raises_internal(self):
         config = ServerConfig(
             bottle_slug="dev",
-            queue_dir=Path("/dev/null/cannot-exist"),
         )
-        with self.assertRaises(_RpcInternalError) as cm:
+        with patch.object(_sv, "write_proposal", side_effect=OSError("disk full")), \
+                self.assertRaises(_RpcInternalError) as cm:
             handle_tools_call(
                 {
                     "name": _sv.TOOL_EGRESS_ALLOW,
@@ -265,21 +265,31 @@ class TestHandleToolsList(unittest.TestCase):
 class TestHandleToolsCall(unittest.TestCase):
     def setUp(self):
         self._tmp = tempfile.TemporaryDirectory(prefix="supervise-server-test.")
-        self.queue_dir = Path(self._tmp.name)
+        self._home_patch = self._patch_home(Path(self._tmp.name))
-        self.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
+        self.config = ServerConfig(bottle_slug="dev")
 
     def tearDown(self):
+        self._home_patch()
         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:
         """Background thread: poll the queue for a fresh proposal, write a
         matching response. Returns the thread so the test can join it."""
         def runner():
             for _ in range(200):
-                pending = _sv.list_pending_proposals(self.queue_dir)
+                pending = _sv.list_pending_proposals("dev")
                 if pending:
                     p = pending[0]
-                    _sv.write_response(self.queue_dir, _sv.Response(
+                    _sv.write_response("dev", _sv.Response(
                         proposal_id=p.id, status=status, notes=notes,
                     ))
                     return
@@ -412,15 +422,11 @@ class TestHandleToolsCall(unittest.TestCase):
         finally:
             responder.join()
         # No pending proposals left after archive.
-        self.assertEqual([], _sv.list_pending_proposals(self.queue_dir))
+        self.assertEqual([], _sv.list_pending_proposals("dev"))
-        # 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):
         config = ServerConfig(
             bottle_slug="dev",
-            queue_dir=self.queue_dir,
             response_timeout_seconds=0.05,
         )
         result = handle_tools_call(
@@ -438,8 +444,7 @@ class TestHandleToolsCall(unittest.TestCase):
         text = result["content"][0]["text"]  # type: ignore[index]
         self.assertIn("status: pending", text)
         self.assertIn("proposal remains queued", text)
-        self.assertEqual(1, len(_sv.list_pending_proposals(self.queue_dir)))
+        self.assertEqual(1, len(_sv.list_pending_proposals("dev")))
-        self.assertFalse((self.queue_dir / "processed").exists())
 
 
 class TestHandleListEgressRoutes(unittest.TestCase):
@@ -461,7 +466,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
         with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
             result = handle_list_egress_routes(
                 {},
-                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
+                ServerConfig(bottle_slug="dev"),
             )
 
         self.assertFalse(result["isError"])  # type: ignore[index]
@@ -476,7 +481,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
         with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
             result = handle_list_egress_routes(
                 {},
-                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
+                ServerConfig(bottle_slug="dev"),
             )
 
         self.assertTrue(result["isError"])  # type: ignore[index]
@@ -544,7 +549,6 @@ class TestHttpEndToEnd(unittest.TestCase):
 
     def setUp(self):
         self._tmp = tempfile.TemporaryDirectory(prefix="supervise-http-test.")
-        self.queue_dir = Path(self._tmp.name)
         # Pick a random port by binding to :0 first.
         import socket
         s = socket.socket()
@@ -552,7 +556,7 @@ class TestHttpEndToEnd(unittest.TestCase):
         self.port = s.getsockname()[1]
         s.close()
         self.server = MCPServer(("127.0.0.1", self.port), MCPHandler)
-        self.server.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
+        self.server.config = ServerConfig(bottle_slug="dev")
         self.thread = threading.Thread(
             target=self.server.serve_forever, daemon=True,
         )