docs(prd): add PRD for egress control plane

Out-of-band egress enforcement & cost-control plane: meter token usage at the egress proxy, evaluate budgets with agent→bottle→parent→global precedence, and force cutoff/freeze/kill without the agent in the loop. Introduces a host-level SQLite ledger behind a thin repository API and a host-only TUI dashboard. Closes the design discussion on #251. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NkwFXLFff9PYPy4wgVBJp9
2026-06-29 11:01:47 -04:00
42 changed files with 740 additions and 1478 deletions
@@ -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/bot-bottle.db         bind-mounted at run time
+#   /run/supervise/queue/                bind-mounted at run time
 #   /home/mitmproxy/.mitmproxy/          mitmproxy CA dir
 #
 # Exposed ports inside the container:
@@ -66,8 +66,6 @@ COPY bot_bottle/egress_dlp_config.py /app/egress_dlp_config.py
 COPY bot_bottle/egress_addon.py      /app/egress_addon.py
 COPY bot_bottle/dlp_detectors.py     /app/dlp_detectors.py
 COPY bot_bottle/yaml_subset.py       /app/yaml_subset.py
-COPY bot_bottle/queue_store.py       /app/queue_store.py
-COPY bot_bottle/audit_store.py       /app/audit_store.py
 COPY bot_bottle/supervise.py         /app/supervise.py
 COPY bot_bottle/supervise_server.py  /app/supervise_server.py
 COPY bot_bottle/sidecar_init.py      /app/sidecar_init.py
@@ -83,7 +81,7 @@ RUN mkdir -p \
      /etc/git-gate \
      /git-gate/creds \
      /git \
-      /run/supervise \
+      /run/supervise/queue \
      /home/mitmproxy/.mitmproxy

 # Documentation only — the compose renderer publishes whichever
@@ -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-83%25-brightgreen)](https://coverage.readthedocs.io/)
-[![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
+[![coverage](https://img.shields.io/badge/coverage-84%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)

 **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.

@@ -209,15 +209,6 @@ 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.

@@ -1,143 +0,0 @@
-"""SQLite-backed audit store for supervise (PRD 0013)."""
-
-from __future__ import annotations
-
-import sqlite3
-from pathlib import Path
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from .supervise import AuditEntry
-
-
-def 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[0] brings a fresh DB (user_version=0)
-# to version 1, _MIGRATIONS[1] to version 2, and so on.  Add new migrations at
-# the end; never edit existing ones.
-_MIGRATIONS: list[str] = [
-    # 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:
-    """SQLite-backed persistent store for supervise audit entries."""
-
-    def __init__(self, db_path: Path | None = None) -> None:
-        self.db_path = db_path or get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        self._init()
-
-    def write_audit_entry(self, entry: AuditEntry) -> Path:
-        with self._connect() as conn:
-            conn.execute(
-                """
-                INSERT INTO supervise_audit_entries (
-                    timestamp, bottle_slug, component, operator_action,
-                    operator_notes, justification, diff
-                ) VALUES (?, ?, ?, ?, ?, ?, ?)
-                """,
-                (
-                    entry.timestamp,
-                    entry.bottle_slug,
-                    entry.component,
-                    entry.operator_action,
-                    entry.operator_notes,
-                    entry.justification,
-                    entry.diff,
-                ),
-            )
-        self._chmod()
-        return self.db_path
-
-    def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]:
-        if not self.db_path.is_file():
-            return []
-        with self._connect() as conn:
-            rows = conn.execute(
-                """
-                SELECT * FROM supervise_audit_entries
-                WHERE component = ? AND bottle_slug = ?
-                ORDER BY id
-                """,
-                (component, slug),
-            ).fetchall()
-        return [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"],
-        )
-
-    def _connect(self) -> sqlite3.Connection:
-        conn = sqlite3.connect(self.db_path)
-        conn.row_factory = sqlite3.Row
-        return conn
-
-    _SCHEMA_KEY = "audit_store"
-
-    def _init(self) -> None:
-        with self._connect() as conn:
-            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(_MIGRATIONS[version:], start=version + 1):
-                conn.execute(sql)
-                conn.execute(
-                    "INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
-                    (self._SCHEMA_KEY, i),
-                )
-        self._chmod()
-
-    def _chmod(self) -> None:
-        try:
-            self.db_path.chmod(0o600)
-        except OSError:
-            pass
-
-
-__all__ = ["AuditStore"]
@@ -34,7 +34,7 @@ from ...egress import (
 from ...git_gate import GIT_GATE_HOSTNAME
 from ...log import die, warn
 from ...supervise import (
-    DB_PATH_IN_CONTAINER,
+    QUEUE_DIR_IN_CONTAINER,
    SUPERVISE_HOSTNAME,
    SUPERVISE_PORT,
 )
@@ -163,15 +163,16 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
    if sp is not None:
        env += [
            f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
+            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
            f"SUPERVISE_PORT={SUPERVISE_PORT}",
        ]
        volumes.append({
            "type": "bind",
-            "source": str(sp.db_path),
-            "target": DB_PATH_IN_CONTAINER,
+            "source": str(sp.queue_dir),
+            "target": QUEUE_DIR_IN_CONTAINER,
            "read_only": False,
        })
+
    internal_aliases = [EGRESS_HOSTNAME]
    if gp.upstreams:
        internal_aliases.append(GIT_GATE_HOSTNAME)
@@ -37,10 +37,7 @@ from pathlib import Path
 from typing import Callable, Generator

 from ...egress import egress_resolve_token_values
-from ...git_gate import (
-    provision_git_gate_dynamic_keys,
-    revoke_git_gate_provisioned_keys,
-)
+from ...git_gate import revoke_git_gate_provisioned_keys
 from ...log import info, warn
 from . import network as network_mod
 from . import util as docker_mod
@@ -121,11 +118,6 @@ 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,
@@ -28,12 +28,9 @@ from ...egress import (
    egress_resolve_token_values,
    egress_sidecar_env_entries,
 )
-from ...git_gate import (
-    provision_git_gate_dynamic_keys,
-    revoke_git_gate_provisioned_keys,
-)
+from ...git_gate import revoke_git_gate_provisioned_keys
 from ...log import die, info, warn
-from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
+from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
 from ..docker.git_gate import (
@@ -101,8 +98,6 @@ 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)
@@ -246,19 +241,6 @@ def _stamp_agent_urls(
    )


-def _provision_git_gate_keys(
-    plan: MacosContainerBottlePlan,
-) -> MacosContainerBottlePlan:
-    if not plan.git_gate_plan.upstreams:
-        return plan
-    git_gate_plan = provision_git_gate_dynamic_keys(
-        plan.manifest.bottle,
-        plan.git_gate_plan,
-        git_gate_state_dir(plan.slug),
-    )
-    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
-
-
 def _stage_git_gate(plan: MacosContainerBottlePlan, sidecar_name: str) -> None:
    gp = plan.git_gate_plan
    if not gp.upstreams:
@@ -379,7 +361,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_DB_PATH={DB_PATH_IN_CONTAINER}",
+            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
            f"SUPERVISE_PORT={SUPERVISE_PORT}",
        ]
    return tuple(env)
@@ -405,7 +387,7 @@ def _sidecar_mounts(

    sp = plan.supervise_plan
    if sp is not None:
-        mounts.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
+        mounts.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))

    return tuple(mounts)

@@ -27,7 +27,7 @@ from ...egress import (
    egress_resolve_token_values,
    egress_sidecar_env_entries,
 )
-from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
+from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
 from ..docker import util as docker_mod
 from ..docker.egress import (
@@ -41,10 +41,7 @@ from ..docker.git_gate import (
    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
    GIT_GATE_HOOK_IN_CONTAINER,
 )
-from ...git_gate import (
-    provision_git_gate_dynamic_keys,
-    revoke_git_gate_provisioned_keys,
-)
+from ...git_gate import revoke_git_gate_provisioned_keys
 from ...log import info, warn
 from ...bottle_state import (
    egress_state_dir,
@@ -177,7 +174,6 @@ 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)
@@ -186,19 +182,6 @@ 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,
@@ -369,10 +352,10 @@ def _bundle_launch_spec(
        daemons.append("supervise")
        env += [
            f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
-            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
+            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
            f"SUPERVISE_PORT={SUPERVISE_PORT}",
        ]
-        volumes.append((str(sp.db_path), DB_PATH_IN_CONTAINER, False))
+        volumes.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))

    # Container ports the agent reaches from the smolvm guest —
    # published on host loopback so the guest can dial via TSI +
@@ -284,8 +284,9 @@ def git_gate_state_dir(identity: str) -> Path:

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


@@ -2,11 +2,6 @@
 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`.
 """
@@ -21,7 +16,7 @@ import tempfile
 from pathlib import Path
 from typing import Callable

-from ..agent_provider import get_provider, runtime_for
+from ..agent_provider import runtime_for
 from ..backend import (
    Bottle,
    BottleSpec,
@@ -36,7 +31,7 @@ from ..bottle_state import (
    is_preserved,
    mark_preserved,
 )
-from ..log import info, die
+from ..log import info
 from ..manifest import Manifest, ManifestIndex
 from ._common import PROG, USER_CWD, read_tty_line
 from . import tui
@@ -55,39 +50,6 @@ 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="?",
@@ -99,12 +61,6 @@ def cmd_start(argv: list[str]) -> int:
    dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"

    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:
@@ -115,6 +71,8 @@ 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
@@ -151,83 +109,6 @@ def cmd_start(argv: list[str]) -> int:
    )


-# --- Headless launch -----------------------------------------------------
-
-
-def _start_headless(
-    manifest: ManifestIndex,
-    args: argparse.Namespace,
-    *,
-    dry_run: bool,
-    backend_name: str | None,
-) -> int:
-    """Non-interactive launch path for orchestrators / CI / webhooks.
-
-    Resolves agent, bottles, label, and color from flags + manifest
-    defaults instead of the TUI selectors, and auto-confirms the
-    preflight. Otherwise runs the same launch core as the interactive
-    path, so the agent still execs on the inherited stdio/PTY — an
-    orchestrator allocates that PTY and relays it to its
-    desktop/mobile clients."""
-    agent_name = args.name
-    if not agent_name:
-        die("--headless requires an agent name: ./cli.py start <agent> --headless")
-    manifest.require_agent(agent_name)  # raises ManifestError if unknown
-
-    prompt = args.prompt
-    if not prompt:
-        die(
-            "--headless requires --prompt: "
-            "./cli.py start <agent> --headless --prompt 'Do the thing'"
-        )
-
-    if args.bottle:
-        bottle_names: tuple[str, ...] = tuple(args.bottle)
-    else:
-        default_bottle = _peek_agent_bottle(manifest, agent_name)
-        if not default_bottle:
-            die(
-                f"--headless: agent '{agent_name}' has no default bottle; "
-                f"pass one or more --bottle NAME"
-            )
-        bottle_names = (default_bottle,)
-
-    label = _uniquify_label_headless(args.label or agent_name)
-
-    spec = BottleSpec(
-        manifest=manifest,
-        agent_name=agent_name,
-        copy_cwd=args.cwd,
-        user_cwd=USER_CWD,
-        label=label,
-        color=args.color or "",
-        bottle_names=bottle_names,
-    )
-    return _launch_bottle(
-        spec,
-        dry_run=dry_run,
-        backend_name=backend_name,
-        assume_yes=True,
-        headless_prompt_text=prompt,
-    )
-
-
-def _uniquify_label_headless(label: str) -> str:
-    """Non-interactive analog of `_resolve_unique_label`: if the label's
-    slug collides with a running bottle, append -2, -3, … until free,
-    logging the chosen label. Orchestrators fire-and-forget many bottles,
-    so silently picking a free name beats erroring on every collision."""
-    active_slugs = {a.slug for a in enumerate_active_agents()}
-    if docker_mod.slugify(label) not in active_slugs:
-        return label
-    n = 2
-    while docker_mod.slugify(f"{label}-{n}") in active_slugs:
-        n += 1
-    chosen = f"{label}-{n}"
-    info(f"label '{label}' already in use; using '{chosen}'")
-    return chosen
-
-
 # --- Launch helpers ------------------------------------------------------


@@ -495,19 +376,10 @@ 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.
-
-    `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."""
+    attaches claude, and prints the resume hint on session end."""
    stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
    identity = ""
    try:
@@ -515,7 +387,7 @@ def _launch_bottle(
            spec,
            stage_dir=stage_dir,
            render_preflight=_text_render_preflight(),
-            prompt_yes=(lambda: True) if assume_yes else _text_prompt_yes,
+            prompt_yes=_text_prompt_yes,
            dry_run=dry_run,
            backend_name=backend_name,
        )
@@ -525,17 +397,10 @@ 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 + extra_args,
+                startup_args=plan.agent_provision.startup_args,
            )
            info(
                f"session ended (exit {exit_code}); "
@@ -45,7 +45,7 @@ from ..supervise import (
    TOOL_EGRESS_BLOCK,
    TOOL_GITLEAKS_ALLOW,
    TOOL_EGRESS_TOKEN_ALLOW,
-    list_all_pending_proposals,
+    list_pending_proposals,
    render_diff,
    write_audit_entry,
    write_response,
@@ -63,9 +63,10 @@ _REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_AL

@dataclass(frozen=True)
 class QueuedProposal:
-    """A pending proposal from the supervise queue."""
+    """A pending proposal plus the queue dir it was found in."""

    proposal: Proposal
+    queue_dir: Path


 # Errors any remediation engine may raise. Caught by the TUI key
@@ -85,11 +86,16 @@ def apply_routes_change(slug: str, content: str) -> tuple[str, str]:


 def discover_pending() -> list[QueuedProposal]:
-    """Collect pending proposals across bottles."""
-    out = [
-        QueuedProposal(proposal=proposal)
-        for proposal in list_all_pending_proposals()
-    ]
+    """Walk ~/.bot-bottle/queue/* and collect pending proposals."""
+    queue_root = _supervise.bot_bottle_root() / "queue"
+    if not queue_root.is_dir():
+        return []
+    out: list[QueuedProposal] = []
+    for slug_dir in sorted(queue_root.iterdir()):
+        if not slug_dir.is_dir():
+            continue
+        for proposal in list_pending_proposals(slug_dir):
+            out.append(QueuedProposal(proposal=proposal, queue_dir=slug_dir))
    out.sort(key=lambda q: q.proposal.arrival_timestamp)
    return out

@@ -112,6 +118,7 @@ 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),
    ]
@@ -158,7 +165,7 @@ def approve(
        notes=notes,
        final_file=final_file,
    )
-    write_response(qp.proposal.bottle_slug, response)
+    write_response(qp.queue_dir, response)
    _write_audit(
        qp, action=status, notes=notes,
        diff_before=diff_before, diff_after=diff_after,
@@ -172,7 +179,7 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
        notes=reason,
        final_file=None,
    )
-    write_response(qp.proposal.bottle_slug, response)
+    write_response(qp.queue_dir, response)
    _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")


@@ -217,7 +217,7 @@ class ClaudeAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -227,13 +227,9 @@ class ClaudeAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
-            # manifest load, but quote the path so a future unvalidated
-            # field can't inject shell metacharacters here either.
-            dst_q = shlex.quote(dst)
-            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")

    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        """Copy the prompt file into the guest, fix ownership/mode.
@@ -313,9 +309,6 @@ class ClaudeAgentProvider(AgentProvider):
                f"claude mcp add --scope user --transport http supervise {supervise_url}"
            )

-    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")
@@ -183,7 +183,7 @@ class CodexAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -193,13 +193,9 @@ class CodexAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
-            # manifest load, but quote the path so a future unvalidated
-            # field can't inject shell metacharacters here either.
-            dst_q = shlex.quote(dst)
-            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")

    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        """Copy the prompt file into the guest, fix ownership/mode.
@@ -279,9 +275,6 @@ class CodexAgentProvider(AgentProvider):
                f"codex mcp add supervise --url {shlex.quote(supervise_url)}"
            )

-    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")
@@ -238,7 +238,7 @@ class PiAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -248,13 +248,9 @@ class PiAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
-            # manifest load, but quote the path so a future unvalidated
-            # field can't inject shell metacharacters here either.
-            dst_q = shlex.quote(dst)
-            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")

    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        prompt_path = _prompt_path(plan.guest_home)
@@ -315,9 +311,6 @@ 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")
@@ -79,13 +79,14 @@ 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_slug)
+        return bool(self._supervise_queue_dir and self._supervise_slug)

    def _reload(self, *, initial: bool = False) -> None:
        try:
@@ -392,8 +393,9 @@ class EgressAddon:
            justification=_TOKEN_ALLOW_JUSTIFICATION,
            current_file_hash=_sv.sha256_hex(payload),
        )
+        queue_dir = Path(self._supervise_queue_dir)
        try:
-            _sv.write_proposal(proposal)
+            _sv.write_proposal(queue_dir, proposal)
        except OSError as e:
            sys.stderr.write(
                f"egress: could not queue token-allow proposal: {e}; "
@@ -409,8 +411,8 @@ class EgressAddon:
            **self._req_ctx(flow),
        }) + "\n")

-        response = await self._await_token_response(proposal.id)
-        _sv.archive_proposal(self._supervise_slug, proposal.id)
+        response = await self._await_token_response(queue_dir, proposal.id)
+        _sv.archive_proposal(queue_dir, proposal.id)

        if response is not None and response.status in (
            _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
@@ -437,15 +439,16 @@ class EgressAddon:

    async def _await_token_response(
        self,
+        queue_dir: Path,
        proposal_id: str,
    ) -> "_sv.Response | None":
-        """Poll the DB for the operator's response without blocking the
+        """Poll the queue dir for the operator's response without blocking the
        proxy event loop. Returns the Response, or None on timeout."""
        loop = asyncio.get_running_loop()
        deadline = loop.time() + self._token_allow_timeout
        while True:
            try:
-                return _sv.read_response(self._supervise_slug, proposal_id)
+                return _sv.read_response(queue_dir, proposal_id)
            except (OSError, ValueError, KeyError):
                # Not written yet, or a partial/malformed write — retry until
                # the deadline, then fail closed.
@@ -30,6 +30,7 @@ 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
@@ -52,7 +53,6 @@ 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,14 +93,20 @@ 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, the returned upstream intentionally
-        has an empty identity file. Backend launch fills that in after
-        the operator confirms the preflight.
+        For `gitea` key entries, also generates and registers
+        a fresh deploy key via the forge API and writes the private key
+        + key ID to `stage_dir`.

        Returned plan is incomplete: the launch step must fill
        `internal_network` / `egress_network` via `dataclasses.replace`
        before passing the plan to `.start`."""
-        upstreams = git_gate_upstreams_for_bottle(bottle)
+        upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
+        for i, entry in enumerate(bottle.git):
+            upstreams_list[i] = dataclasses.replace(
+                upstreams_list[i],
+                identity_file=_resolve_identity_file(entry, slug, stage_dir),
+            )
+        upstreams = tuple(upstreams_list)
        entrypoint = stage_dir / "git_gate_entrypoint.sh"
        entrypoint.write_text(git_gate_render_entrypoint(upstreams))
        entrypoint.chmod(0o600)
@@ -156,7 +162,6 @@ __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",
@@ -9,16 +9,10 @@ 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,
@@ -101,45 +95,8 @@ 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",
 ]
@@ -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 slug:
+if not queue_dir or not slug:
    sys.exit(2)

 try:
@@ -277,19 +277,31 @@ for i, finding in enumerate(raw, 1):
    ])

 payload = "\n".join(lines).rstrip() + "\n"
-proposal = _sv.Proposal.new(
-    bottle_slug=slug,
-    tool=_sv.TOOL_GITLEAKS_ALLOW,
-    proposed_file=payload,
-    justification=(
+proposal_id = str(uuid.uuid4())
+proposal = {
+    "id": proposal_id,
+    "bottle_slug": slug,
+    "tool": "gitleaks-allow",
+    "proposed_file": payload,
+    "justification": (
        "git-gate found gitleaks findings hidden by # gitleaks:allow; "
        "approve only for dummy test fixtures or confirmed false positives"
    ),
-    current_file_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
-    now=datetime.datetime.now(datetime.timezone.utc),
-)
-_sv.write_proposal(proposal)
-print(proposal.id)
+    "arrival_timestamp": datetime.datetime.now(
+        datetime.timezone.utc
+    ).isoformat(),
+    "current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+}
+queue = Path(queue_dir)
+queue.mkdir(parents=True, exist_ok=True)
+path = queue / f"{proposal_id}.proposal.json"
+tmp = path.with_suffix(path.suffix + ".tmp")
+with tmp.open("w", encoding="utf-8") as f:
+    json.dump(proposal, f, indent=2)
+    f.write("\n")
+os.chmod(tmp, 0o600)
+os.replace(tmp, path)
+print(proposal_id)
 PY
  )
  rc=$?
@@ -302,7 +314,8 @@ PY
    return 1
  fi

-  slug=${SUPERVISE_BOTTLE_SLUG:-}
+  queue_dir=${SUPERVISE_QUEUE_DIR:-}
+  response_file="$queue_dir/${proposal_id}.response.json"
  timeout=${SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS:-300}
  case "$timeout" in
    ''|*[!0-9]*)
@@ -314,35 +327,26 @@ PY
  echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
  waited=0
  while [ "$waited" -lt "$timeout" ]; do
-    status=$(python3 - "$slug" "$proposal_id" <<'PY'
+    if [ -f "$response_file" ]; then
+      status=$(python3 - "$response_file" <<'PY'
+import json
 import sys
-
-from bot_bottle import supervise as _sv
-
-slug = sys.argv[1]
 try:
-    response = _sv.read_response(slug, sys.argv[2])
-except FileNotFoundError:
-    sys.exit(2)
-print(response.status)
+    with open(sys.argv[1], encoding="utf-8") as f:
+        raw = json.load(f)
+except (OSError, json.JSONDecodeError):
+    sys.exit(1)
+status = raw.get("status")
+if not isinstance(status, str):
+    sys.exit(1)
+print(status)
 PY
-    )
-    rc=$?
-    if [ "$rc" -eq 2 ]; then
-      status=""
-    elif [ "$rc" -ne 0 ]; then
-      status="invalid"
-    fi
-    if [ -n "$status" ]; then
+      ) || status=""
      case "$status" in
        approved|modified)
-          python3 - "$slug" "$proposal_id" <<'PY' || true
-import sys
-
-from bot_bottle import supervise as _sv
-
-_sv.archive_proposal(sys.argv[1], sys.argv[2])
-PY
+          mkdir -p "$queue_dir/processed"
+          mv -f "$queue_dir/${proposal_id}.proposal.json" "$queue_dir/processed/" 2>/dev/null || true
+          mv -f "$queue_dir/${proposal_id}.response.json" "$queue_dir/processed/" 2>/dev/null || true
          echo "git-gate: supervisor approved # gitleaks:allow for $ref" >&2
          return 0
          ;;
@@ -495,3 +499,4 @@ if ! git -C "$repo_dir" rev-parse --verify HEAD >/dev/null 2>&1; then
 fi
 exit 0
 """
+
@@ -16,16 +16,11 @@ 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

@@ -8,7 +8,7 @@ from typing import cast
 from .agent_provider import PROVIDER_TEMPLATES
 from .manifest_util import ManifestError, as_json_object
 from .manifest_git import ManifestGitUser
-from .manifest_schema import AGENT_MODEL_KEYS, is_valid_entity_name
+from .manifest_schema import AGENT_MODEL_KEYS


@dataclass(frozen=True)
@@ -161,16 +161,6 @@ class ManifestAgent:
                        f"agent '{name}' skills[{i}] must be a string "
                        f"(was {type(skill).__name__})"
                    )
-                # Skill names become host/guest path segments and are
-                # interpolated into provisioning shell commands, so they
-                # must fit the same kebab-case convention as bottle/agent
-                # filenames — rejecting anything that could break out of a
-                # path segment or inject shell metacharacters.
-                if not is_valid_entity_name(skill):
-                    raise ManifestError(
-                        f"agent '{name}' skills[{i}] {skill!r} is not a valid "
-                        f"skill name; must match [a-z][a-z0-9-]*"
-                    )
                collected.append(skill)
            skills = tuple(collected)

@@ -33,20 +33,13 @@ AGENT_KEYS = (
 AGENT_MODEL_KEYS = AGENT_KEYS | frozenset({"prompt"})


-def is_valid_entity_name(name: str) -> bool:
-    """True if `name` fits the kebab-case `[a-z][a-z0-9-]*` convention
-    shared by bottle/agent filenames and skill names. Names that satisfy
-    this are also safe to interpolate into a host/guest path segment."""
-    return bool(_FILENAME_RX.match(name))
-
-
 def entity_name_from_path(path: Path) -> str | None:
    """Return the entity name implied by the filename, or None if the
    filename does not fit the [a-z][a-z0-9-]* convention."""
    if path.suffix != ".md":
        return None
    stem = path.stem
-    if not is_valid_entity_name(stem):
+    if not _FILENAME_RX.match(stem):
        return None
    return stem

@@ -1,270 +0,0 @@
-"""SQLite-backed queue store for supervise proposals and responses (PRD 0013)."""
-
-from __future__ import annotations
-
-import os
-import sqlite3
-from pathlib import Path
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from .supervise import Proposal, Response
-
-
-def 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[0] brings a fresh DB (user_version=0)
-# to version 1, _MIGRATIONS[1] to version 2, and so on.  Add new migrations at
-# the end; never edit existing ones.
-_MIGRATIONS: list[str] = [
-    # 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:
-    """SQLite-backed persistent store for supervise proposals and responses."""
-
-    def __init__(self, queue_key: str, db_path: Path | None = None) -> None:
-        self.queue_key = queue_key
-        if db_path is not None:
-            self.db_path = db_path
-        else:
-            # In the sidecar container SUPERVISE_DB_PATH points at the
-            # bind-mounted host DB. On the host this env var is never set,
-            # so we always fall through to host_db_path().
-            env_path = os.environ.get("SUPERVISE_DB_PATH", "").strip()
-            self.db_path = Path(env_path) if env_path else get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        self._init()
-
-    def write_proposal(self, proposal: Proposal) -> Path:
-        with self._connect() as conn:
-            conn.execute(
-                """
-                INSERT OR REPLACE INTO supervise_proposals (
-                    queue_key, id, bottle_slug, tool, proposed_file, justification,
-                    arrival_timestamp, current_file_hash, archived
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
-                """,
-                (
-                    self.queue_key,
-                    proposal.id,
-                    proposal.bottle_slug,
-                    proposal.tool,
-                    proposal.proposed_file,
-                    proposal.justification,
-                    proposal.arrival_timestamp,
-                    proposal.current_file_hash,
-                ),
-            )
-        self._chmod()
-        return self.db_path
-
-    def read_proposal(self, proposal_id: str) -> Proposal:
-        with self._connect() as conn:
-            row = conn.execute(
-                """
-                SELECT * FROM supervise_proposals
-                WHERE queue_key = ? AND id = ? AND archived = 0
-                """,
-                (self.queue_key, proposal_id),
-            ).fetchone()
-        if row is None:
-            raise FileNotFoundError(proposal_id)
-        return 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"],
-        )
-
-    def _connect(self) -> sqlite3.Connection:
-        conn = sqlite3.connect(self.db_path)
-        conn.row_factory = sqlite3.Row
-        return conn
-
-    _SCHEMA_KEY = "queue_store"
-
-    def _init(self) -> None:
-        with self._connect() as conn:
-            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(_MIGRATIONS[version:], start=version + 1):
-                conn.execute(sql)
-                conn.execute(
-                    "INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
-                    (self._SCHEMA_KEY, i),
-                )
-        self._chmod()
-
-    def _chmod(self) -> None:
-        try:
-            self.db_path.chmod(0o600)
-        except OSError:
-            pass
-
-
-__all__ = ["QueueStore"]
@@ -9,14 +9,15 @@ calls when it needs an operator-reviewed egress change:

 Each tool call: the agent passes the full proposed file plus a
 justification text. The sidecar validates the proposal syntactically,
-writes it to the host SQLite queue table, and holds the tool-call
+writes it to the host's per-bottle queue dir, and holds the tool-call
 connection open. The operator's supervise TUI
 (bot_bottle.cli.supervise) sees the proposal, accepts
-approve / modify / reject, and writes a response row. The sidecar sees
-the response and returns `{status, notes}` to the agent.
+approve / modify / reject, and writes a response file alongside the
+proposal. The sidecar sees the response and returns `{status, notes}`
+to the agent.

 This module defines the host-side library: dataclasses for the queue
-record shapes, queue read/write helpers, the audit log writer, and the
+file shapes, queue read/write helpers, the audit log writer, and the
 diff renderer. The in-container sidecar lives in
 bot_bottle/supervise_server.py; the supervise daemon's container
 lifecycle is owned by the sidecar bundle (PRD 0024).
@@ -33,6 +34,8 @@ from __future__ import annotations
 import dataclasses
 import difflib
 import hashlib
+import json
+import os
 import time
 import uuid
 from abc import ABC
@@ -83,9 +86,8 @@ STATUSES: tuple[str, ...] = (STATUS_APPROVED, STATUS_MODIFIED, STATUS_REJECTED)
 # `routes edit <bottle>` verb writes entries with this action.
 ACTION_OPERATOR_EDIT = "operator-edit"

-DB_PATH_IN_CONTAINER = "/run/supervise/bot-bottle.db"
+QUEUE_DIR_IN_CONTAINER = "/run/supervise/queue"
 DEFAULT_POLL_INTERVAL_SEC = 0.5
-HOST_DB_FILENAME = "bot-bottle.db"


 # --- Paths -----------------------------------------------------------------
@@ -95,6 +97,10 @@ 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"

@@ -103,16 +109,14 @@ 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."""
+    """One pending tool-call from the agent. The sidecar writes one
+    of these to the queue dir on a tool call; the operator's TUI
+    reads them; the sidecar polls for a matching Response."""

    id: str
    bottle_slug: str
@@ -166,7 +170,7 @@ class Proposal:
@dataclass(frozen=True)
 class Response:
    """The operator's decision on a proposal. The TUI writes one of
-    these to the queue table; the sidecar reads it and returns the
+    these to the queue dir; the sidecar reads it and returns the
    `{status, notes}` pair to the agent's tool call.

    `final_file` carries the file content the supervisor will
@@ -219,50 +223,90 @@ 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 write_proposal(proposal: Proposal) -> Path:
-    """Persist `proposal` in the queue database, mode 0o600.
+def _proposal_filename(proposal_id: str) -> str:
+    return f"{proposal_id}.proposal.json"
+
+
+def _response_filename(proposal_id: str) -> str:
+    return f"{proposal_id}.response.json"
+
+
+def _id_from_proposal_filename(path: Path) -> str | None:
+    name = path.name
+    if not name.endswith(".proposal.json"):
+        return None
+    return name[: -len(".proposal.json")]
+
+
+def write_proposal(queue_dir: Path, proposal: Proposal) -> Path:
+    """Persist `proposal` as JSON in the queue dir, mode 0o600.
    Directory is created if missing."""
-    return QueueStore(proposal.bottle_slug).write_proposal(proposal)
+    queue_dir.mkdir(parents=True, exist_ok=True)
+    path = queue_dir / _proposal_filename(proposal.id)
+    payload = json.dumps(proposal.to_dict(), indent=2) + "\n"
+    _atomic_write(path, payload, mode=0o600)
+    return path


-def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal:
-    return QueueStore(bottle_slug).read_proposal(proposal_id)
+def read_proposal(queue_dir: Path, proposal_id: str) -> Proposal:
+    path = queue_dir / _proposal_filename(proposal_id)
+    with path.open() as f:
+        raw = json.load(f)
+    if not isinstance(raw, dict):
+        raise ValueError(f"{path}: top-level must be an object")
+    return Proposal.from_dict(raw)


-def list_pending_proposals(bottle_slug: str) -> list[Proposal]:
-    """All proposals for `bottle_slug` that do not yet have a matching
-    response. Sorted by `arrival_timestamp` so the operator
+def list_pending_proposals(queue_dir: Path) -> list[Proposal]:
+    """All proposals in `queue_dir` that do not yet have a matching
+    response file. Sorted by `arrival_timestamp` so the operator
    sees the queue FIFO."""
-    return QueueStore(bottle_slug).list_pending_proposals()
+    if not queue_dir.is_dir():
+        return []
+    out: list[Proposal] = []
+    for path in sorted(queue_dir.glob("*.proposal.json")):
+        proposal_id = _id_from_proposal_filename(path)
+        if proposal_id is None:
+            continue
+        if (queue_dir / _response_filename(proposal_id)).exists():
+            continue
+        try:
+            with path.open() as f:
+                raw = json.load(f)
+        except (OSError, json.JSONDecodeError):
+            continue
+        if not isinstance(raw, dict):
+            continue
+        try:
+            out.append(Proposal.from_dict(raw))
+        except (KeyError, ValueError):
+            continue
+    out.sort(key=lambda p: p.arrival_timestamp)
+    return out


-def list_all_pending_proposals() -> list[Proposal]:
-    """All pending proposals across bottles, sorted FIFO."""
-    return QueueStore("").list_all_pending_proposals()
+def write_response(queue_dir: Path, response: Response) -> Path:
+    queue_dir.mkdir(parents=True, exist_ok=True)
+    path = queue_dir / _response_filename(response.proposal_id)
+    payload = json.dumps(response.to_dict(), indent=2) + "\n"
+    _atomic_write(path, payload, mode=0o600)
+    return path


-def write_response(bottle_slug: str, response: Response) -> Path:
-    return QueueStore(bottle_slug).write_response(response)
-
-
-def read_response(bottle_slug: str, proposal_id: str) -> Response:
-    return QueueStore(bottle_slug).read_response(proposal_id)
+def read_response(queue_dir: Path, proposal_id: str) -> Response:
+    path = queue_dir / _response_filename(proposal_id)
+    with path.open() as f:
+        raw = json.load(f)
+    if not isinstance(raw, dict):
+        raise ValueError(f"{path}: top-level must be an object")
+    return Response.from_dict(raw)


 def wait_for_response(
-    bottle_slug: str,
+    queue_dir: Path,
    proposal_id: str,
    *,
    poll_interval: float = DEFAULT_POLL_INTERVAL_SEC,
@@ -273,35 +317,90 @@ def wait_for_response(
    which the wait raises TimeoutError. None waits forever — the
    natural shape, since the operator's response time is unbounded.

-    Polls SQLite so the implementation stays portable and stdlib-only."""
-    store = QueueStore(bottle_slug)
+    Polls the filesystem so the implementation stays portable and
+    stdlib-only."""
+    path = queue_dir / _response_filename(proposal_id)
    while True:
+        if path.exists():
            try:
-            return store.read_response(proposal_id)
-        except FileNotFoundError:
+                with path.open() as f:
+                    raw = json.load(f)
+            except (OSError, json.JSONDecodeError):
+                raw = None
+            if isinstance(raw, dict):
+                try:
+                    return Response.from_dict(raw)
+                except (KeyError, ValueError):
                    pass
        if deadline is not None and time.monotonic() >= deadline:
            raise TimeoutError(f"no response for proposal {proposal_id!r}")
        time.sleep(poll_interval)


-def archive_proposal(bottle_slug: str, proposal_id: str) -> None:
-    """Mark both proposal and response rows processed.
-    Idempotent — missing rows are silently skipped."""
-    QueueStore(bottle_slug).archive_proposal(proposal_id)
+def archive_proposal(queue_dir: Path, proposal_id: str) -> None:
+    """Move both proposal and response files to `<queue_dir>/processed/`.
+    Idempotent — missing files are silently skipped."""
+    processed = queue_dir / "processed"
+    processed.mkdir(parents=True, exist_ok=True)
+    for name in (_proposal_filename(proposal_id), _response_filename(proposal_id)):
+        src = queue_dir / name
+        if src.exists():
+            src.rename(processed / name)


 # --- Audit log -------------------------------------------------------------


 def write_audit_entry(entry: AuditEntry) -> Path:
-    """Append `entry` to the host supervise audit table."""
-    return AuditStore().write_audit_entry(entry)
+    """Append `entry` as one JSON-Lines record to the per-bottle
+    audit log. Acquires an advisory exclusive lock so concurrent
+    writers don't interleave bytes."""
+    path = audit_log_path(entry.component, entry.bottle_slug)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(entry.to_dict(), sort_keys=False) + "\n"
+    fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
+    try:
+        _try_flock(fd)
+        try:
+            os.write(fd, line.encode("utf-8"))
+        finally:
+            _try_funlock(fd)
+    finally:
+        os.close(fd)
+    return path


 def read_audit_entries(component: str, slug: str) -> list[AuditEntry]:
-    """Load all audit entries for the given component+slug."""
-    return AuditStore().read_audit_entries(component, slug)
+    """Load all audit entries for the given component+slug. Empty
+    list if the log doesn't exist."""
+    path = audit_log_path(component, slug)
+    if not path.is_file():
+        return []
+    out: list[AuditEntry] = []
+    with path.open() as f:
+        for raw_line in f:
+            raw_line = raw_line.strip()
+            if not raw_line:
+                continue
+            try:
+                raw = json.loads(raw_line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(raw, dict):
+                continue
+            try:
+                out.append(AuditEntry(
+                    timestamp=_require_str(raw, "timestamp"),
+                    bottle_slug=_require_str(raw, "bottle_slug"),
+                    component=_require_str(raw, "component"),
+                    operator_action=_require_str(raw, "operator_action"),
+                    operator_notes=_require_str(raw, "operator_notes"),
+                    justification=_require_str(raw, "justification"),
+                    diff=_require_str(raw, "diff"),
+                ))
+            except ValueError:
+                continue
+    return out


 # --- Diff rendering --------------------------------------------------------
@@ -334,34 +433,35 @@ def sha256_hex(content: str) -> str:
 class SupervisePlan:
    """Output of Supervise.prepare; consumed by .start.

-    `db_path` is the host database bind-mounted into the sidecar at
-    /run/supervise/bot-bottle.db. `internal_network` is empty at
-    prepare time; the backend's launch step fills it via
-    dataclasses.replace before calling .start."""
+    `queue_dir` is the host directory bind-mounted into the sidecar
+    at /run/supervise/queue. `internal_network` is empty at prepare
+    time; the backend's launch step fills it via dataclasses.replace
+    before calling .start."""

    slug: str
-    db_path: Path
+    queue_dir: Path
    internal_network: str = ""


 class Supervise(ABC):
-    """Per-bottle supervise sidecar. Encapsulates host-side database
-    staging; the sidecar's start/stop lifecycle is backend-specific."""
+    """Per-bottle supervise sidecar. Encapsulates the host-side
+    prepare (queue dir staging); the sidecar's start/stop lifecycle
+    is backend-specific."""

    def prepare(
        self,
        slug: str,
        stage_dir: Path,
    ) -> SupervisePlan:
-        """Stage the host database. Returns the plan; `internal_network`
-        must be set by the launch step before .start runs."""
+        """Stage the per-bottle queue dir on the host. Returns the
+        plan; `internal_network` must be set by the launch step before
+        .start runs."""
        del stage_dir
-        db_path = host_db_path()
-        QueueStore(slug)
-        AuditStore(db_path)
+        queue_dir = queue_dir_for_slug(slug)
+        queue_dir.mkdir(parents=True, exist_ok=True)
        return SupervisePlan(
            slug=slug,
-            db_path=db_path,
+            queue_dir=queue_dir,
        )

 # --- Helpers ---------------------------------------------------------------
@@ -374,15 +474,47 @@ 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",
-    "QueueStore",
+    "QUEUE_DIR_IN_CONTAINER",
    "Response",
    "STATUSES",
    "STATUS_APPROVED",
@@ -404,9 +536,8 @@ __all__ = [
    "audit_dir",
    "audit_log_path",
    "bot_bottle_root",
-    "host_db_path",
    "list_pending_proposals",
-    "list_all_pending_proposals",
+    "queue_dir_for_slug",
    "read_audit_entries",
    "read_proposal",
    "read_response",
@@ -7,13 +7,14 @@ config changes when stuck. The tools are `egress-allow`,
 Each queued tool call:

  1. Validates the proposed file syntactically.
-  2. Writes a Proposal to the host SQLite database.
-  3. Blocks polling for a matching Response row.
+  2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
+     the host's ~/.bot-bottle/queue/<slug>/).
+  3. Blocks polling for a matching Response file.
  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). SUPERVISE_DB_PATH
-points at the bind-mounted host database.
+container creation by the backend's start step). The queue dir comes
+from SUPERVISE_QUEUE_DIR (default `/run/supervise/queue`).

 Speaks MCP over HTTP+JSON-RPC. Methods handled:

@@ -41,6 +42,7 @@ 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
@@ -275,6 +277,7 @@ 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


@@ -373,7 +376,7 @@ def handle_tools_call(
        current_file_hash=_sv.sha256_hex(proposed_file),
    )
    try:
-        _sv.write_proposal(proposal)
+        _sv.write_proposal(config.queue_dir, proposal)
    except OSError as e:
        raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e
    sys.stderr.write(
@@ -384,7 +387,7 @@ def handle_tools_call(
    deadline = time.monotonic() + config.response_timeout_seconds
    try:
        response = _sv.wait_for_response(
-            config.bottle_slug,
+            config.queue_dir,
            proposal.id,
            poll_interval=MIN_RESPONSE_POLL_INTERVAL_SECONDS,
            deadline=deadline,
@@ -396,7 +399,7 @@ def handle_tools_call(
            "isError": False,
        }
    try:
-        _sv.archive_proposal(config.bottle_slug, proposal.id)
+        _sv.archive_proposal(config.queue_dir, proposal.id)
    except OSError as e:
        raise _RpcInternalError(f"failed to archive proposal: {e}") from e

@@ -536,7 +539,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
 class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    allow_reuse_address = True
    daemon_threads = True
-    config: ServerConfig = ServerConfig(bottle_slug="")
+    config: ServerConfig = ServerConfig(bottle_slug="", queue_dir=Path())


 # --- Entry point -----------------------------------------------------------
@@ -545,18 +548,21 @@ 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}; "
+        f"slug={bottle_slug!r}; queue={queue_dir}; "
        f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n"  # type: ignore[arg-type]
    )
    sys.stderr.flush()
@@ -575,6 +581,7 @@ 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:
@@ -584,6 +591,7 @@ 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,
@@ -0,0 +1,247 @@
+# PRD prd-new: Egress control plane — metering, budgets, and forced cutoff
+
+- **Status:** Draft
+- **Author:** didericis
+- **Created:** 2026-06-25
+- **Issue:** #251
+
+## Summary
+
+Add an **out-of-band egress enforcement & observability plane**: meter every
+agent's token usage at the egress proxy, decrement budgets without the agent's
+cooperation, and forcibly cut a bottle's egress when a budget is exhausted —
+either automatically or on command from a host-level dashboard. The trigger
+(usage threshold) and the action (route-drop / freeze / kill) both live in the
+egress plane and run with no agent in the loop. This is distinct from the
+supervise sidecar (PRD 0013), which is agent-initiated and therefore cannot
+enforce a cost cutoff on a runaway agent. State (usage ledger, budgets, audit)
+moves into a host-level SQLite database behind a thin repository API, the first
+SQL store in an otherwise flat-file repo.
+
+## Problem
+
+bot-bottle can't currently do two things the cost-overrun case demands:
+
+1. **Forced egress shutdown on limit.** When an agent crosses a token
+   threshold, kill its egress automatically — no human in the loop.
+2. **Remote (host-level) management.** Drive agents from a single surface:
+   see usage, cut egress, stop bottles, to prevent cost overruns.
+
+The existing supervise sidecar (PRD 0013) is **entirely agent-initiated**: every
+action begins with the agent voluntarily calling an MCP tool and an operator
+approving it. A runaway or expensive agent — exactly the cost-overrun case —
+will never call `egress-block` on itself. Supervision is therefore a
+**collaborative recovery** mechanism, not an **enforcement** mechanism; making
+it mandatory (#249) would not deliver forced cost-cutoff.
+
+The requirement forces a distinction the current design blurs:
+
+- **Plane A — enforcement / observability (this PRD).** System → infrastructure.
+  Meter usage, cut egress on threshold or command, account for cost.
+  Out-of-band; independent of the agent. **Unconditional** — an enforcement
+  plane you can opt out of isn't enforcement.
+- **Plane B — agent-facing recovery (the existing supervise sidecar).**
+  Agent → operator, approval-gated. Useful interactively; meaningless for a
+  headless agent with no operator watching its queue. Remains optional.
+
+This PRD builds Plane A. It reframes the "always-on control" invariant of #249
+as "the egress control plane is always present" — a more defensible property
+than "every agent runs the agent-facing supervisor." Unsupervised
+(headless/CI/ephemeral) agents stay first-class: still subject to the mandatory
+meter + kill switch, they simply lack the agent-facing proposal tools they
+couldn't use anyway.
+
+## Goals / Success Criteria
+
+- The egress proxy meters every request to a metered API host (e.g.
+  `api.anthropic.com`) and records authoritative token usage per bottle and per
+  agent provider, with no agent cooperation.
+- A budget can be set at four scopes with deterministic precedence
+  (**agent → bottle → parent bottle → global host budget**); the
+  most-specific applicable budget governs.
+- When usage crosses a budget, the bottle's configured **cutoff policy**
+  (`cutoff` | `freeze` | `kill`) fires automatically, executed host-side on the
+  egress plane — never via the supervise queue.
+- An operator can, from a single **host-level TUI dashboard**, see live per-bottle
+  usage against budget and command a cutoff/stop on demand.
+- Host budgets, default cutoff policy, and per-provider limits are declared in a
+  new host-level `~/.bot-bottle/settings.yml`, parseable by `yaml_subset.py`.
+- All usage, budget state, and enforcement actions persist in a host-level
+  SQLite DB behind a thin repository API, so the store can later be swapped for
+  a cross-host cloud service.
+
+## Non-goals
+
+- **Remote control / cross-host control plane.** Web + mobile remote control,
+  cross-host budgets, and the authn/transport they require are explicitly
+  deferred. v1 is a **host-only TUI** with no remote surface.
+- **Dollar-denominated budgets.** Budgets are token counts keyed by agent
+  provider, not currency. Price tables are out of scope.
+- **Migrating existing flat-file state into SQLite.** Resume `metadata.json`,
+  transcripts, Dockerfile overrides, the supervise queue, and audit logs stay on
+  the filesystem. Only the *new* metering/budget/enforcement ledger is SQL.
+- **Making the supervise sidecar (Plane B) mandatory.** Out of scope here; this
+  PRD is the answer to "what should be unconditional" (Plane A), leaving #249's
+  Plane-B question open.
+- **Per-request hard pre-send blocking as the primary mechanism.** The gate is
+  budget-crossing detected at/after metering; a pre-flight estimator (below) is a
+  refinement, not the core enforcement path.
+
+## Design
+
+### Two measurements: gate vs. account
+
+There are two distinct needs, and they want different signals:
+
+- **Account (authoritative).** Decrement the real budget from the API
+  **response**, which already carries authoritative usage (Anthropic
+  `input_tokens` / `output_tokens`, OpenAI `usage`). The egress addon already
+  has a `response(flow)` hook (`bot_bottle/egress_addon.py:460`), so the real
+  number is available with no extra network call. **Caveat:** agent traffic is
+  mostly streaming SSE, so the response path must tail the stream for the final
+  usage event rather than parse a single JSON body — scoped explicitly as work.
+- **Gate (estimate).** To block *before* sending, only the request is available,
+  so an estimator / provider `count_tokens` endpoint is the only option.
+
+Calling `count_tokens` for accounting would be both less accurate *and* an extra
+metered egress call per request, so accounting uses response `usage` and the
+estimator is reserved for the optional pre-flight gate.
+
+### `count_tokens` on agent providers
+
+Add an abstract `count_tokens(request) -> int` to the `AgentProvider`
+abstraction (`bot_bottle/agent_provider.py`):
+
+- **Default** is a good-enough stdlib estimator. Prefer stdlib only; a small
+  pip dependency *for the sidecar* is acceptable for the fallback if stdlib
+  proves too inaccurate (this does not relax the package's stdlib-first stance —
+  it would be a sidecar-only dep, like the bundle already carries).
+- **Built-in `claude`** uses Anthropic's token-counting endpoint;
+  **built-in `codex`** uses OpenAI's. These are exact for the gate but cost a
+  metered call, so they are gate-only; accounting still comes from the response.
+
+### Budgets and precedence
+
+Budgets are token counts keyed by **agent provider name** (the same names
+bottles already use). Four scopes, most-specific wins:
+
+```
+agent  →  bottle  →  parent bottle  →  global (host)
+```
+
+The global host budget is the highest-priority feature to ship (the cross-host
+control plane will eventually consume it); per-agent and per-bottle budgets
+override it for finer control. A budget can also be supplied **at bottle
+launch** (`--budget` or equivalent), overriding the settings.yml defaults for
+that run. Enforcement evaluates the effective budget as the
+nearest-defined scope at decrement time.
+
+### `~/.bot-bottle/settings.yml`
+
+New **host-level** settings file (the `~/.bot-bottle/` root, *not* the per-repo
+`.bot-bottle/` — host budgets must not be committed per-repo). Parsed by
+`yaml_subset.py`, so it must stay within that bounded subset (flat mappings,
+scalars; no anchors, no multi-line block scalars). Shape:
+
+```yaml
+budget:
+  claude: 5000000      # token budget keyed by agent provider
+  codex: 2000000
+shutdown: cutoff       # default cutoff policy: cutoff | freeze | kill
+```
+
+### Forced cutoff and cutoff policy
+
+On budget exhaustion (or an operator command), the configured per-bottle cutoff
+policy fires. The three policies map onto primitives that already exist:
+
+- **`cutoff`** (default) — drop the bottle's `routes.yaml` to empty and reload
+  (or isolate the bottle from the egress network); the agent/bottle keeps
+  running but can no longer reach metered hosts. This is the route-drop already
+  available on the egress plane (`bot_bottle/backend/egress_apply.py`).
+- **`freeze`** — commit/snapshot state, then kill the agent/bottle; resumable
+  later via `bot_bottle/backend/freeze.py`.
+- **`kill`** — tear the bottle down without saving state (backend teardown).
+
+The trigger lives in the metering path and the action in the egress/backend
+plane; **neither touches the supervise proposal queue** (design constraint from
+#251).
+
+### Host-level SQLite store
+
+**Decision: introduce SQLite now, narrowly.**
+
+- **The dependency objection doesn't apply.** `sqlite3` is in the Python stdlib,
+  so it does not break the AGENTS.md stdlib-first / no-runtime-pip stance — same
+  category as the hand-rolled `yaml_subset.py`, except the stdlib already ships
+  the whole engine.
+- **It fits the problem.** A *global* token budget decremented concurrently by N
+  egress sidecars (today `~/.bot-bottle/` already has `state/`, `audit/`,
+  `queue/` written by parallel bottles) is a read-modify-write race. Over JSON
+  that means hand-rolled file locking; SQLite gives atomic transactions + WAL for
+  free. The per-agent/per-bottle precedence rollup plus "sum across all bottles"
+  is a `GROUP BY`, not an N-directory rescan.
+- **It rehearses the cloud swap.** "Wrap operations in an API so we can swap to a
+  cloud service" maps directly onto a thin repository/DAO over SQLite → Postgres
+  later. A JSON-file store is a worse rehearsal than SQL.
+
+**Costs (real but bounded):** a new paradigm in a flat-file repo needs a
+`schema_version` table + idempotent startup migrations; SQLite serializes
+writers, so WAL mode + `busy_timeout` are required (a non-issue at a handful of
+bottles); test fixtures need temp DBs.
+
+**Scope of the store:** one DB at `~/.bot-bottle/bot-bottle.db` behind a thin
+repository API. Only the **new** metering/budget/enforcement-audit ledger lives
+there. Existing per-bottle blobs (resume `metadata.json`, transcripts,
+Dockerfile overrides, supervise queue) stay on the filesystem — migrating them
+now is churn for no benefit and they lack the concurrency/aggregation problem.
+
+### Host-level controller + dashboard
+
+A single **host-level controller** owns the meter, budget evaluation, and the
+cutoff actions across all bottles (cf. `bot_bottle/cli/supervise.py`'s
+cross-bottle view), rather than a per-bottle daemon. v1 ships one host-level
+**TUI dashboard** that reads live usage-vs-budget from the SQLite store and
+offers on-demand cutoff/stop. The existing supervisor UI should eventually fold
+into this same dashboard; this PRD lays the host-level surface it will move to.
+
+## Implementation chunks
+
+Ordered, individually mergeable:
+
+1. **SQLite repository foundation.** `~/.bot-bottle/bot-bottle.db`, schema +
+   `schema_version` migrations, WAL + `busy_timeout`, thin repository API,
+   temp-DB test fixtures. No behavior wired yet.
+2. **Metering at the egress proxy.** Parse authoritative response `usage`
+   (including SSE final-usage tailing) in the egress addon `response` hook;
+   write per-bottle / per-provider usage rows to the ledger.
+3. **`settings.yml` + budget model.** Host-level `~/.bot-bottle/settings.yml`
+   parsed by `yaml_subset.py`; budget precedence (agent → bottle → parent →
+   global) and the `--budget` launch flag.
+4. **Forced cutoff + cutoff policy.** Wire the threshold trigger to the
+   `cutoff` / `freeze` / `kill` primitives on the egress/backend plane; record
+   enforcement actions to the audit ledger.
+5. **Host-level TUI dashboard.** Live usage-vs-budget view + on-demand
+   cutoff/stop, reading the store.
+6. **`count_tokens` pre-flight gate (optional refinement).** Abstract method +
+   stdlib estimator default; Anthropic/OpenAI endpoints for built-in
+   claude/codex; optional pre-send block.
+
+## Open questions
+
+- **SSE usage tailing robustness.** Buffering streamed responses to extract the
+  final usage event without breaking the agent's own stream consumption — how
+  much of the body must the addon hold, and what's the failure mode if the
+  stream is interrupted mid-flight?
+- **Crossing mid-request.** A single response can push usage past budget only
+  *after* it's already been delivered. Is post-hoc cutoff (next request blocked)
+  sufficient, or is a pre-flight estimator gate (chunk 6) required for v1?
+- **Provider name ↔ metered host mapping.** How does the proxy attribute a
+  flow to an agent-provider budget key — by destination host, by bottle
+  identity, or both?
+- **Parent-bottle budget semantics.** For `bottle extends` (PRD 0025 / 0065)
+  chains, does "parent bottle" mean the manifest parent, the launching bottle,
+  or the full ancestry summed?
+- **Dashboard ↔ controller transport (even host-only).** In-process, a local
+  socket, or polling the SQLite store directly? Picks the seam the future remote
+  control plane will extend.
@@ -1,135 +0,0 @@
-# PRD prd-new: SQLite local storage
-
- **Status:** Active
- **Author:** codex
- **Created:** 2026-07-01
- **Issue:** #319
-
-## Summary
-
-Add a small stdlib SQLite storage layer for bot-bottle host runtime state,
-starting with the supervise queue and audit log. This replaces scattered JSON
-queue files and JSONL audit logs with structured tables while preserving the
-existing public supervise helper functions and sidecar queue mount contract.
-
-## Problem
-
-Bot-bottle currently stores supervise proposals and responses as individual JSON
-files under `~/.bot-bottle/queue/<slug>/`, and audit entries as JSONL files
-under `~/.bot-bottle/audit/`. That worked for the original interactive TUI, but
-new forge-native orchestration needs durable, queryable local state for queues,
-audit trails, watchdogs, and lifecycle records. PR #318 started introducing
-SQLite-shaped boilerplate for forge state; the storage foundation should live in
-its own PR so forge work can build on the shared runtime store instead of adding
-one-off persistence.
-
-## Goals / Success Criteria
-
-1. Supervise proposals and responses are persisted through SQLite.
-2. Audit entries are persisted through SQLite.
-3. Supervise queue helpers use the bottle slug / queue key instead of a queue
-   directory path.
-4. The sidecar receives the host database mount across docker, smolmachines,
-   and macOS-container backends.
-5. The implementation stays stdlib-only.
-6. 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.
@@ -1,188 +0,0 @@
-"""Unit: `cli.py start --headless` non-interactive launch path.
-
-Headless is the keystone for orchestrators, CI, and webhook
-dispatch: agent/bottles/label come from flags + manifest defaults, no
-TUI selectors fire, and the preflight y/N is auto-confirmed
-(`assume_yes=True`). All actual launch work is stubbed so no container
-is created.
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import MagicMock, patch
-
-import bot_bottle.cli.start as start_mod
-import bot_bottle.cli.tui as tui_mod
-from bot_bottle.backend import ActiveAgent
-from bot_bottle.log import Die
-from bot_bottle.manifest import ManifestError
-
-
-def _make_manifest(
-    agent_names: list[str],
-    bottle_names: list[str] | None = None,
-    agent_bottle: str = "",
-):
-    manifest = MagicMock()
-    manifest.agents = {name: MagicMock(bottle=agent_bottle) for name in agent_names}
-    manifest.all_agent_names = sorted(agent_names)
-    manifest.all_bottle_names = sorted(bottle_names or [])
-    manifest.home_md = None  # eager mode so _peek_agent_bottle uses agents dict
-    manifest.require_agent = MagicMock(return_value=None)
-    return manifest
-
-
-def _active_agent(slug: str) -> ActiveAgent:
-    return ActiveAgent(
-        backend_name="docker",
-        slug=slug,
-        agent_name="demo",
-        started_at="2026-01-01T00:00:00+00:00",
-        services=(),
-    )
-
-
-class TestCmdStartHeadless(unittest.TestCase):
-    """Drive `cmd_start --headless` with launch + TUI stubbed out."""
-
-    def setUp(self):
-        self._manifest = _make_manifest(
-            ["researcher", "implementer"], ["claude", "dev"], agent_bottle="claude"
-        )
-        patch(
-            "bot_bottle.cli.start.ManifestIndex.resolve",
-            return_value=self._manifest,
-        ).start()
-        self._launch_mock = patch(
-            "bot_bottle.cli.start._launch_bottle", return_value=0
-        ).start()
-        # No bottles running by default → no label collision.
-        patch(
-            "bot_bottle.cli.start.enumerate_active_agents", return_value=[]
-        ).start()
-        # If any TUI picker fires in headless mode, that's a bug.
-        self._agent_picker = patch.object(tui_mod, "filter_select").start()
-        self._bottle_picker = patch.object(tui_mod, "filter_multiselect").start()
-        self._modal = patch.object(tui_mod, "name_color_modal").start()
-        patch.dict(os.environ, {}, clear=False).start()
-        os.environ.pop("BOT_BOTTLE_BACKEND", None)
-        self.addCleanup(patch.stopall)
-
-    def _spec(self):
-        self._launch_mock.assert_called_once()
-        return self._launch_mock.call_args[0][0]
-
-    # -- no TUI in headless --------------------------------------------
-
-    def test_headless_fires_no_pickers(self):
-        rc = start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
-        )
-        self.assertEqual(0, rc)
-        self._agent_picker.assert_not_called()
-        self._bottle_picker.assert_not_called()
-        self._modal.assert_not_called()
-
-    def test_headless_assume_yes_forwarded(self):
-        start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
-        )
-        self.assertTrue(self._launch_mock.call_args[1]["assume_yes"])
-
-    # -- prompt --------------------------------------------------------
-
-    def test_headless_without_prompt_dies(self):
-        with self.assertRaises(Die):
-            start_mod.cmd_start(["--headless", "researcher", "--bottle", "claude"])
-        self._launch_mock.assert_not_called()
-
-    def test_headless_prompt_forwarded_to_launch(self):
-        start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "claude",
-             "--prompt", "Implement issue #42"]
-        )
-        self.assertEqual(
-            "Implement issue #42",
-            self._launch_mock.call_args[1]["headless_prompt_text"],
-        )
-
-    # -- bottle resolution ---------------------------------------------
-
-    def test_explicit_bottles_forwarded_in_order(self):
-        start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "dev", "--bottle", "claude",
-             "--prompt", "Do it"]
-        )
-        self.assertEqual(("dev", "claude"), self._spec().bottle_names)
-
-    def test_omitted_bottle_falls_back_to_agent_default(self):
-        start_mod.cmd_start(["--headless", "implementer", "--prompt", "Do it"])
-        self.assertEqual(("claude",), self._spec().bottle_names)
-
-    def test_no_bottle_and_no_default_dies(self):
-        manifest = _make_manifest(["researcher"], ["claude"], agent_bottle="")
-        with patch(
-            "bot_bottle.cli.start.ManifestIndex.resolve", return_value=manifest
-        ):
-            with self.assertRaises(Die):
-                start_mod.cmd_start(
-                    ["--headless", "researcher", "--prompt", "Do it"]
-                )
-        self._launch_mock.assert_not_called()
-
-    # -- agent resolution ----------------------------------------------
-
-    def test_missing_agent_name_dies(self):
-        with self.assertRaises(Die):
-            start_mod.cmd_start(["--headless"])
-        self._launch_mock.assert_not_called()
-
-    def test_unknown_agent_raises_manifest_error(self):
-        self._manifest.require_agent.side_effect = ManifestError("agent 'x' not defined")
-        with self.assertRaises(ManifestError):
-            start_mod.cmd_start(
-                ["--headless", "x", "--bottle", "claude", "--prompt", "Do it"]
-            )
-        self._launch_mock.assert_not_called()
-
-    # -- label / color -------------------------------------------------
-
-    def test_label_defaults_to_agent_name(self):
-        start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
-        )
-        self.assertEqual("researcher", self._spec().label)
-
-    def test_explicit_label_and_color_forwarded(self):
-        start_mod.cmd_start(
-            ["--headless", "researcher", "--bottle", "claude",
-             "--label", "nightly", "--color", "green", "--prompt", "Do it"]
-        )
-        spec = self._spec()
-        self.assertEqual("nightly", spec.label)
-        self.assertEqual("green", spec.color)
-
-    def test_label_collision_uniquifies(self):
-        with patch(
-            "bot_bottle.cli.start.enumerate_active_agents",
-            return_value=[_active_agent("researcher")],
-        ):
-            start_mod.cmd_start(
-                ["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
-            )
-        self.assertEqual("researcher-2", self._spec().label)
-
-    # -- backend wiring ------------------------------------------------
-
-    def test_backend_flag_forwarded(self):
-        start_mod.cmd_start(
-            ["--headless", "--backend=docker", "researcher", "--bottle", "claude",
-             "--prompt", "Do it"]
-        )
-        self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"])
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -107,7 +107,7 @@ def _egress_plan(
 def _supervise_plan() -> SupervisePlan:
    return SupervisePlan(
        slug=SLUG,
-        db_path=STATE / "bot-bottle.db",
+        queue_dir=STATE / "supervise" / "queue",
        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.assertIn("SUPERVISE_DB_PATH=/run/supervise/bot-bottle.db", env_strings)
+        self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
        self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))

    def test_volumes_always_includes_egress_ca(self):
@@ -408,7 +408,8 @@ 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.assertIn("/run/supervise/bot-bottle.db", targets)
+        self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
+                            for t in targets))

    def test_extra_hosts_omitted_for_git_upstreams(self):
        sc = self._render(with_git=True)["services"]["sidecars"]
@@ -74,7 +74,7 @@ def _plan(
    if supervise:
        supervise_plan = SupervisePlan(
            slug="demo-abc12",
-            db_path=Path("/tmp/bot-bottle.db"),
+            queue_dir=Path("/tmp/queue"),
        )
    return DockerBottlePlan(
        spec=spec,
@@ -343,14 +343,5 @@ class TestClaudeSuperviseMcp(unittest.TestCase):
        )


-class TestClaudeHeadlessPrompt(unittest.TestCase):
-    def test_returns_p_flag_and_prompt(self):
-        self.assertEqual(["-p", "Do the task"], ClaudeAgentProvider().headless_prompt("Do the task"))
-
-    def test_preserves_prompt_text_verbatim(self):
-        text = "Fix issue #42: the widget breaks on empty input"
-        self.assertEqual(["-p", text], ClaudeAgentProvider().headless_prompt(text))
-
-
 if __name__ == "__main__":
    unittest.main()
@@ -77,7 +77,7 @@ def _plan(
    if supervise:
        supervise_plan = SupervisePlan(
            slug="demo-abc12",
-            db_path=Path("/tmp/bot-bottle.db"),
+            queue_dir=Path("/tmp/queue"),
        )
    return DockerBottlePlan(
        spec=spec,
@@ -314,14 +314,5 @@ class TestCodexSuperviseMcp(unittest.TestCase):
        )


-class TestCodexHeadlessPrompt(unittest.TestCase):
-    def test_returns_prompt_as_positional_arg(self):
-        self.assertEqual(["Do the task"], CodexAgentProvider().headless_prompt("Do the task"))
-
-    def test_preserves_prompt_text_verbatim(self):
-        text = "Fix issue #42: the widget breaks on empty input"
-        self.assertEqual([text], CodexAgentProvider().headless_prompt(text))
-
-
 if __name__ == "__main__":
    unittest.main()
@@ -223,14 +223,5 @@ 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()
@@ -38,7 +38,6 @@ 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()
@@ -47,6 +47,7 @@ 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
@@ -212,6 +212,7 @@ 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"
@@ -385,10 +386,10 @@ def _fake_sv(response_status: str | None) -> types.SimpleNamespace:
    def _sha256_hex(_payload: Any) -> str:
        return "hash"

-    def _noop(*_args: Any) -> None:
+    def _noop(_a: Any, _b: Any) -> None:
        return None

-    def _read_response(_slug: Any, _pid: Any) -> Any:
+    def _read_response(_qd: Any, _pid: Any) -> Any:
        if response_status is None:
            raise OSError("not written yet")  # forces poll -> timeout
        return types.SimpleNamespace(status=response_status)
@@ -408,6 +409,7 @@ 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
@@ -630,13 +632,14 @@ 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(_p: Any) -> None:
+        def _raise(_qd: Any, _p: Any) -> None:
            raise OSError("disk full")

        fake.write_proposal = _raise
@@ -14,7 +14,6 @@ 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,
@@ -210,9 +209,8 @@ 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=_sv.TOOL_GITLEAKS_ALLOW", hook)
-        self.assertIn("_sv.write_proposal", hook)
-        self.assertIn("_sv.read_response", hook)
+        self.assertIn('"tool": "gitleaks-allow"', hook)
+        self.assertIn("SUPERVISE_QUEUE_DIR", hook)
        self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
        self.assertIn("supervisor approved # gitleaks:allow", hook)
        self.assertIn("supervisor rejected # gitleaks:allow", hook)
@@ -373,27 +371,6 @@ class TestDynamicKeyProvisioning(unittest.TestCase):
            self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
        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)

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

@@ -165,22 +165,6 @@ class TestAgentValidation(unittest.TestCase):
        with self.assertRaises(ManifestError):
            ManifestAgent.from_dict("a", {"skills": [5]}, set())

-    def test_skill_name_rejects_shell_metacharacters(self) -> None:
-        # Skill names become host/guest path segments interpolated into
-        # provisioning shell commands; anything outside kebab-case is
-        # rejected at load so it can never reach a `bottle.exec` string.
-        for bad in ("foo; rm -rf /", "../escape", "foo bar", "Foo", "-leading"):
-            with self.assertRaises(ManifestError):
-                ManifestAgent.from_dict("a", {"skills": [bad]}, set())
-
-    def test_skill_name_accepts_kebab_case(self) -> None:
-        agent = ManifestAgent.from_dict(
-            "a", {"skills": ["init-entry", "quality-eval", "skill0"]}, set()
-        )
-        self.assertEqual(
-            agent.skills, ("init-entry", "quality-eval", "skill0")
-        )
-
    def test_prompt_not_string(self) -> None:
        with self.assertRaises(ManifestError):
            ManifestAgent.from_dict("a", {"prompt": 5}, set())
@@ -49,7 +49,6 @@ 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()
@@ -130,7 +129,7 @@ def _plan(
    if supervise:
        supervise_plan = SupervisePlan(
            slug="demo-abc12",
-            db_path=Path("/tmp/bot-bottle.db"),
+            queue_dir=Path("/tmp/queue"),
        )
    return SmolmachinesBottlePlan(
        spec=spec,
@@ -422,14 +421,6 @@ 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(
@@ -1,5 +1,6 @@
 """Unit: supervise queue + audit log + diff helpers (PRD 0013)."""

+import json
 import tempfile
 import threading
 import time
@@ -18,7 +19,7 @@ from bot_bottle.supervise import (
    TOOL_EGRESS_ALLOW,
    TOOL_GITLEAKS_ALLOW,
    archive_proposal,
-    host_db_path,
+    audit_log_path,
    list_pending_proposals,
    read_audit_entries,
    read_proposal,
@@ -111,44 +112,32 @@ class TestResponseRoundtrip(unittest.TestCase):
 class TestQueueIO(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory(prefix="bot-bottle-supervise-test.")
-        self._home_patch = self._patch_home(Path(self._tmp.name))
-        self.slug = "dev"
+        self.queue_dir = Path(self._tmp.name)

    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(p)
+        path = write_proposal(self.queue_dir, p)
        self.assertTrue(path.exists())
-        self.assertEqual(host_db_path(), path)
        self.assertEqual(0o600, path.stat().st_mode & 0o777)
-        loaded = read_proposal(self.slug, p.id)
+        loaded = read_proposal(self.queue_dir, p.id)
        self.assertEqual(p, loaded)

    def test_list_pending_excludes_responded(self):
        a = _proposal(justification="first")
        b = _proposal(justification="second")
-        write_proposal(a)
-        write_proposal(b)
-        write_response(self.slug, Response(
+        write_proposal(self.queue_dir, a)
+        write_proposal(self.queue_dir, b)
+        write_response(self.queue_dir, Response(
            proposal_id=a.id, status=STATUS_APPROVED, notes="",
        ))
-        pending = list_pending_proposals(self.slug)
+        pending = list_pending_proposals(self.queue_dir)
        self.assertEqual([b.id], [p.id for p in pending])

-    def test_list_pending_returns_empty_for_missing_slug(self):
-        self.assertEqual([], list_pending_proposals("nope"))
+    def test_list_pending_returns_empty_for_missing_dir(self):
+        self.assertEqual([], list_pending_proposals(self.queue_dir / "nope"))

    def test_list_pending_sorted_by_arrival(self):
        # Fabricate two with explicit timestamps.
@@ -165,30 +154,30 @@ class TestQueueIO(unittest.TestCase):
            now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
        )
        # Write in reverse order.
-        write_proposal(b)
-        write_proposal(a)
-        ordered = list_pending_proposals(self.slug)
+        write_proposal(self.queue_dir, b)
+        write_proposal(self.queue_dir, a)
+        ordered = list_pending_proposals(self.queue_dir)
        self.assertEqual([a.id, b.id], [p.id for p in ordered])

    def test_write_and_read_response(self):
        r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no")
-        write_response(self.slug, r)
-        self.assertEqual(r, read_response(self.slug, "xyz"))
+        write_response(self.queue_dir, r)
+        self.assertEqual(r, read_response(self.queue_dir, "xyz"))

    def test_wait_for_response_returns_when_file_appears(self):
        p = _proposal()
-        write_proposal(p)
+        write_proposal(self.queue_dir, p)

        def write_after_delay():
            time.sleep(0.05)
-            write_response(self.slug, Response(
+            write_response(self.queue_dir, Response(
                proposal_id=p.id, status=STATUS_APPROVED, notes="ok",
            ))

        t = threading.Thread(target=write_after_delay)
        t.start()
        try:
-            r = wait_for_response(self.slug, p.id, poll_interval=0.01)
+            r = wait_for_response(self.queue_dir, p.id, poll_interval=0.01)
        finally:
            t.join()
        self.assertEqual(STATUS_APPROVED, r.status)
@@ -198,24 +187,25 @@ class TestQueueIO(unittest.TestCase):
        deadline = time.monotonic() + 0.05
        with self.assertRaises(TimeoutError):
            wait_for_response(
-                self.slug, "never",
+                self.queue_dir, "never",
                poll_interval=0.01, deadline=deadline,
            )

-    def test_archive_proposal_hides_rows(self):
+    def test_archive_proposal_moves_both_files(self):
        p = _proposal()
-        write_proposal(p)
-        write_response(self.slug, Response(
+        write_proposal(self.queue_dir, p)
+        write_response(self.queue_dir, Response(
            proposal_id=p.id, status=STATUS_APPROVED, notes="",
        ))
-        archive_proposal(self.slug, p.id)
-        self.assertEqual([], list_pending_proposals(self.slug))
-        with self.assertRaises(FileNotFoundError):
-            read_response(self.slug, p.id)
+        archive_proposal(self.queue_dir, p.id)
+        self.assertFalse((self.queue_dir / f"{p.id}.proposal.json").exists())
+        self.assertFalse((self.queue_dir / f"{p.id}.response.json").exists())
+        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.proposal.json").exists())
+        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.response.json").exists())

    def test_archive_is_idempotent_on_missing_files(self):
        # Should not raise.
-        archive_proposal(self.slug, "nope")
+        archive_proposal(self.queue_dir, "nope")


 class TestAuditLog(unittest.TestCase):
@@ -247,7 +237,6 @@ 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)
@@ -263,13 +252,12 @@ class TestAuditLog(unittest.TestCase):
                justification="",
                diff="",
            ))
-        entries = read_audit_entries("egress", "dev")
-        self.assertEqual(3, len(entries))
-        self.assertEqual(
-            ["2026-05-25T12:00:00+00:00", "2026-05-25T12:00:01+00:00",
-             "2026-05-25T12:00:02+00:00"],
-            [entry.timestamp for entry in entries],
-        )
+        path = audit_log_path("egress", "dev")
+        with path.open() as f:
+            lines = [line for line in f if line.strip()]
+        self.assertEqual(3, len(lines))
+        for line in lines:
+            self.assertTrue(json.loads(line))  # each line is valid JSON

    def test_separate_logs_per_component_slug(self):
        write_audit_entry(AuditEntry(
@@ -391,7 +379,7 @@ class TestSupervisePrepare(unittest.TestCase):

    def test_prepare_creates_queue(self):
        plan = _StubSupervise().prepare("dev", self.stage_dir)
-        self.assertTrue(plan.db_path.is_file())
+        self.assertTrue(plan.queue_dir.is_dir())
        self.assertEqual("dev", plan.slug)
        self.assertEqual("", plan.internal_network)

@@ -77,7 +77,9 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):

    def test_walks_all_slug_subdirs(self):
        for slug in ("dev", "api"):
-            supervise.write_proposal(_proposal(slug=slug))
+            qdir = supervise.queue_dir_for_slug(slug)
+            qdir.mkdir(parents=True)
+            supervise.write_proposal(qdir, _proposal(slug=slug))
        pending = supervise_cli.discover_pending()
        self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})

@@ -95,14 +97,18 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
            now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
        )
        for p in (late, early):
-            supervise.write_proposal(p)
+            qdir = supervise.queue_dir_for_slug(p.bottle_slug)
+            qdir.mkdir(parents=True, exist_ok=True)
+            supervise.write_proposal(qdir, p)
        pending = supervise_cli.discover_pending()
        self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending])

    def test_excludes_already_responded(self):
        p = _proposal()
-        supervise.write_proposal(p)
-        supervise.write_response("dev", supervise.Response(
+        qdir = supervise.queue_dir_for_slug("dev")
+        qdir.mkdir(parents=True)
+        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())
@@ -117,8 +123,10 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):

    def _enqueue(self, tool: str = TOOL_EGRESS_ALLOW):
        p = _proposal(tool=tool)
-        supervise.write_proposal(p)
-        return supervise_cli.QueuedProposal(proposal=p)
+        qdir = supervise.queue_dir_for_slug("dev")
+        qdir.mkdir(parents=True, exist_ok=True)
+        supervise.write_proposal(qdir, p)
+        return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)

    def test_approve_writes_response(self):
        qp = self._enqueue()
@@ -127,7 +135,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
            return_value=("routes: []\n", "routes:\n  - host: example.com\n"),
        ):
            supervise_cli.approve(qp)
-        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_APPROVED, resp.status)
        self.assertIsNone(resp.final_file)

@@ -142,7 +150,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
                final_file="routes:\n  - host: edited.example.com\n",
                notes="tweaked",
            )
-        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_MODIFIED, resp.status)
        self.assertEqual("routes:\n  - host: edited.example.com\n", resp.final_file)
        self.assertEqual("tweaked", resp.notes)
@@ -150,7 +158,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.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_REJECTED, resp.status)
        self.assertEqual("nope", resp.notes)

@@ -173,33 +181,36 @@ 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 DB for the response; TUI must not archive it.
-        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
+        # Gate polls the queue dir for the response; TUI must not archive it.
+        resp = read_response(qp.queue_dir, 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.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, 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 DB for the response; the TUI must
+        # The egress addon polls the queue dir for the response; the TUI must
        # not archive it (the addon archives after reading).
-        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_APPROVED, resp.status)
        self.assertEqual("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)
@@ -211,13 +222,14 @@ 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.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual("legit", resp.notes)

    def test_suffix_for_token_allow_is_txt(self):
@@ -4,6 +4,7 @@ fallback paths."""

 from __future__ import annotations

+import os
 import tempfile
 import time
 import unittest
@@ -11,19 +12,14 @@ 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,
 )


@@ -41,53 +37,58 @@ 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_missing_row(self) -> None:
+    def test_read_proposal_non_dict(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
-                    self.assertRaises(FileNotFoundError):
-                read_proposal("slug", "p")
+            (Path(d) / "p.proposal.json").write_text("[]")
+            with self.assertRaises(ValueError):
+                read_proposal(Path(d), "p")

-    def test_read_response_missing_row(self) -> None:
+    def test_read_response_non_dict(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
-                    self.assertRaises(FileNotFoundError):
-                read_response("slug", "p")
+            (Path(d) / "p.response.json").write_text("[]")
+            with self.assertRaises(ValueError):
+                read_response(Path(d), "p")

-    def test_list_pending_reads_db_only(self) -> None:
+    def test_list_pending_skips_malformed(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}):
-                supervise.write_proposal(_proposal())
-                pending = list_pending_proposals("slug")
+            qd = Path(d)
+            (qd / "bad.proposal.json").write_text("{ not json")
+            (qd / "arr.proposal.json").write_text("[]")
+            (qd / "incomplete.proposal.json").write_text("{}")  # from_dict raises
+            supervise.write_proposal(qd, _proposal())  # one valid
+            pending = list_pending_proposals(qd)
            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:
-            with patch.dict("os.environ", {"HOME": d}):
+            qd = Path(d)
            p = _proposal()
-                supervise.write_proposal(p)
-                supervise.write_response("slug", supervise.Response(
-                    proposal_id=p.id,
-                    status=STATUS_APPROVED,
-                    notes="",
-                ))
-                self.assertEqual([], list_pending_proposals("slug"))
+            supervise.write_proposal(qd, p)
+            (qd / f"{p.id}.response.json").write_text("{}")  # response exists -> skipped
+            self.assertEqual([], list_pending_proposals(qd))


 class TestWaitForResponse(unittest.TestCase):
-    def test_missing_response_times_out(self) -> None:
+    def test_malformed_response_then_timeout(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
-                    self.assertRaises(TimeoutError):
-                wait_for_response("slug", "p", deadline=time.monotonic())
+            (Path(d) / "p.response.json").write_text("{ not json")
+            with self.assertRaises(TimeoutError):
+                wait_for_response(Path(d), "p", deadline=time.monotonic())

-    def test_empty_db_response_does_not_count(self) -> None:
+    def test_incomplete_response_then_timeout(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
-                    self.assertRaises(TimeoutError):
-                wait_for_response("slug", "p", deadline=time.monotonic())
+            (Path(d) / "p.response.json").write_text("{}")  # dict but from_dict raises
+            with self.assertRaises(TimeoutError):
+                wait_for_response(Path(d), "p", deadline=time.monotonic())


 class TestReadAuditEntries(unittest.TestCase):
@@ -96,94 +97,35 @@ class TestReadAuditEntries(unittest.TestCase):
                patch.dict("os.environ", {"HOME": home}):
            self.assertEqual([], read_audit_entries("egress", "nope"))

-    def test_reads_entries_from_db(self) -> None:
-        with tempfile.TemporaryDirectory() as home, \
-                patch.dict("os.environ", {"HOME": home}):
-            write_audit_entry(AuditEntry(
-                timestamp="t",
-                bottle_slug="slug",
-                component="egress",
-                operator_action="approve",
-                operator_notes="",
-                justification="",
-                diff="",
-            ))
-            write_audit_entry(AuditEntry(
-                timestamp="t",
-                bottle_slug="other",
-                component="egress",
-                operator_action="reject",
-                operator_notes="",
-                justification="",
-                diff="",
-            ))
-            entries = read_audit_entries("egress", "slug")
-            self.assertEqual(1, len(entries))
-            self.assertEqual("approve", entries[0].operator_action)
-
-    def test_legacy_audit_log_file_does_not_count(self) -> None:
+    def test_skips_malformed_lines(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(
+            valid = (
                '{"timestamp": "t", "bottle_slug": "slug", "component": "egress",'
                ' "operator_action": "approve", "operator_notes": "",'
-                ' "justification": "", "diff": ""}\n'
+                ' "justification": "", "diff": ""}'
+            )
+            path.write_text(
+                "\n"               # blank line skipped
+                "{ not json\n"     # JSONDecodeError skipped
+                "[]\n"             # not a dict skipped
+                "{}\n"             # missing fields -> ValueError skipped
+                + valid + "\n"
            )
            entries = read_audit_entries("egress", "slug")
-            self.assertEqual([], entries)
+            self.assertEqual(1, len(entries))
+            self.assertEqual("approve", entries[0].operator_action)


-class TestStoreGuardBranches(unittest.TestCase):
-    """Direct QueueStore / AuditStore construction and early-return guard branches."""
-
-    def test_queue_store_explicit_db_path(self):
-        with tempfile.TemporaryDirectory() as d:
-            db = Path(d) / "q.db"
-            store = QueueStore("key", db_path=db)
-            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
+class TestFlockFallback(unittest.TestCase):
+    def test_flock_on_closed_fd_is_swallowed(self) -> None:
+        # flock on a closed fd raises OSError(EBADF), which the helpers swallow.
+        fd = os.open(os.devnull, os.O_RDONLY)
+        os.close(fd)
+        supervise._try_flock(fd)
+        supervise._try_funlock(fd)


 if __name__ == "__main__":
@@ -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")
+        config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
        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 patch.object(_sv, "write_proposal", side_effect=OSError("disk full")), \
-                self.assertRaises(_RpcInternalError) as cm:
+        with self.assertRaises(_RpcInternalError) as cm:
            handle_tools_call(
                {
                    "name": _sv.TOOL_EGRESS_ALLOW,
@@ -265,31 +265,21 @@ class TestHandleToolsList(unittest.TestCase):
 class TestHandleToolsCall(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory(prefix="supervise-server-test.")
-        self._home_patch = self._patch_home(Path(self._tmp.name))
-        self.config = ServerConfig(bottle_slug="dev")
+        self.queue_dir = Path(self._tmp.name)
+        self.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)

    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("dev")
+                pending = _sv.list_pending_proposals(self.queue_dir)
                if pending:
                    p = pending[0]
-                    _sv.write_response("dev", _sv.Response(
+                    _sv.write_response(self.queue_dir, _sv.Response(
                        proposal_id=p.id, status=status, notes=notes,
                    ))
                    return
@@ -422,11 +412,15 @@ class TestHandleToolsCall(unittest.TestCase):
        finally:
            responder.join()
        # No pending proposals left after archive.
-        self.assertEqual([], _sv.list_pending_proposals("dev"))
+        self.assertEqual([], _sv.list_pending_proposals(self.queue_dir))
+        # Both files moved to processed/.
+        processed = list((self.queue_dir / "processed").glob("*.json"))
+        self.assertEqual(2, len(processed))

    def test_pending_response_times_out_without_archive(self):
        config = ServerConfig(
            bottle_slug="dev",
+            queue_dir=self.queue_dir,
            response_timeout_seconds=0.05,
        )
        result = handle_tools_call(
@@ -444,7 +438,8 @@ class TestHandleToolsCall(unittest.TestCase):
        text = result["content"][0]["text"]  # type: ignore[index]
        self.assertIn("status: pending", text)
        self.assertIn("proposal remains queued", text)
-        self.assertEqual(1, len(_sv.list_pending_proposals("dev")))
+        self.assertEqual(1, len(_sv.list_pending_proposals(self.queue_dir)))
+        self.assertFalse((self.queue_dir / "processed").exists())


 class TestHandleListEgressRoutes(unittest.TestCase):
@@ -466,7 +461,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"),
+                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
            )

        self.assertFalse(result["isError"])  # type: ignore[index]
@@ -481,7 +476,7 @@ class TestHandleListEgressRoutes(unittest.TestCase):
        with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
            result = handle_list_egress_routes(
                {},
-                ServerConfig(bottle_slug="dev"),
+                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
            )

        self.assertTrue(result["isError"])  # type: ignore[index]
@@ -549,6 +544,7 @@ 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()
@@ -556,7 +552,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")
+        self.server.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True,
        )