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

Dockerfile.sidecars

@@ -62,7 +62,6 @@ COPY --from=gitleaks-src /usr/bin/gitleaks /usr/bin/gitleaks
 # top-level siblings (absolute imports), matching the prior
 # Dockerfile.egress / Dockerfile.supervise layout.
 COPY bot_bottle/egress_addon_core.py /app/egress_addon_core.py
-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

bot_bottle/egress_addon_core.py

@@ -21,32 +21,6 @@ try:
 except ImportError:  # pragma: no cover - host-side path
     from .yaml_subset import YamlSubsetError, parse_yaml_subset
 
-# DLP detector-config parsing lives in a sibling module (also flat-bundled
-# into the sidecar — see Dockerfile.sidecars). Re-exported below so existing
-# `from egress_addon_core import ON_MATCH_*` callers keep working.
-try:
-    from egress_dlp_config import (  # type: ignore[import-not-found]
-        DEFAULT_OUTBOUND_ON_MATCH,
-        INBOUND_DETECTOR_NAMES,
-        ON_MATCH_BLOCK,
-        ON_MATCH_REDACT,
-        ON_MATCH_SUPERVISE,
-        OUTBOUND_DETECTOR_NAMES,
-        OUTBOUND_ON_MATCH_VALUES,
-        parse_dlp_block,
-    )
-except ImportError:  # pragma: no cover - host-side path
-    from .egress_dlp_config import (
-        DEFAULT_OUTBOUND_ON_MATCH,
-        INBOUND_DETECTOR_NAMES,
-        ON_MATCH_BLOCK,
-        ON_MATCH_REDACT,
-        ON_MATCH_SUPERVISE,
-        OUTBOUND_DETECTOR_NAMES,
-        OUTBOUND_ON_MATCH_VALUES,
-        parse_dlp_block,
-    )
-
 
 # ---------------------------------------------------------------------------
 # Match types (Gateway API HTTPRoute vocabulary, PRD 0053)
@@ -60,6 +34,18 @@ VALID_METHODS = frozenset({
     "CONNECT",
 })
 
+OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets", "entropy"})
+INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})
+
+# Per-route policy for what the proxy does when an outbound DLP detector
+# matches a token (PRD 0062).
+ON_MATCH_BLOCK = "block"          # hard 403, never overridable
+ON_MATCH_REDACT = "redact"        # scrub the matched value, forward the request
+ON_MATCH_SUPERVISE = "supervise"  # queue for operator approval, hold the request
+OUTBOUND_ON_MATCH_VALUES = (ON_MATCH_BLOCK, ON_MATCH_REDACT, ON_MATCH_SUPERVISE)
+# Unset resolves to supervise (fall back to block when supervise is not wired).
+DEFAULT_OUTBOUND_ON_MATCH = ON_MATCH_SUPERVISE
+
 
 @dataclass(frozen=True)
 class PathMatch:
@@ -244,6 +230,72 @@ def _parse_match_entry(idx: int, k: int, raw: object) -> MatchEntry:
     return MatchEntry(paths=paths, methods=methods, headers=headers)
 
 
+def _parse_detectors(
+    idx: int,
+    host: str,
+    raw_dict: dict[str, object],
+) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
+    """Parse the optional `dlp` block on a route, returning
+    (outbound_detectors, inbound_detectors, outbound_on_match)."""
+    dlp_raw = raw_dict.get("dlp")
+    if dlp_raw is None:
+        return None, None, ""
+    label = f"route[{idx}] ({host})"
+    if not isinstance(dlp_raw, dict):
+        raise ValueError(f"{label}: 'dlp' must be an object")
+    dlp = typing.cast(dict[str, object], dlp_raw)
+
+    def _parse_detector_field(
+        field: str,
+        valid_names: frozenset[str],
+    ) -> tuple[str, ...] | None:
+        val = dlp.get(field)
+        if val is None:
+            return None
+        if val is False:
+            return ()
+        if not isinstance(val, list):
+            raise ValueError(
+                f"{label}: dlp.{field} must be false, a list, or omitted"
+            )
+        items = typing.cast(list[object], val)
+        names: list[str] = []
+        for j, item in enumerate(items):
+            if not isinstance(item, str):
+                raise ValueError(
+                    f"{label}: dlp.{field}[{j}] must be a string"
+                )
+            if item not in valid_names:
+                raise ValueError(
+                    f"{label}: dlp.{field}[{j}] {item!r} is not a valid "
+                    f"detector name; valid names: {', '.join(sorted(valid_names))}"
+                )
+            names.append(item)
+        return tuple(names)
+
+    outbound = _parse_detector_field("outbound_detectors", OUTBOUND_DETECTOR_NAMES)
+    inbound = _parse_detector_field("inbound_detectors", INBOUND_DETECTOR_NAMES)
+
+    on_match = ""
+    on_match_raw = dlp.get("outbound_on_match")
+    if on_match_raw is not None:
+        if not isinstance(on_match_raw, str) or on_match_raw not in OUTBOUND_ON_MATCH_VALUES:
+            raise ValueError(
+                f"{label}: dlp.outbound_on_match must be one of "
+                f"{', '.join(OUTBOUND_ON_MATCH_VALUES)} (got {on_match_raw!r})"
+            )
+        on_match = on_match_raw
+
+    for k in dlp:
+        if k not in ("outbound_detectors", "inbound_detectors", "outbound_on_match"):
+            raise ValueError(
+                f"{label}: dlp has unknown key {k!r}; accepted keys "
+                f"are 'outbound_detectors', 'inbound_detectors', "
+                f"'outbound_on_match'"
+            )
+    return outbound, inbound, on_match
+
+
 def parse_routes(payload: object) -> tuple[Route, ...]:
     if not isinstance(payload, dict):
         raise ValueError("routes payload: top-level must be an object")
@@ -312,7 +364,7 @@ def _parse_one(idx: int, raw: object) -> Route:
                 )
 
     # dlp detectors
-    outbound_detectors, inbound_detectors, outbound_on_match = parse_dlp_block(
+    outbound_detectors, inbound_detectors, outbound_on_match = _parse_detectors(
         idx, host, raw_dict,
     )
 
@@ -785,9 +837,6 @@ __all__ = [
     "ON_MATCH_SUPERVISE",
     "OUTBOUND_ON_MATCH_VALUES",
     "DEFAULT_OUTBOUND_ON_MATCH",
-    "OUTBOUND_DETECTOR_NAMES",
-    "INBOUND_DETECTOR_NAMES",
-    "parse_dlp_block",
     "Config",
     "Decision",
     "HeaderMatch",

bot_bottle/egress_dlp_config.py

@@ -1,92 +0,0 @@
-"""DLP detector-config parsing for egress routes (PRD 0053, PRD 0062).
-
-A route's optional `dlp:` block names which outbound/inbound detectors run
-and what the proxy does when an outbound detector matches a token
-(`outbound_on_match`). This module owns parsing and validating that block,
-kept apart from the request-time scan/decision flow in `egress_addon_core`
-so each half reads top-to-bottom without scrolling past the other.
-
-Stdlib-only; ships flat into the sidecar bundle image alongside
-`egress_addon_core.py` — see `Dockerfile.sidecars`."""
-
-from __future__ import annotations
-
-import typing
-
-OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets", "entropy"})
-INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})
-
-# Per-route policy for what the proxy does when an outbound DLP detector
-# matches a token (PRD 0062).
-ON_MATCH_BLOCK = "block"          # hard 403, never overridable
-ON_MATCH_REDACT = "redact"        # scrub the matched value, forward the request
-ON_MATCH_SUPERVISE = "supervise"  # queue for operator approval, hold the request
-OUTBOUND_ON_MATCH_VALUES = (ON_MATCH_BLOCK, ON_MATCH_REDACT, ON_MATCH_SUPERVISE)
-# Unset resolves to supervise (fall back to block when supervise is not wired).
-DEFAULT_OUTBOUND_ON_MATCH = ON_MATCH_SUPERVISE
-
-
-def parse_dlp_block(
-    idx: int,
-    host: str,
-    raw_dict: dict[str, object],
-) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
-    """Parse the optional `dlp` block on a route, returning
-    (outbound_detectors, inbound_detectors, outbound_on_match)."""
-    dlp_raw = raw_dict.get("dlp")
-    if dlp_raw is None:
-        return None, None, ""
-    label = f"route[{idx}] ({host})"
-    if not isinstance(dlp_raw, dict):
-        raise ValueError(f"{label}: 'dlp' must be an object")
-    dlp = typing.cast(dict[str, object], dlp_raw)
-
-    def _parse_detector_field(
-        field: str,
-        valid_names: frozenset[str],
-    ) -> tuple[str, ...] | None:
-        val = dlp.get(field)
-        if val is None:
-            return None
-        if val is False:
-            return ()
-        if not isinstance(val, list):
-            raise ValueError(
-                f"{label}: dlp.{field} must be false, a list, or omitted"
-            )
-        items = typing.cast(list[object], val)
-        names: list[str] = []
-        for j, item in enumerate(items):
-            if not isinstance(item, str):
-                raise ValueError(
-                    f"{label}: dlp.{field}[{j}] must be a string"
-                )
-            if item not in valid_names:
-                raise ValueError(
-                    f"{label}: dlp.{field}[{j}] {item!r} is not a valid "
-                    f"detector name; valid names: {', '.join(sorted(valid_names))}"
-                )
-            names.append(item)
-        return tuple(names)
-
-    outbound = _parse_detector_field("outbound_detectors", OUTBOUND_DETECTOR_NAMES)
-    inbound = _parse_detector_field("inbound_detectors", INBOUND_DETECTOR_NAMES)
-
-    on_match = ""
-    on_match_raw = dlp.get("outbound_on_match")
-    if on_match_raw is not None:
-        if not isinstance(on_match_raw, str) or on_match_raw not in OUTBOUND_ON_MATCH_VALUES:
-            raise ValueError(
-                f"{label}: dlp.outbound_on_match must be one of "
-                f"{', '.join(OUTBOUND_ON_MATCH_VALUES)} (got {on_match_raw!r})"
-            )
-        on_match = on_match_raw
-
-    for k in dlp:
-        if k not in ("outbound_detectors", "inbound_detectors", "outbound_on_match"):
-            raise ValueError(
-                f"{label}: dlp has unknown key {k!r}; accepted keys "
-                f"are 'outbound_detectors', 'inbound_detectors', "
-                f"'outbound_on_match'"
-            )
-    return outbound, inbound, on_match

docs/prds/prd-new-egress-control-plane.md

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