docs(research): refine open/paid boundary — orchestrator as paid control plane

Captures the four-turn working-through of the monetization line under the forge-as-orchestrator shape: - The orchestrator IS the control plane and can be closed/private from day one; the runtime stays OSS. - Charge for the moat (see-inside-the-run + cross-run aggregation), not the webhook/orchestration plumbing the forge vendors build free. - Heuristic: single-run/single-node = free; cross-run aggregation + central enforcement + identity/fleet = paid (== individual vs team). - Provenance: emit signed provenance via a free API (tamper-evident offline, BYO-SIEM); sell retention/search/policy. Forge footer is an optional off-by-default consumer, not the audit record. - On-prem priority: self-hosted runners > self-hosted provenance; sell the governed fleet, not a single runner (which is just the free runtime). - Fly = metered capacity line, not the moat; self-host == same closed control plane licensed, not a separate product. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WL77TgFxKbs3cidGMG9dz7
docs(research): add forge-native orchestration as the delivery vehicle
2026-06-30 18:57:04 -04:00 · 2026-06-29 12:02:23 -04:00 · 2026-06-29 11:43:33 -04:00
41 changed files with 972 additions and 1441 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,10 +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/migrations.py        /app/migrations.py
 COPY bot_bottle/db_store.py          /app/db_store.py
 COPY bot_bottle/queue_store.py       /app/queue_store.py
 COPY bot_bottle/audit_store.py       /app/audit_store.py
 COPY bot_bottle/supervise.py         /app/supervise.py
 COPY bot_bottle/supervise_server.py  /app/supervise_server.py
 COPY bot_bottle/sidecar_init.py      /app/sidecar_init.py
@@ -85,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/)
+[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
-[![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
+[![core coverage](https://img.shields.io/badge/core%20coverage-96%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
 **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
@@ -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,113 +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
 try:
    from .db_store import DbStore
    from .migrations import TableMigrations
 except ImportError:
    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 def get_supervise_mod() -> object:
    """Lazy import of supervise to avoid a circular-import at module init time.
    Mirrors our own module identity so patches on supervise.bot_bottle_root
    propagate correctly in both flat (sidecar / sys.path-injection tests) and
    package contexts."""
    import sys
    sv_name = "supervise" if __name__ == "audit_store" else "bot_bottle.supervise"
    if sv_name in sys.modules:
        return sys.modules[sv_name]
    try:
        import bot_bottle.supervise as _m
    except ImportError:
        import supervise as _m  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
    return _m
 # One entry per schema version: _MIGRATIONS.migrations[0] brings a fresh DB
 # to version 1, [1] to version 2, and so on.  Add new migrations at the end;
 # never edit existing ones.
 _MIGRATIONS = TableMigrations("audit_store", [
    # v1 — initial schema
    """
    CREATE TABLE IF NOT EXISTS supervise_audit_entries (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT NOT NULL,
        bottle_slug TEXT NOT NULL,
        component TEXT NOT NULL,
        operator_action TEXT NOT NULL,
        operator_notes TEXT NOT NULL,
        justification TEXT NOT NULL,
        diff TEXT NOT NULL
    )
    """,
 ])
 class AuditStore(DbStore):
    """SQLite-backed persistent store for supervise audit entries."""
    def __init__(self, db_path: Path | None = None) -> None:
        resolved = db_path or get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
        super().__init__(resolved, _MIGRATIONS)
    def write_audit_entry(self, entry: AuditEntry) -> Path:
        with self._connect() as conn:
            conn.execute(
                """
                INSERT INTO supervise_audit_entries (
                    timestamp, bottle_slug, component, operator_action,
                    operator_notes, justification, diff
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    entry.timestamp,
                    entry.bottle_slug,
                    entry.component,
                    entry.operator_action,
                    entry.operator_notes,
                    entry.justification,
                    entry.diff,
                ),
            )
        self._chmod()
        return self.db_path
    def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]:
        if not self.db_path.is_file():
            return []
        with self._connect() as conn:
            rows = conn.execute(
                """
                SELECT * FROM supervise_audit_entries
                WHERE component = ? AND bottle_slug = ?
                ORDER BY id
                """,
                (component, slug),
            ).fetchall()
        return [self._row_to_entry(row) for row in rows]
    @staticmethod
    def _row_to_entry(row: sqlite3.Row) -> AuditEntry:
        m = get_supervise_mod()
        return m.AuditEntry(  # type: ignore[attr-defined]
            timestamp=row["timestamp"],
            bottle_slug=row["bottle_slug"],
            component=row["component"],
            operator_action=row["operator_action"],
            operator_notes=row["operator_notes"],
            justification=row["justification"],
            diff=row["diff"],
        )
 __all__ = ["AuditStore"]
@@ -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),
+            "source": str(sp.queue_dir),
-            "target": DB_PATH_IN_CONTAINER,
+            "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 (
+from ...git_gate import revoke_git_gate_provisioned_keys
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import info, warn
 from . 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 (
+from ...git_gate import revoke_git_gate_provisioned_keys
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import die, info, warn
-from ...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 (
+from ...git_gate import revoke_git_gate_provisioned_keys
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import info, warn
 from ...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
+    The queue dir is intentionally NOT under here — it lives at
-    database, so they survive state-dir cleanup."""
+    ~/.bot-bottle/queue/<slug>/ alongside the audit logs, so it
    survives state-dir cleanup."""
    return bottle_state_dir(identity) / _SUPERVISE_SUBDIR
@@ -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.
+    attaches claude, and prints the resume hint on session end."""
    `assume_yes` skips the interactive y/N confirmation (headless /
    orchestrator launches), where there is no human at the prompt.
    `headless_prompt_text` is passed to the provider's `headless_prompt`
    method and the resulting args are appended to startup_args so the
    agent receives the initial task without interactive input."""
    stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
    identity = ""
    try:
@@ -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."""
+    """Walk ~/.bot-bottle/queue/* and collect pending proposals."""
-    out = [
+    queue_root = _supervise.bot_bottle_root() / "queue"
-        QueuedProposal(proposal=proposal)
+    if not queue_root.is_dir():
-        for proposal in list_all_pending_proposals()
+        return []
-    ]
+    out: list[QueuedProposal] = []
    for slug_dir in sorted(queue_root.iterdir()):
        if not slug_dir.is_dir():
            continue
        for proposal in list_pending_proposals(slug_dir):
            out.append(QueuedProposal(proposal=proposal, queue_dir=slug_dir))
    out.sort(key=lambda q: q.proposal.arrival_timestamp)
    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="")
@@ -313,9 +313,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")
@@ -279,9 +279,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")
@@ -315,9 +315,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")
@@ -1,40 +0,0 @@
 """Shared SQLite-backed store base class for bot-bottle (PRD 0013)."""
 from __future__ import annotations
 import sqlite3
 from pathlib import Path
 try:
    from .migrations import TableMigrations
 except ImportError:
    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 class DbStore:
    """Base for SQLite-backed stores. Subclasses resolve db_path then call super().__init__."""
    def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
        self.db_path = db_path
        self._migrations = migrations
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init()
    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn
    def _init(self) -> None:
        with self._connect() as conn:
            self._migrations.apply(conn)
        self._chmod()
    def _chmod(self) -> None:
        try:
            self.db_path.chmod(0o600)
        except OSError:
            pass
 __all__ = ["DbStore"]
@@ -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)
+        response = await self._await_token_response(queue_dir, proposal.id)
-        _sv.archive_proposal(self._supervise_slug, proposal.id)
+        _sv.archive_proposal(queue_dir, proposal.id)
        if response is not None and response.status in (
            _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
+        For `gitea` key entries, also generates and registers
-        has an empty identity file. Backend launch fills that in after
+        a fresh deploy key via the forge API and writes the private key
-        the operator confirms the preflight.
+        + key ID to `stage_dir`.
        Returned plan is incomplete: the launch step must fill
        `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(
+proposal_id = str(uuid.uuid4())
-    bottle_slug=slug,
+proposal = {
-    tool=_sv.TOOL_GITLEAKS_ALLOW,
+    "id": proposal_id,
-    proposed_file=payload,
+    "bottle_slug": slug,
-    justification=(
+    "tool": "gitleaks-allow",
    "proposed_file": payload,
    "justification": (
        "git-gate found gitleaks findings hidden by # gitleaks:allow; "
        "approve only for dummy test fixtures or confirmed false positives"
    ),
-    current_file_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+    "arrival_timestamp": datetime.datetime.now(
-    now=datetime.datetime.now(datetime.timezone.utc),
+        datetime.timezone.utc
-)
+    ).isoformat(),
-_sv.write_proposal(proposal)
+    "current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
-print(proposal.id)
+}
 queue = Path(queue_dir)
 queue.mkdir(parents=True, exist_ok=True)
 path = queue / f"{proposal_id}.proposal.json"
 tmp = path.with_suffix(path.suffix + ".tmp")
 with tmp.open("w", encoding="utf-8") as f:
    json.dump(proposal, f, indent=2)
    f.write("\n")
 os.chmod(tmp, 0o600)
 os.replace(tmp, path)
 print(proposal_id)
 PY
  )
  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])
+    with open(sys.argv[1], encoding="utf-8") as f:
-except FileNotFoundError:
+        raw = json.load(f)
-    sys.exit(2)
+except (OSError, json.JSONDecodeError):
-print(response.status)
+    sys.exit(1)
 status = raw.get("status")
 if not isinstance(status, str):
    sys.exit(1)
 print(status)
 PY
-    )
+      ) || status=""
    rc=$?
    if [ "$rc" -eq 2 ]; then
      status=""
    elif [ "$rc" -ne 0 ]; then
      status="invalid"
    fi
    if [ -n "$status" ]; then
      case "$status" in
        approved|modified)
-          python3 - "$slug" "$proposal_id" <<'PY' || true
+          mkdir -p "$queue_dir/processed"
-import sys
+          mv -f "$queue_dir/${proposal_id}.proposal.json" "$queue_dir/processed/" 2>/dev/null || true
-
+          mv -f "$queue_dir/${proposal_id}.response.json" "$queue_dir/processed/" 2>/dev/null || true
 from bot_bottle import supervise as _sv
 _sv.archive_proposal(sys.argv[1], sys.argv[2])
 PY
          echo "git-gate: supervisor approved # gitleaks:allow for $ref" >&2
          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
@@ -1,37 +0,0 @@
 """SQLite migration runner for bot-bottle stores."""
 from __future__ import annotations
 import sqlite3
 class TableMigrations:
    """Runs a sequential list of DDL migrations tracked by schema_key in schema_versions."""
    def __init__(self, schema_key: str, migrations: list[str]) -> None:
        self.schema_key = schema_key
        self.migrations = migrations
    def apply(self, conn: sqlite3.Connection) -> None:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS schema_versions (
                module TEXT PRIMARY KEY,
                version INTEGER NOT NULL DEFAULT 0
            )
            """
        )
        row = conn.execute(
            "SELECT version FROM schema_versions WHERE module = ?",
            (self.schema_key,),
        ).fetchone()
        version = row[0] if row else 0
        for i, sql in enumerate(self.migrations[version:], start=version + 1):
            conn.execute(sql)
            conn.execute(
                "INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
                (self.schema_key, i),
            )
 __all__ = ["TableMigrations"]
@@ -1,240 +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
 try:
    from .db_store import DbStore
    from .migrations import TableMigrations
 except ImportError:
    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 def get_supervise_mod() -> object:
    """Lazy import of supervise to avoid a circular-import at module init time.
    By the time any QueueStore method is called, both modules are fully loaded.
    Mirrors our own module identity: when we are 'queue_store' (sidecar flat
    context or tests that inject bot_bottle/ into sys.path) we use the flat
    'supervise' module so that patches on supervise.bot_bottle_root propagate
    correctly. When we are 'bot_bottle.queue_store' we use 'bot_bottle.supervise'."""
    import sys
    sv_name = "supervise" if __name__ == "queue_store" else "bot_bottle.supervise"
    if sv_name in sys.modules:
        return sys.modules[sv_name]
    try:
        import bot_bottle.supervise as _m
    except ImportError:
        import supervise as _m  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
    return _m
 # One entry per schema version: _MIGRATIONS.migrations[0] brings a fresh DB
 # to version 1, [1] to version 2, and so on.  Add new migrations at the end;
 # never edit existing ones.
 _MIGRATIONS = TableMigrations("queue_store", [
    # v1 — proposals table
    """
    CREATE TABLE IF NOT EXISTS supervise_proposals (
        queue_key TEXT NOT NULL,
        id TEXT NOT NULL,
        bottle_slug TEXT NOT NULL,
        tool TEXT NOT NULL,
        proposed_file TEXT NOT NULL,
        justification TEXT NOT NULL,
        arrival_timestamp TEXT NOT NULL,
        current_file_hash TEXT NOT NULL,
        archived INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (queue_key, id)
    )
    """,
    # v2 — responses table
    """
    CREATE TABLE IF NOT EXISTS supervise_responses (
        queue_key TEXT NOT NULL,
        proposal_id TEXT NOT NULL,
        status TEXT NOT NULL,
        notes TEXT NOT NULL,
        final_file TEXT,
        archived INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (queue_key, proposal_id)
    )
    """,
 ])
 class QueueStore(DbStore):
    """SQLite-backed persistent store for supervise proposals and responses."""
    def __init__(self, queue_key: str, db_path: Path | None = None) -> None:
        self.queue_key = queue_key
        if db_path is not None:
            resolved = db_path
        else:
            # In the sidecar container SUPERVISE_DB_PATH points at the
            # bind-mounted host DB. On the host this env var is never set,
            # so we always fall through to host_db_path().
            env_path = os.environ.get("SUPERVISE_DB_PATH", "").strip()
            resolved = Path(env_path) if env_path else get_supervise_mod().host_db_path()  # type: ignore[attr-defined]
        super().__init__(resolved, _MIGRATIONS)
    def write_proposal(self, proposal: Proposal) -> Path:
        with self._connect() as conn:
            conn.execute(
                """
                INSERT OR REPLACE INTO supervise_proposals (
                    queue_key, id, bottle_slug, tool, proposed_file, justification,
                    arrival_timestamp, current_file_hash, archived
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
                """,
                (
                    self.queue_key,
                    proposal.id,
                    proposal.bottle_slug,
                    proposal.tool,
                    proposal.proposed_file,
                    proposal.justification,
                    proposal.arrival_timestamp,
                    proposal.current_file_hash,
                ),
            )
        self._chmod()
        return self.db_path
    def read_proposal(self, proposal_id: str) -> Proposal:
        with self._connect() as conn:
            row = conn.execute(
                """
                SELECT * FROM supervise_proposals
                WHERE queue_key = ? AND id = ? AND archived = 0
                """,
                (self.queue_key, proposal_id),
            ).fetchone()
        if row is None:
            raise FileNotFoundError(proposal_id)
        return self._row_to_proposal(row)
    def list_pending_proposals(self) -> list[Proposal]:
        if not self.db_path.is_file():
            return []
        with self._connect() as conn:
            rows = conn.execute(
                """
                SELECT p.* FROM supervise_proposals p
                WHERE p.archived = 0
                  AND p.queue_key = ?
                  AND NOT EXISTS (
                    SELECT 1 FROM supervise_responses r
                    WHERE r.queue_key = p.queue_key
                      AND r.proposal_id = p.id
                      AND r.archived = 0
                  )
                ORDER BY p.arrival_timestamp, p.id
                """,
                (self.queue_key,),
            ).fetchall()
        return [self._row_to_proposal(row) for row in rows]
    def list_all_pending_proposals(self) -> list[Proposal]:
        if not self.db_path.is_file():
            return []
        with self._connect() as conn:
            rows = conn.execute(
                """
                SELECT p.* FROM supervise_proposals p
                WHERE p.archived = 0
                  AND NOT EXISTS (
                    SELECT 1 FROM supervise_responses r
                    WHERE r.queue_key = p.queue_key
                      AND r.proposal_id = p.id
                      AND r.archived = 0
                  )
                ORDER BY p.arrival_timestamp, p.id
                """
            ).fetchall()
        return [self._row_to_proposal(row) for row in rows]
    def write_response(self, response: Response) -> Path:
        with self._connect() as conn:
            conn.execute(
                """
                INSERT OR REPLACE INTO supervise_responses (
                    queue_key, proposal_id, status, notes, final_file, archived
                ) VALUES (?, ?, ?, ?, ?, 0)
                """,
                (
                    self.queue_key,
                    response.proposal_id,
                    response.status,
                    response.notes,
                    response.final_file,
                ),
            )
        self._chmod()
        return self.db_path
    def read_response(self, proposal_id: str) -> Response:
        with self._connect() as conn:
            row = conn.execute(
                """
                SELECT * FROM supervise_responses
                WHERE queue_key = ? AND proposal_id = ? AND archived = 0
                """,
                (self.queue_key, proposal_id),
            ).fetchone()
        if row is None:
            raise FileNotFoundError(proposal_id)
        return self._row_to_response(row)
    def archive_proposal(self, proposal_id: str) -> None:
        if not self.db_path.is_file():
            return
        with self._connect() as conn:
            conn.execute(
                """
                UPDATE supervise_proposals SET archived = 1
                WHERE queue_key = ? AND id = ?
                """,
                (self.queue_key, proposal_id),
            )
            conn.execute(
                """
                UPDATE supervise_responses SET archived = 1
                WHERE queue_key = ? AND proposal_id = ?
                """,
                (self.queue_key, proposal_id),
            )
    @staticmethod
    def _row_to_proposal(row: sqlite3.Row) -> Proposal:
        m = get_supervise_mod()
        return m.Proposal(  # type: ignore[attr-defined]
            id=row["id"],
            bottle_slug=row["bottle_slug"],
            tool=row["tool"],
            proposed_file=row["proposed_file"],
            justification=row["justification"],
            arrival_timestamp=row["arrival_timestamp"],
            current_file_hash=row["current_file_hash"],
        )
    @staticmethod
    def _row_to_response(row: sqlite3.Row) -> Response:
        m = get_supervise_mod()
        return m.Response(  # type: ignore[attr-defined]
            proposal_id=row["proposal_id"],
            status=row["status"],
            notes=row["notes"],
            final_file=row["final_file"],
        )
 __all__ = ["QueueStore"]
@@ -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
+approve / modify / reject, and writes a response file alongside the
-the response and returns `{status, notes}` to the agent.
+proposal. The sidecar sees the response and returns `{status, notes}`
 to the agent.
 This module defines the host-side library: dataclasses for the queue
-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:
+def _proposal_filename(proposal_id: str) -> str:
-    """Persist `proposal` in the queue database, mode 0o600.
+    return f"{proposal_id}.proposal.json"
 def _response_filename(proposal_id: str) -> str:
    return f"{proposal_id}.response.json"
 def _id_from_proposal_filename(path: Path) -> str | None:
    name = path.name
    if not name.endswith(".proposal.json"):
        return None
    return name[: -len(".proposal.json")]
 def write_proposal(queue_dir: Path, proposal: Proposal) -> Path:
    """Persist `proposal` as JSON in the queue dir, mode 0o600.
    Directory is created if missing."""
-    return QueueStore(proposal.bottle_slug).write_proposal(proposal)
+    queue_dir.mkdir(parents=True, exist_ok=True)
    path = queue_dir / _proposal_filename(proposal.id)
    payload = json.dumps(proposal.to_dict(), indent=2) + "\n"
    _atomic_write(path, payload, mode=0o600)
    return path
-def read_proposal(bottle_slug: str, proposal_id: str) -> Proposal:
+def read_proposal(queue_dir: Path, proposal_id: str) -> Proposal:
-    return QueueStore(bottle_slug).read_proposal(proposal_id)
+    path = queue_dir / _proposal_filename(proposal_id)
    with path.open() as f:
        raw = json.load(f)
    if not isinstance(raw, dict):
        raise ValueError(f"{path}: top-level must be an object")
    return Proposal.from_dict(raw)
-def list_pending_proposals(bottle_slug: str) -> list[Proposal]:
+def list_pending_proposals(queue_dir: Path) -> list[Proposal]:
-    """All proposals for `bottle_slug` that do not yet have a matching
+    """All proposals in `queue_dir` that do not yet have a matching
-    response. Sorted by `arrival_timestamp` so the operator
+    response file. Sorted by `arrival_timestamp` so the operator
    sees the queue FIFO."""
-    return QueueStore(bottle_slug).list_pending_proposals()
+    if not queue_dir.is_dir():
        return []
    out: list[Proposal] = []
    for path in sorted(queue_dir.glob("*.proposal.json")):
        proposal_id = _id_from_proposal_filename(path)
        if proposal_id is None:
            continue
        if (queue_dir / _response_filename(proposal_id)).exists():
            continue
        try:
            with path.open() as f:
                raw = json.load(f)
        except (OSError, json.JSONDecodeError):
            continue
        if not isinstance(raw, dict):
            continue
        try:
            out.append(Proposal.from_dict(raw))
        except (KeyError, ValueError):
            continue
    out.sort(key=lambda p: p.arrival_timestamp)
    return out
-def list_all_pending_proposals() -> list[Proposal]:
+def write_response(queue_dir: Path, response: Response) -> Path:
-    """All pending proposals across bottles, sorted FIFO."""
+    queue_dir.mkdir(parents=True, exist_ok=True)
-    return QueueStore("").list_all_pending_proposals()
+    path = queue_dir / _response_filename(response.proposal_id)
    payload = json.dumps(response.to_dict(), indent=2) + "\n"
    _atomic_write(path, payload, mode=0o600)
    return path
-def write_response(bottle_slug: str, response: Response) -> Path:
+def read_response(queue_dir: Path, proposal_id: str) -> Response:
-    return QueueStore(bottle_slug).write_response(response)
+    path = queue_dir / _response_filename(proposal_id)
-
+    with path.open() as f:
-
+        raw = json.load(f)
-def read_response(bottle_slug: str, proposal_id: str) -> Response:
+    if not isinstance(raw, dict):
-    return QueueStore(bottle_slug).read_response(proposal_id)
+        raise ValueError(f"{path}: top-level must be an object")
    return Response.from_dict(raw)
 def wait_for_response(
-    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."""
+    Polls the filesystem so the implementation stays portable and
-    store = QueueStore(bottle_slug)
+    stdlib-only."""
    path = queue_dir / _response_filename(proposal_id)
    while True:
        if path.exists():
            try:
-            return store.read_response(proposal_id)
+                with path.open() as f:
-        except FileNotFoundError:
+                    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:
+def archive_proposal(queue_dir: Path, proposal_id: str) -> None:
-    """Mark both proposal and response rows processed.
+    """Move both proposal and response files to `<queue_dir>/processed/`.
-    Idempotent — missing rows are silently skipped."""
+    Idempotent — missing files are silently skipped."""
-    QueueStore(bottle_slug).archive_proposal(proposal_id)
+    processed = queue_dir / "processed"
    processed.mkdir(parents=True, exist_ok=True)
    for name in (_proposal_filename(proposal_id), _response_filename(proposal_id)):
        src = queue_dir / name
        if src.exists():
            src.rename(processed / name)
 # --- Audit log -------------------------------------------------------------
 def write_audit_entry(entry: AuditEntry) -> Path:
-    """Append `entry` to the host supervise audit table."""
+    """Append `entry` as one JSON-Lines record to the per-bottle
-    return AuditStore().write_audit_entry(entry)
+    audit log. Acquires an advisory exclusive lock so concurrent
    writers don't interleave bytes."""
    path = audit_log_path(entry.component, entry.bottle_slug)
    path.parent.mkdir(parents=True, exist_ok=True)
    line = json.dumps(entry.to_dict(), sort_keys=False) + "\n"
    fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
    try:
        _try_flock(fd)
        try:
            os.write(fd, line.encode("utf-8"))
        finally:
            _try_funlock(fd)
    finally:
        os.close(fd)
    return path
 def read_audit_entries(component: str, slug: str) -> list[AuditEntry]:
-    """Load all audit entries for the given component+slug."""
+    """Load all audit entries for the given component+slug. Empty
-    return AuditStore().read_audit_entries(component, slug)
+    list if the log doesn't exist."""
    path = audit_log_path(component, slug)
    if not path.is_file():
        return []
    out: list[AuditEntry] = []
    with path.open() as f:
        for raw_line in f:
            raw_line = raw_line.strip()
            if not raw_line:
                continue
            try:
                raw = json.loads(raw_line)
            except json.JSONDecodeError:
                continue
            if not isinstance(raw, dict):
                continue
            try:
                out.append(AuditEntry(
                    timestamp=_require_str(raw, "timestamp"),
                    bottle_slug=_require_str(raw, "bottle_slug"),
                    component=_require_str(raw, "component"),
                    operator_action=_require_str(raw, "operator_action"),
                    operator_notes=_require_str(raw, "operator_notes"),
                    justification=_require_str(raw, "justification"),
                    diff=_require_str(raw, "diff"),
                ))
            except ValueError:
                continue
    return out
 # --- Diff rendering --------------------------------------------------------
@@ -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
+    `queue_dir` is the host directory bind-mounted into the sidecar
-    /run/supervise/bot-bottle.db. `internal_network` is empty at
+    at /run/supervise/queue. `internal_network` is empty at prepare
-    prepare time; the backend's launch step fills it via
+    time; the backend's launch step fills it via dataclasses.replace
-    dataclasses.replace before calling .start."""
+    before calling .start."""
    slug: str
-    db_path: Path
+    queue_dir: Path
    internal_network: str = ""
 class Supervise(ABC):
-    """Per-bottle supervise sidecar. Encapsulates host-side database
+    """Per-bottle supervise sidecar. Encapsulates the host-side
-    staging; the sidecar's start/stop lifecycle is backend-specific."""
+    prepare (queue dir staging); the sidecar's start/stop lifecycle
    is backend-specific."""
    def prepare(
        self,
        slug: str,
        stage_dir: Path,
    ) -> SupervisePlan:
-        """Stage the host database. Returns the plan; `internal_network`
+        """Stage the per-bottle queue dir on the host. Returns the
-        must be set by the launch step before .start runs."""
+        plan; `internal_network` must be set by the launch step before
        .start runs."""
        del stage_dir
-        db_path = host_db_path()
+        queue_dir = queue_dir_for_slug(slug)
-        QueueStore(slug)
+        queue_dir.mkdir(parents=True, exist_ok=True)
        AuditStore(db_path)
        return SupervisePlan(
            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.
+  2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
-  3. Blocks polling for a matching Response row.
+     the host's ~/.bot-bottle/queue/<slug>/).
  3. Blocks polling for a matching Response file.
  4. Returns the operator's `{status, notes}` to the agent.
 The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
-container creation by the backend's start step). SUPERVISE_DB_PATH
+container creation by the backend's start step). The queue dir comes
-points at the bind-mounted host database.
+from SUPERVISE_QUEUE_DIR (default `/run/supervise/queue`).
 Speaks MCP over HTTP+JSON-RPC. Methods handled:
@@ -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,
@@ -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.
@@ -0,0 +1,490 @@
 # Monetization & competitive positioning
 Where, if anywhere, bot-bottle has a paid wedge — given a 2026
 competitive field that has largely commoditized "sandbox a coding
 agent." Folds together the agent-provider-agnostic framing, the Fly
 remote-backend idea, the supervisor/egress-audit play, and the
 solo-dev/Linux brand instinct, then asks the only question that
 matters: is there a viable path to revenue that the competition does
 not already foreclose?
 Companion to
 [`agent-sandbox-landscape.md`](agent-sandbox-landscape.md) (the
 isolation-tech survey),
 [`built-in-supervisor-design.md`](built-in-supervisor-design.md) (the
 supervise surface this would extend), and
 [`secret-minimization-over-dlp.md`](secret-minimization-over-dlp.md)
 (why custody, not detection, is the real moat).
 Market data current as of June 2026.
 ## Summary
 **Verdict: a path exists, but it is narrow, and it is not the path the
 project is currently shaped for.** Every individual property bot-bottle
 leans on — isolation, BYO-image, egress filtering, OSS, self-hosting —
 is matched by some competitor, and several are now *free* from the agent
 vendors themselves. There is exactly one defensible position left: the
 **bundle** that no single competitor occupies —
 > uniform egress audit + secret custody + policy, across *heterogeneous
 > coding agents you don't trust*, on your infra or a managed pool.
 Monetization is viable **only** if the product is sold as cross-vendor
 **fleet governance + egress audit for teams**, not as solo-dev agent
 safety (which the labs give away free). The solo-dev/Linux/anti-corporate
 energy is real and worth using — but as a *distribution and trust*
 engine that drives bottom-up adoption into teams, never as the revenue
 positioning itself. Get those two wires crossed and the business dies:
 you'd be courting the lowest-willingness-to-pay audience on earth while
 repelling the only buyer who pays.
 Net: **viable, conditional, and unforgiving of positioning error.** Do
 Phase 1 (self-hostable egress-audit dashboard) regardless — it's
 low-risk and it's the demo that makes everything else legible. Gate the
 go/no-go on whether 5–10 teams confirm they'd pay for cross-vendor
 egress audit *before* building the hosted tier.
 ## The two axes of "agnostic"
 bot-bottle differentiates on two orthogonal axes, and conflating them
 muddies the pitch:
 1. **Agent-provider agnostic** — run Claude Code, Codex, Aider, a local
   model, behind one control layer. Already real in the code
   (`agent_provider.py`, Claude/Codex templates, BYO Dockerfile). This
   is the axis the labs *structurally cannot* match — Anthropic only
   runs Claude, OpenAI only their models. Durable.
 2. **Compute backend** — local (docker / Apple Container / smolmachines)
   today; a remote **Fly** backend would add a managed pool. This is the
   axis that makes "fleet" literal for orgs and opens metered billing.
   Fly is a strong first remote backend because it also subsumes remote
   spin-up (Machines API) and the tunnel problem (6PN/WireGuard) — but
   "provider-agnostic compute" should be *earned* after backend #2, not
   designed up front (premature generalization trap).
 ## Competitive field, by capability
 The field doesn't have one competitor; it has a different set on each
 capability bot-bottle touches. Five dimensions:
 | Capability | Who has it | bot-bottle's standing |
 | :-- | :-- | :-- |
 | **Isolation / sandbox** | Anthropic & OpenAI **native, free**; OSS devcontainer wrappers; E2B/Modal/Daytona/Northflank | Commoditized. Not a wedge. |
 | **Arbitrary BYO Docker image** | Sandbox PaaS (E2B/Modal/Daytona/Northflank) yes; **managed agents: ~none** (Codex = fixed `codex-universal` + setup scripts; Copilot "not supported"; Devin/Jules constrained) | Wedge **vs. managed agents** (structural: it's their infra). Table stakes vs. PaaS. |
 | **Egress audit + alerts** | LLM-observability tools (Braintrust/Langfuse/Phoenix/Helicone/Datadog) — but on *model calls*, wrong layer. Network-egress security (DeepInspect, AI gateways) — right layer, but decoupled from the agent, not cross-vendor. Sandbox PaaS = gateway/filter, not an audit surface. | **~Nobody in bot-bottle's exact shape** (per-agent egress, tied to the sandbox, with DLP context, cross-vendor). This is the wedge. |
 | **OSS / self-hosting** | Managed agents: ~none. Sandbox PaaS: ~half (E2B OSS+self-host; Northflank BYOC; Modal closed; **Daytona leaving OSS**). Devcontainer wrappers: ~all. Observability: several. | Real wedge **vs. managed agents only**. Table stakes vs. PaaS, zero differentiation vs. wrappers. |
 | **Cross-vendor uniformity** | Nobody — the labs won't, PaaS is agent-neutral infra not agent-aware control, wrappers are single-tool | Wedge. The connective tissue of the whole position. |
 The pattern: **isolation and OSS/self-host are commodity; BYO-image and
 cross-vendor are wedges only against the managed agents; egress-audit in
 the integrated form is the one thing genuinely unoccupied.**
 ## Where bot-bottle is alone vs. where it's table stakes
 - **Alone (the moat):** egress audit + secret custody + policy, *tied to
  the agent sandbox*, *with DLP context* (which secret, which host,
  which agent/task), *uniform across vendors*. No competitor bundles
  these. An enterprise *could* bolt DeepInspect-style egress monitoring
  onto a sandbox, so the defensibility is the **integration and
  per-agent context**, not "we can see egress."
 - **Table stakes (do not lead with these):** "we sandbox agents" (free
  from the labs), "we're open source" (E2B is; the wrapper crowd all
  is), "we self-host" (Northflank BYOC, E2B, every wrapper).
 ## The two existential competitive facts
 1. **The agent vendors ship good-enough sandboxing for free.** Claude
   Code now has Seatbelt/bubblewrap + a network proxy natively; Codex
   has its own sandbox + approvals. This compresses the *single-vendor,
   single-dev* market to ~zero willingness-to-pay. It is *why* the
   product must be cross-vendor fleet governance, not local agent
   safety.
 2. **Northflank is converging from the infra side.** It already ships
   dedicated egress gateways + proxy-based secret injection + BYOC.
   It is the nearest thing to bot-bottle's differentiator as a managed
   platform — but infra-first and agent-neutral, not agent-aware,
   cross-vendor, or audit-first. Watch it.
 ## Monetization path (sequenced)
 Open-core: **give away the sandbox, charge for the control plane.**
 - **Phase 0 — validate (1–2 wks, parallel).** Ask 5–10 teams running 2+
  agents: would you pay for one egress-audit + policy plane across
  Claude *and* Codex? Gate the rest on a yes.
 - **Phase 1 — the wedge (self-hostable, OSS).** Multi-bottle egress
  dashboard + web approval queue + exportable audit log, built over the
  existing `supervise_server.py` JSON-RPC and the egress event levels
  (`LOG_BLOCKS` / `LOG_FULL`). Low risk, half-built, and the 30-second
  demo that sells everything. The compliance hook (75% of enterprises
  rank auditability #1) lives here.
 - **Phase 2 — the paywall (hosted team tier).** Multi-tenant supervisor:
  SSO/RBAC, audit retention, alerting, **centralized policy push**
  (define egress allowlist + DLP once, enforce across all agents —
  the moat made concrete). Gate on team/compliance features, *never* on
  the core security.
 - **Phase 3 — Fly remote backend.** Managed agent pool → "fleet" becomes
  literal; metered (agent-hours) billing; subsumes remote spin-up +
  tunnel.
 - **Phase 4 — deepen.** Second agent provider done deeply (lean
  open-source/open-weight for rug-pull resistance); egress anomaly
  detection (the DLP stream becomes a product); SOC2/audit-export for
  larger buyers.
 **Do not build first:** the p2p mobile app (least monetizable, 6PN
 gives the tunnel free), a generic multi-cloud abstraction (premature),
 or the hosted SaaS before Phase 0.
 ## Brand vs. revenue: the solo-dev / Linux instinct
 The instinct to court Linux/hacker/solo-dev users and stay "not too
 corporate" is **right for distribution, dangerous as strategy.**
 - **Right:** it's how OSS infra gets discovered and trusted (HN, stars,
  word-of-mouth, security-circle vouching); authenticity is a real moat
  vs. the corporate players *because the architecture sincerely embodies
  it* (local-first, `$HOME` trust boundary, no phone-home); and it fits
  the founder.
 - **Dangerous:** that audience is the lowest-WTP cohort that exists
  (self-hosts the free thing, forks rather than pays), and "not too
  corporate" reads to a VP of Eng as "not enterprise-ready." Building an
  anti-SaaS brand and then shipping a paid tier invites the sell-out /
  rug-pull backlash — which **Daytona just triggered** going closed.
 **Resolution — be Tailscale, not a manifesto.** Use the developer-first,
 respects-you energy as the *funnel*; sell *through* the solo advocate,
 bottom-up, into the team that pays. Two guardrails:
 1. "Anti-corporate" must not mean "anti-team-features." SSO/RBAC/audit
   retention *are* the monetization; build them in a developer-respecting
   way (Tailscale has SSO and is still beloved). Tone is the brand; team
   features are the product.
 2. Set the open-core social contract publicly **on day one** — core
   sandbox open and self-hostable forever; hosted control plane is how
   the lights stay on. The communities that don't revolt are the ones
   told the deal upfront.
 Concrete: the README frames the Docker/**Linux** backend as "legacy."
 If courting the Linux crowd, make the Linux path (Docker+gVisor,
 libkrun/smolmachines) first-class in the docs, not the fallback.
 ## Individuals, mobile, and the Pi-ecosystem reality check
 "Individual devs won't pay" (above) is too blunt and needs refining.
 The accurate claim: individuals won't pay for **safety-as-insurance**
 (abstract risk reduction the labs give away free), but they *do* pay for
 **capability/convenience felt daily** — Claude Pro, Cursor, Tailscale
 Personal. "Drive my self-hosted agent from my phone" is capability, not
 insurance, so it has a real (low-priced, high-churn) WTP profile. The
 self-hoster/Linux crowd specifically pays for **sovereignty/control**,
 just not for enterprise insurance. So an individual "sovereign remote
 agent access" tier is *not* unreasonable in principle.
 **But the market has already run that experiment, in public, for free.**
 The Pi ecosystem (pi.dev) has commoditized every convenience layer an
 individual product would charge for:
 | Capability | Already free/OSS | bot-bottle differentiates? |
 | :-- | :-- | :-- |
 | Remote control from mobile | remote-pi, Paseo, TelePi | ❌ commoditized |
 | Multi-agent orchestration from mobile | Paseo, pi-agent-dashboard | ❌ commoditized |
 | **Launch** new agents from mobile | Paseo (`paseo run`) | ❌ commoditized |
 | Launch into a **sandboxed, egress-audited** env | nobody | ✅ the moat |
 Paseo (`getpaseo/paseo`, on the App Store) does the full thing an
 individual remote-control tier would charge for — launch *and* attach
 agents on a laptop/VM/dev-server, driven from mobile over an E2E relay —
 free and open source. It *orchestrates* agents; it does **not** sandbox them, run
 an egress chokepoint, DLP-scan, or audit. None of the Pi-ecosystem tools
 do. So the residue, yet again, is **isolation + governance**, not
 remote/launch convenience.
 Two takeaways:
 1. **Don't compete on orchestration/launch/remote UX** — it's a solved,
   free, fast-moving, App-Store-shipping space around Pi. You won't win
   it and it isn't the moat.
 2. **Be the safe runtime orchestrators launch *into*.** Launch-from-mobile
   is table stakes; *launch-into-a-sealed-egress-audited-bottle* is the
   differentiator. bot-bottle is the sandbox an orchestrator like Paseo
   would target, or that you wrap thin orchestration around — never the
   orchestrator itself.
 Capability layers commoditize fast: every individual/mobile angle
 probed in this analysis collapsed back to the same cross-vendor +
 sandbox + egress-audit + custody bundle. Mobile remote belongs as a
 *funnel delighter* on top of the team product, not a standalone paid
 line.
 ## Forge-native orchestration as the delivery vehicle
 The strongest concrete *product shape* for the moat is not a bespoke
 dashboard and not a Paseo competitor — it is **the git forge as the
 orchestrator, with bot-bottle as the safe runtime it launches into.**
 The forge already provides, for free, everything an orchestrator would
 otherwise have to build: identity (agent/bot users, signed commits),
 state (issues, labels, PRs/MRs, comments), triggers (webhooks, CI,
 comment commands), review (diffs, approvals, status checks), audit
 (commits/comments/reviews), and permissions (repo access, protected
 branches, token scopes). bot-bottle supplies the one thing the forge
 doesn't: **least-privilege, secret-isolated, audited execution of
 untrusted agents.** Same moat (custody + audit + policy), better
 vehicle — and it lands the product where teams already live, so it
 avoids building an agent dashboard before one is needed.
 The flow is essentially free to assemble:
 ```
 issue/PR/MR event → webhook → policy/router → assign agent user +
 branch/worktree → run agent in an isolated bottle (no ambient secrets)
 → commit as agent identity → open PR/MR → CI + human review + merge
 ```
 **Crowding (why this is less saturated than it looks):**
 | Layer | How crowded |
 | :-- | :-- |
 | Generic multi-agent orchestrators (worktree/TUI/dashboard) | very — 50–100+ |
 | Forge-native issue/PR/MR orchestration | moderate — ~10–30 serious |
 | Self-hostable, least-privilege, audited, forge-portable | **single digits** |
 The deeper you go toward *untrusted-agent safety + auditability +
 self-hostable + forge-portable*, the emptier it gets.
 **The GitHub/GitLab first-party trap → lead Gitea + sovereignty.**
 GitHub (Agentic Workflows, Copilot coding agent) and GitLab (Duo Agent
 Platform) are the forge *vendors* building native issue-to-PR agent
 orchestration with native identity/permissions/audit. On their turf you
 lose the integration-depth battle the same way single-vendor agent
 safety loses to Anthropic/OpenAI — the same "incumbent ships it free,
 deeper" dynamic, one layer up. So the durable opening is **Gitea +
 self-hosted** (no first-party agent platform exists — the open Gitea
 feature request for an AI code agent confirms the vacuum) plus
 **cross-forge *untrusted-agent* safety**, which no forge vendor will
 build because they want you running *their* agent, not arbitrary ones
 under uniform least-privilege across competitors' forges. Cross-vendor
 neutrality, applied to forges.
 **Buyer reconciliation.** The least-crowded opening (self-hosted Gitea)
 overlaps the lowest-WTP crowd (indie self-hosters), while the paying
 teams sit on GitHub/GitLab where first-party competition is fiercest.
 The intersection that resolves it: **orgs running self-hosted forges for
 sovereignty/compliance reasons** (regulated, air-gapped, security-
 conscious, on-prem). They have budget, they run self-hosted GitLab/Gitea,
 *and* shipping code to a cloud agent vendor is a non-starter — so "run
 untrusted agents sandboxed, least-privilege, fully audited, inside our
 forge, on our infra" is a procurement checkbox, not a nicety. That is
 where "least-crowded" finally meets "has money."
 **Separate moat-hard-parts from cost-hard-parts.** The orchestration
 "hard parts" are two different things, and conflating them oversells the
 fit:
 | Moat (your differentiated strength) | Undifferentiated cost (everyone faces) |
 | :-- | :-- |
 | permission isolation | idempotency / dedupe / run ledger |
 | secret handling under malicious prompts | concurrency, locks, cancellation |
 | run provenance | queueing / scheduling / cleanup |
 | policy language | merge-conflict handling (~27% agent-PR conflict rate) |
 The right column is generic distributed-systems plumbing that wins you
 nothing and that merge-conflict resolution especially is a *different
 competency* from sandbox/custody. Keep it thin in the MVP; do not build a
 policy DSL + durable ledger + conflict resolver before one org pays.
 **The killer feature: run provenance on every agent PR.** A check/comment
 answering — which agent, which model, which prompt, which base commit,
 which policy, which tools, which network egress, which test results —
 attached at the moment a human reviews. It renders the (invisible)
 custody + egress-audit work as a PR artifact the buyer sees at the exact
 trust-decision point. No forge vendor's first-party agent will show you
 "here is everything the untrusted agent could reach." Build this first.
 **MVP** (`@bot-bottle fix this`): create an isolated worktree/bottle →
 check out the issue branch → run the selected harness as a named agent
 user → deny ambient secrets by default → record prompt/model/tools/policy
 → commit with bot identity → open PR/MR → attach the run-provenance
 footer (log + tests + permission/egress summary) → require human merge.
 The security model *is* the product. This rides the headless launch
 primitive directly: webhook → `start --headless` into an isolated bottle
 → commit as agent identity → PR with provenance.
 Open-core line, refined in the next section: the trigger *convention*
 (label/assignee) stays open so anyone can adopt it, but the
 **orchestrator that receives webhooks and governs lifecycle is the paid
 control plane**; the runtime — and a signed-provenance emission API —
 stay free.
 ## The open/paid boundary, refined: orchestrator as the paid control plane
 The forge-native shape sharpens the open-core line past the rough
 "trigger free, execution paid" cut above. Working it through four
 constraints — value capture, provenance integrity, the sovereignty
 buyer, and what the forge *structurally cannot do* — yields a precise
 boundary.
 **The orchestrator is the control plane, and the control plane is the
 paid product.** With the forge supplying identity / state / triggers /
 review, bot-bottle's orchestrator (`bot-bottle-orchestrator`, already
 specced as a separate binary in the forge-native PRD) is where webhooks
 land and bottle lifecycle + governance live. That binary can stay
 **closed/private from day one** without breaking the open-core contract:
 the runtime stays OSS; the control plane is how the lights stay on. This
 is "give away the sandbox, charge for the control plane" made literal —
 the orchestrator *is* the control plane.
 **Charge for the moat, not the webhook.** Holding webhooks and managing
 bottle lifecycle is commodity — the forge vendors build it first-party,
 and it's the "undifferentiated cost" column above (idempotency, queueing,
 dispatch). If the pitch is "we catch the webhook," they out-build it
 free. The paid value is the two things the forge *cannot* do:
 1. **See inside the run** — which model / prompt / policy / tools / egress
   produced the diff, whether a secret nearly left. Runtime-level data
   only the bottle holds.
 2. **Aggregate and enforce across runs** — retain / search / export every
   run across every repo; push one egress/DLP/capability policy
   fleet-wide and detect drift.
 The explainable heuristic: **anything legible within a single run on a
 single node is free; anything requiring cross-run aggregation, central
 enforcement, or identity/fleet management is paid.** That is also the
 individual-vs-team line — individuals live in single runs, teams need the
 aggregate.
 **Provenance: emit free (signed), sell the product.** The forge is the
 wrong system of record for provenance — a markdown footer is mutable by
 any maintainer, unsigned, per-PR, with no aggregation, so a maintainer
 could simply edit it. The authoritative record therefore lives in the
 (paid) control plane. The *runtime* emits **signed** provenance through a
 **free API** — tamper-evident offline (edit it and the signature breaks;
 verify with no server), so on-prem teams can route it into their own
 SIEM. What's paid is the *product* over that stream: retention, search,
 cross-run, export, policy. Whether a copy also lands in the PR footer is
 an optional, off-by-default marketing dial — one consumer of the free
 API, not a free provenance surface, and never the audit record. The
 mutability "bug" becomes a paid feature: the control plane flags *"PR
 footer edited / doesn't match the signed run."* (Prometheus model:
 `/metrics` is free to scrape; managed retention + dashboards are the
 business.)
 **On-prem priority: self-hosted runners over self-hosted provenance.**
 The sovereignty buyer's *hard structural constraint* is where the agent
 **executes** against private code, secrets, and network — that's the
 runner, and it cannot leave the perimeter. Audit metadata is softer; many
 regulated orgs ship logs to SaaS while keeping the workload inside. So:
 - Self-hosted **runner** = baseline, always, for that buyer.
 - Self-hosted **provenance store** = premium tier of the strictest subset
  (air-gapped, hard data-residency) — and largely covered by the free
  emission API → their own SIEM, so it may never need to be a product you
  build.
 - Precision so you don't trip your own free tier: a single self-hosted
  runner *is the OSS runtime on their box* — free. What's paid is the
  **fleet control plane**: enrolling/managing many runners, central
  policy push, dispatch/identity/quota, health/scaling. You don't sell
  "a runner," you sell **running a governed fleet**.
 **Resulting tiers:**
 | Layer | What it is | Open/Paid | Deployment |
 | :-- | :-- | :-- | :-- |
 | **Runtime** | isolation + ephemeral bottles, cred-proxy, supervise, `start --headless`, signed-provenance emission API | Free / OSS | Always self-host |
 | **Single runner** | the OSS runtime on a box | Free / OSS | Self-host |
 | **Control plane** | cross-run audit retention/search/export, central policy push, SSO/RBAC dispatch, fleet management of runners, alerting | **Paid** | Hosted *or* self-host-licensed — same code |
 | **Capacity** | managed Fly runner pool, metered (agent-hours) | **Paid add-on** | Hosted only |
 Fly stays a **capacity/convenience line, not the moat** — it monetizes
 even solo hackers (capability, not insurance), but a managed runner pool
 is reselling compute against Fly/E2B/Northflank on price. It's a bundle
 attached to the governance, never the thing defended. Self-host is *not*
 a separate product: on-prem buyers get the same closed control plane,
 licensed, pointed at their own runners.
 ## Risks to the thesis
 - **Lab encroachment.** If Anthropic/OpenAI add cross-agent governance
  or open their managed egress logs, the wedge narrows. Mitigate by
  going deep on cross-vendor + custody + audit *now*, while they're
  single-vendor.
 - **Rug-pull dependency.** You run the labs' agents; they can restrict
  their agent to their own sandbox via ToS/tech. Hedge toward
  open-source/open-weight agents for durability.
 - **Northflank (or E2B) ships agent-aware audit.** Plausible from the
  infra side. Your defense is agent-awareness + the supervise approval
  loop + cross-vendor, not raw egress visibility.
 - **WTP may simply not be there.** The honest failure mode: teams like
  the audit but won't pay because "we already sandbox in CI." Phase 0
  exists to find this out cheaply before building Phase 2/3.
 - **Forge-vendor encroachment (forge-native path).** GitHub Agentic
  Workflows / Copilot and GitLab Duo are first-party and deepening.
  Defense: aim at self-hosted Gitea + sovereignty buyers where no
  first-party agent platform exists, and at cross-forge untrusted-agent
  neutrality the vendors won't build. Don't fight them GitHub-native.
 - **Orchestration-reliability scope creep.** The forge-native build
  drags in idempotency, queueing, concurrency, and merge-conflict
  handling — undifferentiated plumbing that isn't the moat. Keep it thin
  until a paying org forces it.
 ## Recommendation
 Build Phase 1 now — it's low-risk, half-built, and the proof artifact.
 Run Phase 0 in parallel. Treat a clear yes from 5–10 teams as the
 green light for the hosted tier; treat a soft maybe as a signal to stay
 an excellent OSS tool with a tip-jar/support model rather than a
 venture-shaped SaaS. The technology is not the risk — the codebase is
 exemplary and the architecture already supports the pivot. The risk is
 **positioning discipline**: sell cross-vendor fleet governance to teams,
 use the indie brand as the funnel, and never let the anti-corporate
 aesthetic veto the features that pay.
 ## Sources
 - Anthropic — Claude Code sandboxing:
  https://www.anthropic.com/engineering/claude-code-sandboxing
 - OpenAI Codex — cloud environments:
  https://developers.openai.com/codex/cloud/environments ;
  custom-image feature request:
  https://community.openai.com/t/feature-request-custom-docker-images/1265333
 - GitHub Copilot — custom container image (not supported), discussion
  #194105: https://github.com/orgs/community/discussions/194105
 - DeepInspect — AI egress monitoring:
  https://www.deepinspect.ai/blog/ai-egress-monitoring
 - Braintrust — AI agent observability/alerting:
  https://www.braintrust.dev/articles/best-ai-agent-observability-tools-2026
 - E2B (OSS, Apache-2.0): https://github.com/e2b-dev/e2b ;
  infra/self-host: https://github.com/e2b-dev/infra
 - Daytona going closed source:
  https://www.daytona.io/dotfiles/updates/daytona-is-going-closed-source
 - Northflank — BYOC / egress gateways:
  https://northflank.com/blog/what-is-byoc-in-cloud-computing ;
  https://northflank.com/blog/self-hostable-alternatives-to-e2b-for-ai-agents
 - Modal Sandboxes: https://modal.com/products/sandboxes
 - AI agent orchestration / enterprise governance (75% cite
  auditability):
  https://viston.tech/ai-agent-orchestration-in-2026-moving-from-pilots-to-enterprise-wide-execution/
 - Pi harness (provider-agnostic CLI): https://pi.dev/packages/remote-pi ;
  https://github.com/earendil-works/pi
 - Paseo (launch + attach agents from desktop/mobile, OSS):
  https://github.com/getpaseo/paseo ;
  https://apps.apple.com/us/app/paseo-remote-coding-agents/id6758887924
 - pi-agent-dashboard (mobile-first remote control via mDNS/zrok):
  https://github.com/BlackBeltTechnology/pi-agent-dashboard
 - TelePi (Telegram remote control for Pi):
  https://futurelab.studio/blog/telepi-telegram-remote-control-for-pi/
 - Forge-native landscape (provided via conversation, not independently
  re-verified):
  - awesome-agent-orchestrators (50+ generic orchestrators):
    https://github.com/andyrewlee/awesome-agent-orchestrators
  - GitHub Agentic Workflows (first-party repo automation):
    https://github.blog/ai-and-ml/automate-repository-tasks-with-github-agentic-workflows/
  - GitLab Duo Agent Platform GA:
    https://ir.gitlab.com/news/news-details/2026/GitLab-Announces-the-General-Availability-of-GitLab-Duo-Agent-Platform/default.aspx
  - ai-review (cross-forge review incl. Gitea):
    https://github.com/Nikita-Filonov/ai-review
  - Gitea feature request — AI code agent (the vacuum):
    https://github.com/go-gitea/gitea/issues/34527
  - Phoenix — safe GitHub issue resolution (label-based webhook state
    machine): https://arxiv.org/abs/2606.20243
  - AgenticFlict — ~27% merge-conflict rate in agent PRs:
    https://arxiv.org/abs/2604.03551
@@ -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('"tool": "gitleaks-allow"', hook)
-        self.assertIn("_sv.write_proposal", hook)
+        self.assertIn("SUPERVISE_QUEUE_DIR", hook)
        self.assertIn("_sv.read_response", hook)
        self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
        self.assertIn("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(
+        SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
            db_path=Path("/state/bot-bottle.db"),
        )
        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,
        )
@@ -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.queue_dir = Path(self._tmp.name)
        self.slug = "dev"
    def tearDown(self):
        self._home_patch()
        self._tmp.cleanup()
    def _patch_home(self, fake_home: Path):
        original = supervise.bot_bottle_root
        def fake_root() -> Path:
            return fake_home / ".bot-bottle"
        supervise.bot_bottle_root = fake_root  # type: ignore[assignment]
        return lambda: setattr(supervise, "bot_bottle_root", original)
    def test_write_and_read_proposal(self):
        p = _proposal()
-        path = write_proposal(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(self.queue_dir, a)
-        write_proposal(b)
+        write_proposal(self.queue_dir, b)
-        write_response(self.slug, Response(
+        write_response(self.queue_dir, Response(
            proposal_id=a.id, status=STATUS_APPROVED, notes="",
        ))
-        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):
+    def test_list_pending_returns_empty_for_missing_dir(self):
-        self.assertEqual([], list_pending_proposals("nope"))
+        self.assertEqual([], list_pending_proposals(self.queue_dir / "nope"))
    def test_list_pending_sorted_by_arrival(self):
        # 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(self.queue_dir, b)
-        write_proposal(a)
+        write_proposal(self.queue_dir, a)
-        ordered = list_pending_proposals(self.slug)
+        ordered = list_pending_proposals(self.queue_dir)
        self.assertEqual([a.id, b.id], [p.id for p in ordered])
    def test_write_and_read_response(self):
        r = Response(proposal_id="xyz", status=STATUS_REJECTED, notes="no")
-        write_response(self.slug, r)
+        write_response(self.queue_dir, r)
-        self.assertEqual(r, read_response(self.slug, "xyz"))
+        self.assertEqual(r, read_response(self.queue_dir, "xyz"))
    def test_wait_for_response_returns_when_file_appears(self):
        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_proposal(self.queue_dir, p)
-        write_response(self.slug, Response(
+        write_response(self.queue_dir, Response(
            proposal_id=p.id, status=STATUS_APPROVED, notes="",
        ))
-        archive_proposal(self.slug, p.id)
+        archive_proposal(self.queue_dir, p.id)
-        self.assertEqual([], list_pending_proposals(self.slug))
+        self.assertFalse((self.queue_dir / f"{p.id}.proposal.json").exists())
-        with self.assertRaises(FileNotFoundError):
+        self.assertFalse((self.queue_dir / f"{p.id}.response.json").exists())
-            read_response(self.slug, p.id)
+        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.proposal.json").exists())
        self.assertTrue((self.queue_dir / "processed" / f"{p.id}.response.json").exists())
    def test_archive_is_idempotent_on_missing_files(self):
        # 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")
+        path = audit_log_path("egress", "dev")
-        self.assertEqual(3, len(entries))
+        with path.open() as f:
-        self.assertEqual(
+            lines = [line for line in f if line.strip()]
-            ["2026-05-25T12:00:00+00:00", "2026-05-25T12:00:01+00:00",
+        self.assertEqual(3, len(lines))
-             "2026-05-25T12:00:02+00:00"],
+        for line in lines:
-            [entry.timestamp for entry in entries],
+            self.assertTrue(json.loads(line))  # each line is valid JSON
        )
    def test_separate_logs_per_component_slug(self):
        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)
+        qdir = supervise.queue_dir_for_slug("dev")
-        supervise.write_response("dev", supervise.Response(
+        qdir.mkdir(parents=True)
        supervise.write_proposal(qdir, p)
        supervise.write_response(qdir, supervise.Response(
            proposal_id=p.id, status=STATUS_APPROVED, notes="",
        ))
        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)
+        qdir = supervise.queue_dir_for_slug("dev")
-        return supervise_cli.QueuedProposal(proposal=p)
+        qdir.mkdir(parents=True, exist_ok=True)
        supervise.write_proposal(qdir, p)
        return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
    def test_approve_writes_response(self):
        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.
+        # Gate polls the queue dir for the response; TUI must not archive it.
-        resp = read_response(qp.proposal.bottle_slug, qp.proposal.id)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_APPROVED, resp.status)
        self.assertEqual("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}), \
+            (Path(d) / "p.proposal.json").write_text("[]")
-                    self.assertRaises(FileNotFoundError):
+            with self.assertRaises(ValueError):
-                read_proposal("slug", "p")
+                read_proposal(Path(d), "p")
-    def test_read_response_missing_row(self) -> None:
+    def test_read_response_non_dict(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
+            (Path(d) / "p.response.json").write_text("[]")
-                    self.assertRaises(FileNotFoundError):
+            with self.assertRaises(ValueError):
-                read_response("slug", "p")
+                read_response(Path(d), "p")
-    def test_list_pending_reads_db_only(self) -> None:
+    def test_list_pending_skips_malformed(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}):
+            qd = Path(d)
-                supervise.write_proposal(_proposal())
+            (qd / "bad.proposal.json").write_text("{ not json")
-                pending = list_pending_proposals("slug")
+            (qd / "arr.proposal.json").write_text("[]")
            (qd / "incomplete.proposal.json").write_text("{}")  # from_dict raises
            supervise.write_proposal(qd, _proposal())  # one valid
            pending = list_pending_proposals(qd)
            self.assertEqual(1, len(pending))
            self.assertEqual("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_proposal(qd, p)
-                supervise.write_response("slug", supervise.Response(
+            (qd / f"{p.id}.response.json").write_text("{}")  # response exists -> skipped
-                    proposal_id=p.id,
+            self.assertEqual([], list_pending_proposals(qd))
                    status=STATUS_APPROVED,
                    notes="",
                ))
                self.assertEqual([], list_pending_proposals("slug"))
 class TestWaitForResponse(unittest.TestCase):
-    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}), \
+            (Path(d) / "p.response.json").write_text("{ not json")
-                    self.assertRaises(TimeoutError):
+            with self.assertRaises(TimeoutError):
-                wait_for_response("slug", "p", deadline=time.monotonic())
+                wait_for_response(Path(d), "p", deadline=time.monotonic())
-    def test_empty_db_response_does_not_count(self) -> None:
+    def test_incomplete_response_then_timeout(self) -> None:
        with tempfile.TemporaryDirectory() as d:
-            with patch.dict("os.environ", {"HOME": d}), \
+            (Path(d) / "p.response.json").write_text("{}")  # dict but from_dict raises
-                    self.assertRaises(TimeoutError):
+            with self.assertRaises(TimeoutError):
-                wait_for_response("slug", "p", deadline=time.monotonic())
+                wait_for_response(Path(d), "p", deadline=time.monotonic())
 class TestReadAuditEntries(unittest.TestCase):
@@ -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:
+    def test_skips_malformed_lines(self) -> None:
        with tempfile.TemporaryDirectory() as home, \
                patch.dict("os.environ", {"HOME": home}):
            write_audit_entry(AuditEntry(
                timestamp="t",
                bottle_slug="slug",
                component="egress",
                operator_action="approve",
                operator_notes="",
                justification="",
                diff="",
            ))
            write_audit_entry(AuditEntry(
                timestamp="t",
                bottle_slug="other",
                component="egress",
                operator_action="reject",
                operator_notes="",
                justification="",
                diff="",
            ))
            entries = read_audit_entries("egress", "slug")
            self.assertEqual(1, len(entries))
            self.assertEqual("approve", entries[0].operator_action)
    def test_legacy_audit_log_file_does_not_count(self) -> None:
        with tempfile.TemporaryDirectory() as home, \
                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):
+class TestFlockFallback(unittest.TestCase):
-    """Direct QueueStore / AuditStore construction and early-return guard branches."""
+    def test_flock_on_closed_fd_is_swallowed(self) -> None:
-
+        # flock on a closed fd raises OSError(EBADF), which the helpers swallow.
-    def test_queue_store_explicit_db_path(self):
+        fd = os.open(os.devnull, os.O_RDONLY)
-        with tempfile.TemporaryDirectory() as d:
+        os.close(fd)
-            db = Path(d) / "q.db"
+        supervise._try_flock(fd)
-            store = QueueStore("key", db_path=db)
+        supervise._try_funlock(fd)
            self.assertTrue(db.is_file())
            self.assertEqual(db, store.db_path)
    def test_queue_store_missing_db_list_pending_returns_empty(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "q.db"
            store = QueueStore("key", db_path=db)
            db.unlink()
            self.assertEqual([], store.list_pending_proposals())
    def test_queue_store_missing_db_list_all_returns_empty(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "q.db"
            store = QueueStore("key", db_path=db)
            db.unlink()
            self.assertEqual([], store.list_all_pending_proposals())
    def test_queue_store_missing_db_archive_is_noop(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "q.db"
            store = QueueStore("key", db_path=db)
            db.unlink()
            store.archive_proposal("anything")  # must not raise
    def test_queue_store_chmod_oserror_is_swallowed(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "q.db"
            with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
                QueueStore("key", db_path=db)  # must not raise
    def test_audit_store_missing_db_read_returns_empty(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "a.db"
            store = AuditStore(db_path=db)
            db.unlink()
            self.assertEqual([], store.read_audit_entries("egress", "slug"))
    def test_audit_store_chmod_oserror_is_swallowed(self):
        with tempfile.TemporaryDirectory() as d:
            db = Path(d) / "a.db"
            with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
                AuditStore(db_path=db)  # must not raise
 if __name__ == "__main__":
@@ -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")), \
+        with self.assertRaises(_RpcInternalError) as cm:
                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.queue_dir = Path(self._tmp.name)
-        self.config = ServerConfig(bottle_slug="dev")
+        self.config = ServerConfig(bottle_slug="dev", queue_dir=self.queue_dir)
    def tearDown(self):
        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,
        )