docs(prd): add PRD for egress control plane

Out-of-band egress enforcement & cost-control plane: meter token usage at the egress proxy, evaluate budgets with agent→bottle→parent→global precedence, and force cutoff/freeze/kill without the agent in the loop. Introduces a host-level SQLite ledger behind a thin repository API and a host-only TUI dashboard. Closes the design discussion on #251. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NkwFXLFff9PYPy4wgVBJp9
2026-06-29 11:01:47 -04:00
13 changed files with 265 additions and 431 deletions
@@ -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.
@@ -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 (e.g. Paseo),
 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 (e.g. Paseo), 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 like Paseo 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}); "
@@ -217,7 +217,7 @@ class ClaudeAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -227,13 +227,9 @@ class ClaudeAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            # manifest load, but quote the path so a future unvalidated
            # field can't inject shell metacharacters here either.
            dst_q = shlex.quote(dst)
            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")
    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        """Copy the prompt file into the guest, fix ownership/mode.
@@ -313,9 +309,6 @@ class ClaudeAgentProvider(AgentProvider):
                f"claude mcp add --scope user --transport http supervise {supervise_url}"
            )
    def headless_prompt(self, prompt: str) -> list[str]:
        return ["-p", prompt]
 def _exec(bottle: "Bottle", script: str, error: str) -> None:
    result = bottle.exec(script, user="root")
@@ -183,7 +183,7 @@ class CodexAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -193,13 +193,9 @@ class CodexAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            # manifest load, but quote the path so a future unvalidated
            # field can't inject shell metacharacters here either.
            dst_q = shlex.quote(dst)
            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")
    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        """Copy the prompt file into the guest, fix ownership/mode.
@@ -279,9 +275,6 @@ class CodexAgentProvider(AgentProvider):
                f"codex mcp add supervise --url {shlex.quote(supervise_url)}"
            )
    def headless_prompt(self, prompt: str) -> list[str]:
        return [prompt]
 def _exec(bottle: "Bottle", script: str, error: str) -> None:
    result = bottle.exec(script, user="root")
@@ -238,7 +238,7 @@ class PiAgentProvider(AgentProvider):
        if not agent.skills:
            return
        skills_dir = _skills_dir(plan.guest_home)
-        bottle.exec(f"mkdir -p {shlex.quote(skills_dir)}", user="root")
+        bottle.exec(f"mkdir -p {skills_dir}", user="root")
        for name in agent.skills:
            src = host_skill_dir(name)
            if not os.path.isdir(src):
@@ -248,13 +248,9 @@ class PiAgentProvider(AgentProvider):
                )
            dst = f"{skills_dir}/{name}"
            info(f"copying skill {name} into {bottle.name}:{dst}")
-            # Defense in depth: skill names are validated kebab-case at
+            bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
            # manifest load, but quote the path so a future unvalidated
            # field can't inject shell metacharacters here either.
            dst_q = shlex.quote(dst)
            bottle.exec(f"rm -rf {dst_q} && mkdir -p {dst_q}", user="root")
            bottle.cp_in(f"{src}/.", f"{dst}/")
-            bottle.exec(f"chown -R node:node {dst_q}", user="root")
+            bottle.exec(f"chown -R node:node {dst}", user="root")
    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
        prompt_path = _prompt_path(plan.guest_home)
@@ -315,9 +311,6 @@ class PiAgentProvider(AgentProvider):
    ) -> None:
        del plan, bottle, supervise_url
    def headless_prompt(self, prompt: str) -> list[str]:
        return ["-p", prompt]
 def _exec(bottle: "Bottle", script: str, error: str) -> None:
    result = bottle.exec(script, user="root")
@@ -8,7 +8,7 @@ from typing import cast
 from .agent_provider import PROVIDER_TEMPLATES
 from .manifest_util import ManifestError, as_json_object
 from .manifest_git import ManifestGitUser
-from .manifest_schema import AGENT_MODEL_KEYS, is_valid_entity_name
+from .manifest_schema import AGENT_MODEL_KEYS
@dataclass(frozen=True)
@@ -161,16 +161,6 @@ class ManifestAgent:
                        f"agent '{name}' skills[{i}] must be a string "
                        f"(was {type(skill).__name__})"
                    )
                # Skill names become host/guest path segments and are
                # interpolated into provisioning shell commands, so they
                # must fit the same kebab-case convention as bottle/agent
                # filenames — rejecting anything that could break out of a
                # path segment or inject shell metacharacters.
                if not is_valid_entity_name(skill):
                    raise ManifestError(
                        f"agent '{name}' skills[{i}] {skill!r} is not a valid "
                        f"skill name; must match [a-z][a-z0-9-]*"
                    )
                collected.append(skill)
            skills = tuple(collected)
@@ -33,20 +33,13 @@ AGENT_KEYS = (
 AGENT_MODEL_KEYS = AGENT_KEYS | frozenset({"prompt"})
 def is_valid_entity_name(name: str) -> bool:
    """True if `name` fits the kebab-case `[a-z][a-z0-9-]*` convention
    shared by bottle/agent filenames and skill names. Names that satisfy
    this are also safe to interpolate into a host/guest path segment."""
    return bool(_FILENAME_RX.match(name))
 def entity_name_from_path(path: Path) -> str | None:
    """Return the entity name implied by the filename, or None if the
    filename does not fit the [a-z][a-z0-9-]* convention."""
    if path.suffix != ".md":
        return None
    stem = path.stem
-    if not is_valid_entity_name(stem):
+    if not _FILENAME_RX.match(stem):
        return None
    return stem
@@ -0,0 +1,247 @@
 # PRD prd-new: Egress control plane — metering, budgets, and forced cutoff
 - **Status:** Draft
 - **Author:** didericis
 - **Created:** 2026-06-25
 - **Issue:** #251
 ## Summary
 Add an **out-of-band egress enforcement & observability plane**: meter every
 agent's token usage at the egress proxy, decrement budgets without the agent's
 cooperation, and forcibly cut a bottle's egress when a budget is exhausted —
 either automatically or on command from a host-level dashboard. The trigger
 (usage threshold) and the action (route-drop / freeze / kill) both live in the
 egress plane and run with no agent in the loop. This is distinct from the
 supervise sidecar (PRD 0013), which is agent-initiated and therefore cannot
 enforce a cost cutoff on a runaway agent. State (usage ledger, budgets, audit)
 moves into a host-level SQLite database behind a thin repository API, the first
 SQL store in an otherwise flat-file repo.
 ## Problem
 bot-bottle can't currently do two things the cost-overrun case demands:
 1. **Forced egress shutdown on limit.** When an agent crosses a token
   threshold, kill its egress automatically — no human in the loop.
 2. **Remote (host-level) management.** Drive agents from a single surface:
   see usage, cut egress, stop bottles, to prevent cost overruns.
 The existing supervise sidecar (PRD 0013) is **entirely agent-initiated**: every
 action begins with the agent voluntarily calling an MCP tool and an operator
 approving it. A runaway or expensive agent — exactly the cost-overrun case —
 will never call `egress-block` on itself. Supervision is therefore a
 **collaborative recovery** mechanism, not an **enforcement** mechanism; making
 it mandatory (#249) would not deliver forced cost-cutoff.
 The requirement forces a distinction the current design blurs:
 - **Plane A — enforcement / observability (this PRD).** System → infrastructure.
  Meter usage, cut egress on threshold or command, account for cost.
  Out-of-band; independent of the agent. **Unconditional** — an enforcement
  plane you can opt out of isn't enforcement.
 - **Plane B — agent-facing recovery (the existing supervise sidecar).**
  Agent → operator, approval-gated. Useful interactively; meaningless for a
  headless agent with no operator watching its queue. Remains optional.
 This PRD builds Plane A. It reframes the "always-on control" invariant of #249
 as "the egress control plane is always present" — a more defensible property
 than "every agent runs the agent-facing supervisor." Unsupervised
 (headless/CI/ephemeral) agents stay first-class: still subject to the mandatory
 meter + kill switch, they simply lack the agent-facing proposal tools they
 couldn't use anyway.
 ## Goals / Success Criteria
 - The egress proxy meters every request to a metered API host (e.g.
  `api.anthropic.com`) and records authoritative token usage per bottle and per
  agent provider, with no agent cooperation.
 - A budget can be set at four scopes with deterministic precedence
  (**agent → bottle → parent bottle → global host budget**); the
  most-specific applicable budget governs.
 - When usage crosses a budget, the bottle's configured **cutoff policy**
  (`cutoff` | `freeze` | `kill`) fires automatically, executed host-side on the
  egress plane — never via the supervise queue.
 - An operator can, from a single **host-level TUI dashboard**, see live per-bottle
  usage against budget and command a cutoff/stop on demand.
 - Host budgets, default cutoff policy, and per-provider limits are declared in a
  new host-level `~/.bot-bottle/settings.yml`, parseable by `yaml_subset.py`.
 - All usage, budget state, and enforcement actions persist in a host-level
  SQLite DB behind a thin repository API, so the store can later be swapped for
  a cross-host cloud service.
 ## Non-goals
 - **Remote control / cross-host control plane.** Web + mobile remote control,
  cross-host budgets, and the authn/transport they require are explicitly
  deferred. v1 is a **host-only TUI** with no remote surface.
 - **Dollar-denominated budgets.** Budgets are token counts keyed by agent
  provider, not currency. Price tables are out of scope.
 - **Migrating existing flat-file state into SQLite.** Resume `metadata.json`,
  transcripts, Dockerfile overrides, the supervise queue, and audit logs stay on
  the filesystem. Only the *new* metering/budget/enforcement ledger is SQL.
 - **Making the supervise sidecar (Plane B) mandatory.** Out of scope here; this
  PRD is the answer to "what should be unconditional" (Plane A), leaving #249's
  Plane-B question open.
 - **Per-request hard pre-send blocking as the primary mechanism.** The gate is
  budget-crossing detected at/after metering; a pre-flight estimator (below) is a
  refinement, not the core enforcement path.
 ## Design
 ### Two measurements: gate vs. account
 There are two distinct needs, and they want different signals:
 - **Account (authoritative).** Decrement the real budget from the API
  **response**, which already carries authoritative usage (Anthropic
  `input_tokens` / `output_tokens`, OpenAI `usage`). The egress addon already
  has a `response(flow)` hook (`bot_bottle/egress_addon.py:460`), so the real
  number is available with no extra network call. **Caveat:** agent traffic is
  mostly streaming SSE, so the response path must tail the stream for the final
  usage event rather than parse a single JSON body — scoped explicitly as work.
 - **Gate (estimate).** To block *before* sending, only the request is available,
  so an estimator / provider `count_tokens` endpoint is the only option.
 Calling `count_tokens` for accounting would be both less accurate *and* an extra
 metered egress call per request, so accounting uses response `usage` and the
 estimator is reserved for the optional pre-flight gate.
 ### `count_tokens` on agent providers
 Add an abstract `count_tokens(request) -> int` to the `AgentProvider`
 abstraction (`bot_bottle/agent_provider.py`):
 - **Default** is a good-enough stdlib estimator. Prefer stdlib only; a small
  pip dependency *for the sidecar* is acceptable for the fallback if stdlib
  proves too inaccurate (this does not relax the package's stdlib-first stance —
  it would be a sidecar-only dep, like the bundle already carries).
 - **Built-in `claude`** uses Anthropic's token-counting endpoint;
  **built-in `codex`** uses OpenAI's. These are exact for the gate but cost a
  metered call, so they are gate-only; accounting still comes from the response.
 ### Budgets and precedence
 Budgets are token counts keyed by **agent provider name** (the same names
 bottles already use). Four scopes, most-specific wins:
 ```
 agent  →  bottle  →  parent bottle  →  global (host)
 ```
 The global host budget is the highest-priority feature to ship (the cross-host
 control plane will eventually consume it); per-agent and per-bottle budgets
 override it for finer control. A budget can also be supplied **at bottle
 launch** (`--budget` or equivalent), overriding the settings.yml defaults for
 that run. Enforcement evaluates the effective budget as the
 nearest-defined scope at decrement time.
 ### `~/.bot-bottle/settings.yml`
 New **host-level** settings file (the `~/.bot-bottle/` root, *not* the per-repo
 `.bot-bottle/` — host budgets must not be committed per-repo). Parsed by
 `yaml_subset.py`, so it must stay within that bounded subset (flat mappings,
 scalars; no anchors, no multi-line block scalars). Shape:
 ```yaml
 budget:
  claude: 5000000      # token budget keyed by agent provider
  codex: 2000000
 shutdown: cutoff       # default cutoff policy: cutoff | freeze | kill
 ```
 ### Forced cutoff and cutoff policy
 On budget exhaustion (or an operator command), the configured per-bottle cutoff
 policy fires. The three policies map onto primitives that already exist:
 - **`cutoff`** (default) — drop the bottle's `routes.yaml` to empty and reload
  (or isolate the bottle from the egress network); the agent/bottle keeps
  running but can no longer reach metered hosts. This is the route-drop already
  available on the egress plane (`bot_bottle/backend/egress_apply.py`).
 - **`freeze`** — commit/snapshot state, then kill the agent/bottle; resumable
  later via `bot_bottle/backend/freeze.py`.
 - **`kill`** — tear the bottle down without saving state (backend teardown).
 The trigger lives in the metering path and the action in the egress/backend
 plane; **neither touches the supervise proposal queue** (design constraint from
 #251).
 ### Host-level SQLite store
 **Decision: introduce SQLite now, narrowly.**
 - **The dependency objection doesn't apply.** `sqlite3` is in the Python stdlib,
  so it does not break the AGENTS.md stdlib-first / no-runtime-pip stance — same
  category as the hand-rolled `yaml_subset.py`, except the stdlib already ships
  the whole engine.
 - **It fits the problem.** A *global* token budget decremented concurrently by N
  egress sidecars (today `~/.bot-bottle/` already has `state/`, `audit/`,
  `queue/` written by parallel bottles) is a read-modify-write race. Over JSON
  that means hand-rolled file locking; SQLite gives atomic transactions + WAL for
  free. The per-agent/per-bottle precedence rollup plus "sum across all bottles"
  is a `GROUP BY`, not an N-directory rescan.
 - **It rehearses the cloud swap.** "Wrap operations in an API so we can swap to a
  cloud service" maps directly onto a thin repository/DAO over SQLite → Postgres
  later. A JSON-file store is a worse rehearsal than SQL.
 **Costs (real but bounded):** a new paradigm in a flat-file repo needs a
 `schema_version` table + idempotent startup migrations; SQLite serializes
 writers, so WAL mode + `busy_timeout` are required (a non-issue at a handful of
 bottles); test fixtures need temp DBs.
 **Scope of the store:** one DB at `~/.bot-bottle/bot-bottle.db` behind a thin
 repository API. Only the **new** metering/budget/enforcement-audit ledger lives
 there. Existing per-bottle blobs (resume `metadata.json`, transcripts,
 Dockerfile overrides, supervise queue) stay on the filesystem — migrating them
 now is churn for no benefit and they lack the concurrency/aggregation problem.
 ### Host-level controller + dashboard
 A single **host-level controller** owns the meter, budget evaluation, and the
 cutoff actions across all bottles (cf. `bot_bottle/cli/supervise.py`'s
 cross-bottle view), rather than a per-bottle daemon. v1 ships one host-level
 **TUI dashboard** that reads live usage-vs-budget from the SQLite store and
 offers on-demand cutoff/stop. The existing supervisor UI should eventually fold
 into this same dashboard; this PRD lays the host-level surface it will move to.
 ## Implementation chunks
 Ordered, individually mergeable:
 1. **SQLite repository foundation.** `~/.bot-bottle/bot-bottle.db`, schema +
   `schema_version` migrations, WAL + `busy_timeout`, thin repository API,
   temp-DB test fixtures. No behavior wired yet.
 2. **Metering at the egress proxy.** Parse authoritative response `usage`
   (including SSE final-usage tailing) in the egress addon `response` hook;
   write per-bottle / per-provider usage rows to the ledger.
 3. **`settings.yml` + budget model.** Host-level `~/.bot-bottle/settings.yml`
   parsed by `yaml_subset.py`; budget precedence (agent → bottle → parent →
   global) and the `--budget` launch flag.
 4. **Forced cutoff + cutoff policy.** Wire the threshold trigger to the
   `cutoff` / `freeze` / `kill` primitives on the egress/backend plane; record
   enforcement actions to the audit ledger.
 5. **Host-level TUI dashboard.** Live usage-vs-budget view + on-demand
   cutoff/stop, reading the store.
 6. **`count_tokens` pre-flight gate (optional refinement).** Abstract method +
   stdlib estimator default; Anthropic/OpenAI endpoints for built-in
   claude/codex; optional pre-send block.
 ## Open questions
 - **SSE usage tailing robustness.** Buffering streamed responses to extract the
  final usage event without breaking the agent's own stream consumption — how
  much of the body must the addon hold, and what's the failure mode if the
  stream is interrupted mid-flight?
 - **Crossing mid-request.** A single response can push usage past budget only
  *after* it's already been delivered. Is post-hoc cutoff (next request blocked)
  sufficient, or is a pre-flight estimator gate (chunk 6) required for v1?
 - **Provider name ↔ metered host mapping.** How does the proxy attribute a
  flow to an agent-provider budget key — by destination host, by bottle
  identity, or both?
 - **Parent-bottle budget semantics.** For `bottle extends` (PRD 0025 / 0065)
  chains, does "parent bottle" mean the manifest parent, the launching bottle,
  or the full ancestry summed?
 - **Dashboard ↔ controller transport (even host-only).** In-process, a local
  socket, or polling the SQLite store directly? Picks the seam the future remote
  control plane will extend.
@@ -1,188 +0,0 @@
 """Unit: `cli.py start --headless` non-interactive launch path.
 Headless is the keystone for orchestrators (Paseo), 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()
@@ -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()
@@ -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()
@@ -165,22 +165,6 @@ class TestAgentValidation(unittest.TestCase):
        with self.assertRaises(ManifestError):
            ManifestAgent.from_dict("a", {"skills": [5]}, set())
    def test_skill_name_rejects_shell_metacharacters(self) -> None:
        # Skill names become host/guest path segments interpolated into
        # provisioning shell commands; anything outside kebab-case is
        # rejected at load so it can never reach a `bottle.exec` string.
        for bad in ("foo; rm -rf /", "../escape", "foo bar", "Foo", "-leading"):
            with self.assertRaises(ManifestError):
                ManifestAgent.from_dict("a", {"skills": [bad]}, set())
    def test_skill_name_accepts_kebab_case(self) -> None:
        agent = ManifestAgent.from_dict(
            "a", {"skills": ["init-entry", "quality-eval", "skill0"]}, set()
        )
        self.assertEqual(
            agent.skills, ("init-entry", "quality-eval", "skill0")
        )
    def test_prompt_not_string(self) -> None:
        with self.assertRaises(ManifestError):
            ManifestAgent.from_dict("a", {"prompt": 5}, set())